使用阿帕奇光束在镶木地板中书写原型对象

Question

我从 google pub/sub 获取 protobuf 数据并将数据反序列化为消息类型对象。所以我得到了PCollection<Message>类型的对象。下面是示例代码：

public class ProcessPubsubMessage extends DoFn<PubsubMessage, Message> {
@ProcessElement
public void processElement(@Element PubsubMessage element, OutputReceiver<Message> receiver) {
byte[] payload = element.getPayload();
try {
Message message = Message.parseFrom(payload);
receiver.output(message);
} catch (InvalidProtocolBufferException e) {
LOG.error("Got exception while parsing message from pubsub. Exception =>" + e.getMessage());
}
}
}
PCollection<Message> event = psMessage.apply("Parsing data from pubsub message",
ParDo.of(new ProcessPubsubMessage()));

我想在PCollection<Message> event上应用转换以镶木地板格式编写。我知道 apache beam 提供了ParquetIO，但它适用于PCollection<GenericRecord>类型，从Message到GenericRecord的转换可能会解决问题(但不知道如何做到这一点(。有什么简单的方法可以用镶木地板格式书写吗？

Answer 1

可以使用以下库解决：

<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro-protobuf</artifactId>
<version>1.7.7</version>
</dependency>
private GenericRecord getGenericRecord(Event event) throws IOException {
ProtobufDatumWriter<Event> datumWriter = new ProtobufDatumWriter<Event>(Event.class);
ByteArrayOutputStream os = new ByteArrayOutputStream();
Encoder e = EncoderFactory.get().binaryEncoder(os, null);
datumWriter.write(event, e);
e.flush();
ProtobufDatumReader<Event> datumReader = new ProtobufDatumReader<Event>(Event.class);
GenericDatumReader<GenericRecord> genericDatumReader = new GenericDatumReader<GenericRecord>(datumReader.getSchema());
GenericRecord record = genericDatumReader.read(null, DecoderFactory.get().binaryDecoder(new ByteArrayInputStream(os.toByteArray()), null));
return record;
}

详情： https://gist.github.com/alexvictoor/1d3937f502c60318071f