在 Spark UDF 中操作数据帧

我有一个 UDF，可以从数据帧中过滤和选择值，但它遇到“对象不可序列化”错误。详细信息如下。

假设我有一个数据框 df1 ，其中包含名称为（“ID”，“Y1”，“Y2”，“Y3”，“Y4”，“Y5”，“Y6”，“Y7”，“Y8”，“Y9”的列”，“Y10”）。我想根据另一个数据帧 df2 中匹配的“ID”和“值”对“Y”列的子集进行求和。我尝试了以下方法：

val y_list = ("Y1", "Y2", "Y3", "Y4", "Y5", "Y6", "Y7", "Y8", "Y9", "Y10").map(c => col(c))

def udf_test(ID: String, value: Int): Double = {
  df1.filter($"ID" === ID).select(y_list:_*).first.toSeq.toList.take(value).foldLeft(0.0)(_+_)
}
sqlContext.udf.register("udf_test", udf_test _)

val df_result = df2.withColumn("Result", callUDF("udf_test", $"ID", $"Value"))

这给了我以下形式的错误：

java.io.NotSerializableException: org.apache.spark.sql.Column
Serialization stack:
- object not serializable (class: org.apache.spark.sql.Column, value: Y1)

我查了一下，发现 Spark Column 是不可序列化的。我想知道：

1) 有什么方法可以在 UDF 中操作数据帧吗？

2）如果不是，实现上述操作类型的最佳方法是什么？我的真实案例比这更复杂。它要求我根据大数据帧中的某些列从多个小数据帧中选择值，然后计算回大数据帧的值。

我使用的是 Spark 1.6.3。谢谢！

您不能在 UDF 内使用数据集操作。 UDF 只能操作现有列并生成一个结果列。它不能过滤数据集或进行聚合，但可以在过滤器内部使用。 UDAF 还可以聚合值。

相反，您可以使用.as[SomeCaseClass]从 DataFrame 生成数据集，并在过滤器、映射、归约中使用普通的强类型函数。

编辑：如果你想将你的bigDF与smallDFs列表中的每个小DF一起加入，你可以这样做：

import org.apache.spark.sql.functions._
val bigDF = // some processing
val smallDFs = Seq(someSmallDF1, someSmallDF2)
val joined = smallDFs.foldLeft(bigDF)((acc, df) => acc.join(broadcast(df), "join_column"))

broadcast是给小DF添加Broadcast Hint的功能，这样小DF会使用更高效的Broadcast Join而不是Sort Merge Join