Spark 生成自定义列函数，用户定义的函数

Question

我正在使用 Scala，并希望构建自己的数据帧函数。例如，我想将一列视为数组，循环访问每个元素并进行计算。

首先，我尝试实现自己的getMax方法。因此列x的值为[3,8,2,5,9]，该方法的预期输出为9。

下面是它在Scala中的样子

def getMax(inputArray: Array[Int]): Int = {
   var maxValue = inputArray(0)
   for (i <- 1 until inputArray.length if inputArray(i) > maxValue) {
     maxValue = inputArray(i)
   }
   maxValue
}

这是我目前所知道的，并得到这个错误

"value length is not a member of org.apache.spark.sql.column", 

我不知道如何迭代该列。

def getMax(col: Column): Column = {
var maxValue = col(0)
for (i <- 1 until col.length if col(i) > maxValue){
    maxValue = col(i)
}
maxValue

}

一旦我能够实现自己的方法，我将创建一个列函数

val value_max:org.apache.spark.sql.Column=getMax(df.col(“value”)).as(“value_max”)

然后我希望能够在SQL的陈述中使用它

val sample = sqlContext.sql("SELECT value_max(x) FROM table")

给定输入列[3，8，2，5，9]，预期输出将是9

我遵循另一个线程Spark Scala的答案-我如何迭代数据帧中的行，并将计算值添加为数据帧的新列，其中它们为标准偏差创建了一个私有方法。我将要做的计算将比这更复杂（例如，我将比较列中的每个元素），我的方向是否正确，还是应该更多地研究用户定义的函数？

Answer 1

在优秀的留档中给出了一个简单的例子，其中一整个部分专门用于UDF：

import org.apache.spark.sql._

val df = Seq(("id1", 1), ("id2", 4), ("id3", 5)).toDF("id", "value")
val spark = df.sparkSession
spark.udf.register("simpleUDF", (v: Int) => v * v)
df.select($"id", callUDF("simpleUDF", $"value"))

Answer 2

在Spark DataFrame中，您不能使用您想到的方法遍历列的元素，因为列不是可迭代的对象。

但是，要处理列的值，您有一些选项，正确的选项取决于您的任务：

1）使用现有的内置函数

Spark SQL已经有很多用于处理列的有用函数，包括聚合和转换函数。其中大多数可以在< code>functions包中找到(此处为文档)。其他一些函数(一般是二进制函数)可以直接在< code>Column对象中找到(这里是文档)。所以，如果你能使用它们，通常是最好的选择。注意:不要忘记窗口函数。

2）创建UDF

如果无法使用内置函数完成任务，则可以考虑定义 UDF（用户定义函数）。如果您可以独立处理列的每个项目，并且希望生成与原始列具有相同行数的新列（而不是聚合列），则它们非常有用。这种方法非常简单：首先，定义一个简单的函数，然后将其注册为UDF，然后使用它。例：

def myFunc: (String => String) = { s => s.toLowerCase }

import org.apache.spark.sql.functions.udf
val myUDF = udf(myFunc)

val newDF = df.withColumn("newCol", myUDF(df("oldCol")))

要了解更多信息，这里有一篇很好的文章。

3） 使用UDAF

如果您的任务是创建聚合数据，则可以定义UDAF（用户定义的聚合函数）。我在这方面没有太多经验，但我可以为您指出一个很好的教程：

https://ragrawal . WordPress . com/2015/11/03/spark-custom-udaf-example/

4） 回退到 RDD 处理

如果您真的不能使用上面的选项，或者如果您的处理任务依赖于不同的行来处理，并且它不是一个聚合，那么我认为您必须选择您想要的列，并使用相应的RDD来处理它。示例:

val singleColumnDF = df("column")

val myRDD = singleColumnDF.rdd

// process myRDD

所以，这是我能想到的选择。希望有帮助。