Написание объекта protobuf в паркете с использованием Apache Beam - PullRequest
Новая
       40

Написание объекта protobuf в паркете с использованием Apache Beam

0 голосов
/ 18 сентября 2018

Я получаю данные protobuf из google pub / sub и десериализирую данные в объект типа Message. Итак, я получаю 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 может решить проблему (пока не знаю, как это сделать). Есть какой-нибудь простой способ написать в формате паркета?

1 Ответ

0 голосов
/ 27 сентября 2018

Это можно решить с помощью следующей библиотеки: 

<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

...