在Spark数据帧udf中，像结构（col1， col2）的函数参数类型是什么？

Question

背景：

我有一个数据帧，它有三列:< code>id，x，y。x，y是双精度的。

• 首先，我构造（“x”），col（“y”））来获取坐标列。
• 然后组比（“id”）和 agg（collect_list（“坐标”））

所以现在df只有两列：id，坐标。

我认为坐标的数据类型是集合.可变.包裹阵列[（双，双）]。所以我把它传给了udf。但是，数据类型是错误的。我在运行代码时遇到错误。我不知道为什么。结构（col1，col2）的真实数据类型是什么？或者有另一种方法可以轻松获得正确答案？

这是代码：

def getMedianPoint = udf((array1: collection.mutable.WrappedArray[(Double,Double)]) => {  
    var l = (array1.length/2)
    var c = array1(l)
    val x = c._1.asInstanceOf[Double]
    val y = c._2.asInstanceOf[Double]
    (x,y)
})

df.withColumn("coordinate",struct(col("x"),col("y")))
  .groupBy(col("id"))
  .agg(collect_list("coordinate").as("coordinate")
  .withColumn("median",getMedianPoint(col("coordinate")))

非常感谢！

Answer 1

我认为坐标的数据类型是collection.mutable.WrappedArray[（双，双）]

是的，你这么说完全正确。你在udf函数中定义的数据类型和你传递的参数也是正确的。但是主要问题是struct列的键的名称。因为您肯定有以下问题

由于数据类型不匹配，无法解析“UDF（坐标）：参数 1 需要数组

只需使用< code>alias将结构键重命名为

df.withColumn("coordinate",struct(col("x").as("_1"),col("y").as("_2")))
  .groupBy(col("id"))
  .agg(collect_list("coordinate").as("coordinate"))
    .withColumn("median",getMedianPoint(col("coordinate")))

以便keyname匹配。

但

这将引起另一个问题

  var c = array1(l)

原因：java.lang.类投射异常： 组织.apache.spark.sql.catalyst.expressions.通用行与化学不能被投射到标拉。元组 2

因此，我建议您将udf函数更改为

import org.apache.spark.sql.functions._

def getMedianPoint = udf((array1: Seq[Row]) => {
  var l = (array1.length/2)
  (array1(l)(0).asInstanceOf[Double], array1(l)(1).asInstanceOf[Double])
})

这样您甚至不需要使用alias。所以完整的解决方案是

import org.apache.spark.sql.functions._

def getMedianPoint = udf((array1: Seq[Row]) => {
  var l = (array1.length/2)
  (array1(l)(0).asInstanceOf[Double], array1(l)(1).asInstanceOf[Double])
})

df.withColumn("coordinate",struct(col("x"),col("y")))
  .groupBy(col("id"))
  .agg(collect_list("coordinate").as("coordinate"))
    .withColumn("median",getMedianPoint(col("coordinate")))
  .show(false)

我希望答案有帮助