javaAPI中是否有将数据集转换为map（）并返回数据集？

Question

我正在使用spark-sql-2.4。1v和Java 8。我有一个用例，如下所示，

Dataset<Row> ds = //a Dataset<Row> read from DB

我需要根据另一个数据集的条目进行一些操作。

List<String> codesList = Array.asList("code1","code2")
Dataset<Row> codes = sc.createDataSet(codesList , Encoders.bean(String.class))

我需要并行处理所有代码。做同样的事情，我正在尝试如下：

 Dataset<Row> ds_res =  codes.map( x_cod ->   //map throwing an error
        calcFunction(sparkSession, filePath, ds ,x_cod );
    }).reduce(new Function2<Dataset<Row> df1,Dataset<Row> df2) => df1.union(df2))

 ds_res .write().path(filePath).mode("append").save();

    public static Dataset<Row> calcFunction(sparkSession, filePath, ds ,x_cod ){
         //some complex calculation based on x_cod 

        return ds_res ; // return ds_res  for further processing
    }

如何在集群上并行工作？

Answer 1

将列表编码到数据集比编码更可行。如果您计划使用bean类，您可以将其编码为该类型的Dataset

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.util.Arrays;
import java.util.List;

public class ParallelizeArray {

    public static void main(String[] args) {
        final SparkSession sparkSess = Constant.getSparkSess();
        List<String> codesList = Arrays.asList("code1", "code2");
        final Dataset<String> dataFrame = sparkSess.createDataset(codesList, Encoders.STRING());
        dataFrame.write().mode(SaveMode.Append).csv("src/main/resources/paraArray");
    }
}

或者使用

final Encoder<Dataset> bean = Encoders.bean(Dataset.class);
Dataset<Row> ds_res = codes.map((MapFunction<String, Dataset>) x_cod -> calcFunction(sparkSess, filePath, ds ,x_cod),bean)
                .reduce((ReduceFunction<Dataset>) (df1, df2) -> df1.union(df2));



    public static Dataset<Row> calcFunction(SparkSession sparkSession, String filePath, Dataset<Row> ds ,String x_cod ){
        Dataset<Row> ds_res = null;//some complex calculation based on x_cod
        return ds_res ; // return ds_res  for further processing
    }