行类型 Spark 数据集的编码器
Encoder for Row Type Spark Datasets
我想为 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(我没有这样做),也我正在寻找行 class 的编码器而不是解决错误。最后,我在 Java 中寻找解决方案,而不是在 Scala 中。
答案是使用RowEncoder and the schema of the dataset using 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);
我遇到了同样的问题...Encoders.kryo(Row.class))
对我有效。
作为奖励,Apache Spark 调优文档引用了 Kryo,因为它的序列化速度更快 "often as much as 10x":
我想为 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(),我正试图找到一个 运行 示例,它可以演示类似的东西(即行类型的编码器)
编辑 - 这不是提到的问题的副本:
答案是使用RowEncoder and the schema of the dataset using 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);
我遇到了同样的问题...Encoders.kryo(Row.class))
对我有效。
作为奖励,Apache Spark 调优文档引用了 Kryo,因为它的序列化速度更快 "often as much as 10x":