当 udf 函数不接受足够大的输入变量时 Spark DataFrames [英] Spark DataFrames when udf functions do not accept large enough input variables

问题描述

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

I am preparing a DataFrame with an id and a vector of my features to be used later for doing predictions. I do a groupBy on my dataframe, and in my groupBy I am merging couple of columns as lists into a new column:

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 函数最多接受的数字.

This is how I am creating my feature vectors and their labels. It has been working for me so far but this is the first time that my feature vector with this method is getting bigger than number 10 which is what at maximum a udf function in Spark accepts.

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

I am not sure how else I can fix this? Is the size of udf inputs in Spark going to get bigger, am have I understood them incorrectly, or there is a better way?

推荐答案

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

User defined functions are defined for up to 22 parameters. Only udf helper is define for at most 10 arguments. To handle functions with larger number of parameters you can use 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 对象:

You can also create an UserDefinedFunction object:

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

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

实际上，具有 22 个参数的函数并不是很有用，除非您想使用 Scala 反射来生成这些参数，否则会有维护噩梦.

In practice having a function with 22 arguments is not very useful and unless you want to use Scala reflection to generate these there are maintenance nightmare.

我会考虑使用集合(array、map)或 struct 作为输入或将其划分为多个模块.例如:

I would either consider using collections (array, map) or struct as an input or divide this into multiple modules. For example:

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

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

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

这篇关于当 udf 函数不接受足够大的输入变量时 Spark DataFrames的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！