当 udf 函数不接受足够大的输入变量时 Spark DataFrames答案

【问题标题】：Spark DataFrames when udf functions do not accept large enough input variables当 udf 函数不接受足够大的输入变量时 Spark DataFrames
【发布时间】：2017-01-22 12:09:30
【问题描述】：

我正在准备一个带有 id 和我的特征向量的 DataFrame，以便稍后用于进行预测。我在我的数据框上做了一个 groupBy，在我的 groupBy 中，我将几列作为列表合并到一个新列中：

def mergeFunction(...) // with 14 input variables

val myudffunction( mergeFunction ) // Spark doesn't support this

df.groupBy("id").agg(
   collect_list(df(...)) as ...
   ... // too many of these (something like 14 of them)
).withColumn("features_labels",
  myudffunction(
     col(...)
     , col(...) )
.select("id", "feature_labels")

这就是我创建特征向量及其标签的方式。到目前为止它一直对我有用，但这是我的特征向量第一次使用这种方法变得大于数字 10，这是 Spark 中的 udf 函数最多接受的。

我不确定我还能如何解决这个问题？是 udf 输入的大小 Spark 会变大，我是不是理解错了，或者有更好的方法吗？

【问题讨论】：

标签： scala apache-spark dataframe apache-spark-sql apache-spark-mllib

【解决方案1】：

用户定义的函数最多可定义 22 个参数。最多为 10 个参数定义 udf 助手。要处理具有大量参数的函数，您可以使用org.apache.spark.sql.UDFRegistration。

例如

val dummy = ((
  x0: Int, x1: Int, x2: Int, x3: Int, x4: Int, x5: Int, x6: Int, x7: Int, 
  x8: Int, x9: Int, x10: Int, x11: Int, x12: Int, x13: Int, x14: Int, 
  x15: Int, x16: Int, x17: Int, x18: Int, x19: Int, x20: Int, x21: Int) => 1)

范被注册：

import org.apache.spark.sql.expressions.UserDefinedFunction

val dummyUdf: UserDefinedFunction = spark.udf.register("dummy", dummy)

直接使用

val df = spark.range(1)
val exprs =  (0 to 21).map(_ => lit(1))

df.select(dummyUdf(exprs: _*))

或通过callUdf按名称命名

import org.apache.spark.sql.functions.callUDF

df.select(
  callUDF("dummy", exprs:  _*).alias("dummy")
)

或 SQL 表达式：

df.selectExpr(s"""dummy(${Seq.fill(22)(1).mkString(",")})""")

您也可以创建UserDefinedFunction 对象：

import org.apache.spark.sql.expressions.UserDefinedFunction

Seq(1).toDF.select(UserDefinedFunction(dummy, IntegerType, None)(exprs: _*))

在实践中，具有 22 个参数的函数并不是很有用，除非您想使用 Scala 反射来生成这些参数，否则维护方面的噩梦。

我会考虑使用集合（array、map）或struct 作为输入，或者将其分成多个模块。例如：

val aLongArray = array((0 to 256).map(_ => lit(1)): _*)

val udfWitharray = udf((xs: Seq[Int]) => 1)

Seq(1).toDF.select(udfWitharray(aLongArray).alias("dummy"))

【讨论】：

【解决方案2】：

只是为了扩展零的答案，可以让.withColumn() 函数与具有超过 10 个参数的 UDF 一起使用。只需要spark.udf.register() 函数，然后使用expr 作为添加列的参数（而不是udf）。

例如，这样的事情应该可以工作：

def mergeFunction(...) // with 14 input variables

spark.udf.register("mergeFunction", mergeFunction) // make available in expressions

df.groupBy("id").agg(
   collect_list(df(...)) as ...
   ... // too many of these (something like 14 of them)
).withColumn("features_labels",
  expr("mergeFunction(col1, col2, col3, col4, ...)") ) //pass in the 14 column names
.select("id", "feature_labels")

底层表达式解析器似乎可以处理超过 10 个参数，所以我认为您不必求助于传递数组来调用函数。另外，如果它们的参数碰巧是不同的数据类型，数组就不能很好地工作。

【讨论】：