行类型 Spark 数据集的编码器

Question

我想为 DataSet 中的 Row 类型编写一个编码器，用于我正在执行的映射操作。本质上，我不明白如何编写编码器。

下面是一个地图操作的例子：

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()，并且我正在尝试找到一个可以演示类似内容的运行示例（即行类型的编码器）

编辑 - 这不是所提到问题的副本：尝试将数据帧行映射到更新的行时出现编码器错误，因为答案谈到在 Spark 2.x 中使用 Spark 1.x （我没有这样做），也我正在寻找 Row 类的编码器，而不是解决错误。最后，我在 Java 中寻找解决方案，而不是在 Scala 中。

Answer 1

答案是使用 RowEncoder 并使用 StructType 的数据集模式。

下面是使用数据集进行平面地图操作的工作示例：

    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);

Answer 2

我有同样的问题...

Encoders.kryo(Row.class))

为我工作。

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

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

Answer 3

Rowencoder

是未记录的内部API，当您升级Spark时会意外中断。 （正如上面的评论所说，

RowEncoder.apply

在 Spark 3.5 中不再起作用。）

这是完全依赖公共 API 的正确方法。

import java.util.List;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

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

Encoder<Row> encoder = spark.createDataFrame(List.of(), structType).encoder();

或者，如果您使用的是 Spark 3.5，还有一种更简单的方法。

Encoder<Row> encoder = Encoders.row(structType);