Spark 在 UDF Java 中从 WrappedArray<WrappedArray<Double>> 获取值答案

【问题标题】：Spark get value form WrappedArray<WrappedArray<Double>> in UDF JavaSpark 在 UDF Java 中从 WrappedArray<WrappedArray<Double>> 获取值
【发布时间】：2018-01-25 06:01:32
【问题描述】：

我的Dataset<Row> 中有一个包含WrappedArray<WrappedArray<Double>> 的列。我将此列传递给UDF 以提取其中一个值。

我将如何访问这个嵌套结构中的 Doubles？

我想做这样的事情：

sparkSession.udf().register(ADD_START_TOTAL, (UDF1<WrappedArray<WrappedArray<Double>>, Double>) (totals) -> totals[0][1], DataTypes.DoubleType);

下面是我调用 Dataset.show() 方法时列的示例，我的数据集如下所示。

[WrappedArray(2.0...

编辑：找到这篇文章How to cast a WrappedArray[WrappedArray[Float]] to Array[Array[Float]] in spark (scala) 但不确定如何将其转换为 Java。

【问题讨论】：

标签： java apache-spark apache-spark-sql user-defined-functions

【解决方案1】：

考虑您的 Dataset<Row> ds1 具有以下架构的 value 列

root
 |-- value: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: double (containsNull = false)

如下定义您的 UDF1 函数。

static UDF1<WrappedArray<WrappedArray<Double>>, List<Double>> getValue = new UDF1<WrappedArray<WrappedArray<Double>>, List<Double>>() {
public List<Double> call(WrappedArray<WrappedArray<Double>> data) throws Exception {
        List<Double> doubleList = new ArrayList<Double>();
        for(int i=0; i<data.size(); i++){
            doubleList.addAll(JavaConversions.seqAsJavaList(data.apply(i)));
        }
        return doubleList;
    }
}

现在注册并调用UDF1函数，如下所示。

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.callUDF;
import scala.collection.JavaConversions;

// register UDF
spark.udf().register("getValue", getValue, DataTypes.createArrayType(DataTypes.DoubleType));

// Call UDF
Dataset<Row> ds2 = ds1.select(col("*"), callUDF("getValue", col("value")).as("udf-value"));
ds2.show(false);

【讨论】：