如何将 Dataset<Row> 列转换为非原始数据类型

Question

我有一个Dataset<Row>，其中有四列，两列是非原始数据类型List<Long> and List<String>。

  +------+---------------+---------------------------------------------+---------------+
  |    Id| value         |     time                                      |aggregateType  |
  +------+---------------+---------------------------------------------+---------------+
  |0001  |  [1.5,3.4,4.5]| [1551502200000,1551502200000,1551502200000] | Sum             |
  +------+---------------+---------------------------------------------+---------------+

我有一个 UDF3，它接受三个参数并返回一个 Double 值 UDF3<String,List<Long>,List<String>,Double>。

所以当我调用 UDF 时，它会抛出一个异常

错误

caused by java.lang.classcastexception scala.collection.mutable.wrappedarray$ofref cannot be cast to java.lang.List

但如果我将类型更改为 String 就像 UDF3<String,String,String,Double> 一样，它不会抱怨。

抛出异常的代码

 UDF3<String,List<Long>,List<String>,Double> getAggregate = new UDF3<String,List<Long>,List<String>,Double>() {

 public Double call(String t1,List<Long> t2,List<String> t3) throws Exception {

 //do some process to return double

  return double;
  }

  sparkSession.udf().register("getAggregate_UDF",getAggregate, DataTypes.DoubleType);

  inputDS = inputDs.withColumn("value_new",callUDF("getAggregate_UDF",col("aggregateType"),col("time"),col("value")));

将所有类型改为String后的代码

 UDF3<String,String,String,Double> getAggregate = new UDF3<String,String,String,Double>() {

 public Double call(String t1,String t2,String t3) throws Exception {

 //code to convert t2 and t3 to List<Long> and List<String> respectively

 //do some process to return double

  return double;
  }

  sparkSession.udf().register("getAggregate_UDF",getAggregate, DataTypes.DoubleType);

  inputDS = inputDs.withColumn("value_new",callUDF("getAggregate_UDF",col("aggregateType"),col("time").cast("String"),col("value").cast("String")));

上述代码有效，但需要手动转换String to List。

需要帮助

I) 如何在数据集中转换非原始数据类型List<Long> and List<String> 以克服caused by java.lang.classcastexception scala.collection.mutable.wrappedarray$ofref cannot be cast to java.lang.List

II) 如果有任何解决方法，请建议我

谢谢。

Answer 1

您的 UDF 将始终接收 WrappedArray 实例而不是 List，因为这是引擎存储它们的方式。

你需要这样写：

import scala.collection.mutable.WrappedArray;
import scala.collection.JavaConversions;

UDF3<String, WrappedArray<Long>, WrappedArray<String>, Double> myUDF = new UDF3<String, WrappedArray<Long>, WrappedArray<String>, Double> () {
      public Double call(String param1, WrappedArray<Long> param2, WrappedArray<String> param3) throws Exception {
        List<Long> param1AsList = JavaConversions.seqAsJavaList(param1);
        List<String> param2AsList = JavaConversions.seqAsJavaList(param2);

        ... do work ...

        return myDoubleResult;
    }
};

Answer 2

这是我的例子，你必须使用 WrappedArray 来接收数组并转换为列表

 /*
     +------+---------------+---------------------------------------------+---------------+
     |    Id| value         |     time                                      |aggregateType  |
     +------+---------------+---------------------------------------------+---------------+
     |0001  |  [1.5,3.4,4.5]| [1551502200000,1551502200000,1551502200000] | Sum             |
     +------+---------------+---------------------------------------------+---------------+
     **/

    StructType dataSchema = new StructType(new StructField[] {createStructField("Id", DataTypes.StringType, true),
                                                              createStructField("value",
                                                                                DataTypes.createArrayType(DataTypes.DoubleType,
                                                                                                          false),
                                                                                false),

                                                              createStructField("time",
                                                                                DataTypes.createArrayType(DataTypes.LongType,
                                                                                                          false),
                                                                                false),
                                                              createStructField("aggregateType",
                                                                                DataTypes.StringType,
                                                                                true),});

    List<Row> data = new ArrayList<>();

    data.add(RowFactory.create("0001",
                               Arrays.asList(1.5, 3.4, 4.5),
                               Arrays.asList(1551502200000L, 1551502200000L, 1551502200000L),
                               "sum"));
    Dataset<Row> example = spark.createDataFrame(data, dataSchema);
    example.show(false);

    UDF3<String, WrappedArray<Long>, WrappedArray<Double>, Double> myUDF = (param1, param2, param3) -> {

        List<Long> param1AsList = JavaConversions.seqAsJavaList(param2);
        List<Double> param2AsList = JavaConversions.seqAsJavaList(param3);

        //Example
        double myDoubleResult = 0;
        if ("sum".equals(param1)) {

            myDoubleResult = param2AsList.stream()
                                         .mapToDouble(f -> f)
                                         .sum();
        }

        return myDoubleResult;
    };

    spark.udf()
         .register("myUDF", myUDF, DataTypes.DoubleType);

    example = example.withColumn("new", callUDF("myUDF", col("aggregateType"), col("time"), col("value")));
    example.show(false);

您可以从github获取它