如何使用 Java UDF 向火花数据帧添加新列

Question

我有一个数据集

数据集输入

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

预期数据集输出DS

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

我试过的代码。

 inputDS.withColumn("value_new",functions.lit(inputDS.map(new MapFunction<Row,Double>(){

 public double call(Row row){
 String aggregateType = row.getAS("aggregateType");
 List<long> timeList = row.getList("time");
 List<long> valueList= row.getList("value");  

 return  getAggregate(aggregateType ,timeList,valueList);    

 }}),Encoders.DOUBLE())));

错误

 Unsupported literal type class org.apache.spark.sql.Dataset [value:double]

请注意，如果我错误地使用了< code>map功能，请告诉我是否有任何解决方法。

谢谢你。！

Answer 1

您会收到此错误，因为您正在尝试使用 Dataset.map（） 的结果创建函数文本 （lit（）），您可以在文档中看到它是一个数据集。您可以在 Dataset.with 哥伦布（） 的 API 中看到，您需要一个作为列的参数。

看起来您需要创建一个用户定义的函数。看看如何使用JAVA调用Spark数据帧上的UDF？