spark UDF对阵列进行操作

Question

我有一个spark数据帧，如：

+-------------+------------------------------------------+
|a            |destination                               |
+-------------+------------------------------------------+
|[a,Alice,1]  |[[b,Bob,0], [e,Esther,0], [h,Fraudster,1]]|
|[e,Esther,0] |[[f,Fanny,0], [d,David,0]]                |
|[c,Charlie,0]|[[b,Bob,0]]                               |
|[b,Bob,0]    |[[c,Charlie,0]]                           |
|[f,Fanny,0]  |[[c,Charlie,0], [h,Fraudster,1]]          |
|[d,David,0]  |[[a,Alice,1], [e,Esther,0]]               |
+-------------+------------------------------------------+

以

|-- destination: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- id: string (nullable = true)
 |    |    |-- name: string (nullable = true)
 |    |    |-- var_only_0_and_1: integer (nullable = false)

如何构造一个在列目标上运行的UDF，即由火花的UDF collect_list创建的包装数组，以计算变量var_only_0_and_1平均值？

Answer 1

为此，您可以使用本机 spark sql 函数。

df.withColumn("dest",explode(col("destination")).
groupBy("a").agg(avg(col("dest").getField("var_only_0_and_1")))

Answer 2

只要UDF的方法签名正确，就可以直接对数组进行操作（这在过去对我打击很大）。数组列在UDF中作为序列可见，在结构中作为行可见，因此您需要这样的东西：

def test (in:Seq[Row]): String = {
  // return a named field from the second struct in the array
  in(2).getAs[String]("var_only_0_and_1")
}

var udftest = udf(test _)

我已经在看起来像你的数据上测试了这一点。我猜想有可能迭代Seq[Row]的字段，以实现您想要的。

老实说，我一点也不确定这样做的类型安全性，而且我相信爆炸是按照@ayplam的说法做的更好的方法。内置函数通常比开发人员提供的任何UDF都快，因为Spark无法优化UDF。