如何在 Java8 中用 Spark 代码编写 mapFunction?错误:映射(函数 1，编码器)不适用于参数

Question

>我有一个地图函数如下

Function1<CompanyInfoRecordNew, CompanyInfoTransformedRecord> mapFunction = ( record ) ->{
CompanyInfoTransformedRecord  transRec = new CompanyInfoTransformedRecord();
//popluate 
return transRec ;
};

它采用 CompanyInfoRecordNew 对象并返回 CompanyInfoTransformedRecord。

但是在下面调用它时

JavaRDD companyInfoTransformedRecord = baseDs.map(mapFunction， comanyInfoTransEncoder(;

给出错误 ：

数据集类型中的方法映射(函数 1、编码器(不适用于参数(函数 1、编码器(

这是怎么回事？

相同的代码链接：

https://gist.github.com/BdLearnerr/cbfea1c8471557bb33449f882cc1854a

如何从地图函数返回列表？

链接中更新的代码

https://gist.github.com/BdLearnerr/cbfea1c8471557bb33449f882cc1854a

第二节 ：

MapFunction<CompanyInfoRecordNew, List<CompanyInfoTransformedRecord>>  mapFunction = ( record ) ->{
List<CompanyInfoTransformedRecord>  transRecList = new ArrayList<CompanyInfoTransformedRecord>();
return transRecList ;
}

Dataset<List<CompanyInfoTransformedRecord>> companyInfoTransformedRecords = baseDs.map(mapFunction, comanyInfoTransEncoder);

//错误

方法映射(函数 1，编码器( 类型 数据集不适用于参数 地图功能>， 编码器(

companyInfoTransformedRecord.show();

由于错误而无法显示，如何让 show(( 工作。 ？？

Answer 1

您的导入是错误的，而不是这个：

import org.apache.calcite.linq4j.function.Function1;

使用这个

org.apache.spark.api.java.function.MapFunction;
baseDs.map(new MapFunction<CompanyInfoRecordNew, CompanyInfoTransformedRecord>() {...}, encoder);

Answer 2

要获取对象并返回对象列表，我们需要使用平面图而不是地图，如下所示

FlatMapFunction<Original, Derived>  flatMapFunction = ( record ) ->{
List<Derived>  transRecList = new ArrayList<Derived>();
Derived  transRec = new Derived();
transRecList.add(transRec);
Derived  transRec2 = new Derived();
transRecList.add(transRec2);
return (Iterator<Derived>) transRecList.iterator();
};

Dataset Dederds = baseDs.flatMap(flatMapFunction， comanyInfoTransEncoder(;