用于行类型Spark数据集的编码器-Java 学习之路

我想在DataSet中为Row类型编写一个编码器，用于我正在进行的 Map 操作 . 基本上，我不明白如何编写编码器 .

以下是 Map 操作的示例：

In the example below, instead of returning Dataset<String>, I would like to return Dataset<Row>

Dataset<String> output = dataset1.flatMap(new FlatMapFunction<Row, String>() {
            @Override
            public Iterator<String> call(Row row) throws Exception {

                ArrayList<String> obj = //some map operation
                return obj.iterator();
            }
        },Encoders.STRING());

据我所知，编码器需要编写如下代码：

Encoder<Row> encoder = new Encoder<Row>() {
        @Override
        public StructType schema() {
            return join.schema();
            //return null;
        }

        @Override
        public ClassTag<Row> clsTag() {
            return null;
        }
    };

但是，我不理解编码器中的clsTag（），我试图找到一个可以演示类似的运行示例（即行类型的编码器）

编辑 - 这不是所提问题的副本：Encoder error while trying to map dataframe row to updated row因为答案谈到在Spark 2.x中使用Spark 1.x（我没有这样做），我也在寻找Row类的编码器而不是解决一个错误 . 最后，我一直在寻找Java解决方案，而不是Scala .

2 回答

15

我有同样的问题... Encoders.kryo(Row.class)) 为我工作 .

作为奖励，Apache Spark调优文档引用Kryo它，因为序列化速度更快“通常高达10倍”：

https://spark.apache.org/docs/latest/tuning.html

回复于 2024-04-29T21:49:11+08:00

答案是使用StructType和数据集的模式使用StructType .

以下是使用数据集进行flatmap操作的工作示例：

StructType structType = new StructType();
    structType = structType.add("id1", DataTypes.LongType, false);
    structType = structType.add("id2", DataTypes.LongType, false);

    ExpressionEncoder<Row> encoder = RowEncoder.apply(structType);

    Dataset<Row> output = join.flatMap(new FlatMapFunction<Row, Row>() {
        @Override
        public Iterator<Row> call(Row row) throws Exception {
            // a static map operation to demonstrate
            List<Object> data = new ArrayList<>();
            data.add(1l);
            data.add(2l);
            ArrayList<Row> list = new ArrayList<>();
            list.add(RowFactory.create(data.toArray()));
            return list.iterator();
        }
    }, encoder);

回复于 2024-04-29T21:49:11+08:00

用于行类型Spark数据集的编码器

2 回答

相关问题