java - Beam 写入 avro 文件序列化错误
问题描述
我按照在Beam 文档中编写AVRO 文件的示例进行操作。但这给了我一个错误Caused by: java.io.NotSerializableException: org.apache.avro.Schema$RecordSchema
。p.run().waitUntilFinish()
但是,如果我从 AVRO 文件中读取并将其写入另一个 AVRO 输出,则它可以正常工作。我的目标是从任意输入源编写 AVRO 文件。有没有人见过类似的问题?你是怎么解决的?
public class WriteAvro {
public interface CsvToAvroOptions extends PipelineOptions {
@Description("Path of the file to read from")
@Default.String("test.avro")
String getInputFile();
void setInputFile(String value);
}
static void run(CsvToAvroOptions options) throws IOException {
final Schema schema = new Schema.Parser().parse(Resources.getResource("person.avsc").openStream());
Pipeline p = Pipeline.create(options);
// This works fine
// PCollection<GenericRecord> input = p.apply(AvroIO.readGenericRecords(schema).from(options.getInputFile()));
// This doesn't work
PCollection<GenericRecord> input =
p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
.apply(ParDo.of(new DoFn<String, GenericRecord>() {
@ProcessElement
public void processElement(ProcessContext c) {
GenericRecord record = new GenericData.Record(schema);
record.put("name", "John Doe");
record.put("age", 42);
record.put("siblingnames", Lists.newArrayList("Jimmy", "Jane"));
c.output(record);
}
}))
.setCoder(AvroCoder.of(GenericRecord.class, schema));
input.apply(AvroIO.writeGenericRecords(schema).to("prefix"));
p.run().waitUntilFinish();
}
public static void main(String[] args) throws IOException {
CsvToAvroOptions options =
PipelineOptionsFactory.fromArgs(args).withValidation().as(CsvToAvroOptions.class);
run(options);
}
}
- 梁版本:2.11.0
- 亚军:直接亚军
解决方案
Schema
不可序列化导致此错误。您可以将架构存储为文本并在设置 DoFn 时对其进行解析。
这是你如何做到的。
public interface CsvToAvroOptions extends PipelineOptions {
@Description("Path of the file to read from")
@Default.String("test.avro")
String getInputFile();
void setInputFile(String value);
}
private static class ConstructAvroRecordsFn extends DoFn<String, GenericRecord> {
private final String schemaJson;
private Schema schema;
ConstructAvroRecordsFn(Schema schema){
schemaJson = schema.toString();
}
@Setup
public void setup(){
schema = new Schema.Parser().parse(schemaJson);
}
@ProcessElement
public void processElement(ProcessContext c) {
GenericRecord record = new GenericData.Record(schema);
record.put("name", "John Doe");
record.put("age", 42);
record.put("siblingnames", Lists.newArrayList("Jimmy", "Jane"));
c.output(record);
}
}
static void run(CsvToAvroOptions options) throws IOException {
final Schema schema = new Schema.Parser().parse(Resources.getResource("person.avsc").openStream());
Pipeline p = Pipeline.create(options);
// This works fine
// PCollection<GenericRecord> input = p.apply(AvroIO.readGenericRecords(schema).from(options.getInputFile()));
// This doesn't work
PCollection<GenericRecord> input =
p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
.apply(ParDo.of(new ConstructAvroRecordsFn(schema)))
.setCoder(AvroCoder.of(GenericRecord.class, schema));
input.apply(AvroIO.writeGenericRecords(schema).to("prefix"));
p.run().waitUntilFinish();
}
public static void main(String[] args) throws IOException {
CsvToAvroOptions options =
PipelineOptionsFactory.fromArgs(args).withValidation().as(CsvToAvroOptions.class);
run(options);
}
}
推荐阅读
- javascript - 未捕获(承诺中)类型错误:无法读取未定义的属性“启用”
- mysql - 我的 SUM 查询返回 2 行,需要一些建议
- html - 无法将 CSS 文件链接到 HTML 文件
- javascript - 有没有办法制作没有边框的Material ui Card
- javascript - 如何摆脱反应图标(引导程序)中的可用空间?
- python - 为什么我不能使用 Gensim 下载 API 下载数据集
- c# - 在 C# 中连接到 MS Access 数据库时出现错误
- python - 尽管网络大、样本量小且数据经过预处理,Tensorflow LSTM 仍无法学习
- swiftui - 检测用户何时停止/暂停在 TextField 中写入 - SwiftUI
- c# - Xamarin.Forms - 根据设备分辨率更改设备方向