Spark scala 数据框 udf 返回行

Question

假设我有一个数据帧，其中包含一个列(称为colA ),这是一个行序列。我想在可乐的每个记录中添加一个新字段。(而新的字段与之前的记录关联，所以我必须写一个udf。)这个udf应该怎么写？

我尝试编写一个udf，它将colA作为输入，并输出Seq[Row]，其中每个记录都包含新字段。但问题是udf无法返回Seq[Row]/异常是“org.apache.spark.sql类型的模式”。不支持行“”。我该怎么办？

我编写的udf：＜code＞val convert＝udf[Seq[Row]，Seq[Row]]（blablabla…

Answer 1

这是个老问题，我只是想根据新版Spark更新一下。

从Spark 3.0.0开始，@Raphael罗斯提到的方法被弃用。因此，您可能会得到一个分析异常。原因是使用此方法的输入闭包没有类型检查，并且在涉及空值时，行为可能与我们在SQL中预期的行为不同。

如果您真的知道自己在做什么，则需要设置＜code＞spark.sql.legacy。allowUntypedScalaUDF配置为true。

另一种解决方案是使用＜code＞case class

case class Foo(field1: String, field2: String)

val convertFunction: Seq[Row] => Seq[Foo] = input => {
    input.map {
        x => // do something with x and convert to Foo
    }
}

val myUdf = udf(convertFunction)

Answer 2

从 spark 2.0 开始，您可以创建返回 Row / Seq[Row] 的 UDF，但您必须提供返回类型的架构，例如，如果您使用双精度数组：

val schema = ArrayType(DoubleType)

val myUDF = udf((s: Seq[Row]) => {
  s // just pass data without modification
}, schema)

但我真的无法想象这在哪里有用，我宁愿从UDF返回元组或case类（或其Seq）。

编辑：如果您的行包含超过22个字段（元组/案例类的字段限制），则这可能很有用