Флинк, как десериализовать protobuf с помощью ProtobufSerializer? - PullRequest
Новая
       39

Флинк, как десериализовать protobuf с помощью ProtobufSerializer?

0 голосов
/ 28 февраля 2019

Мне нужно использовать kafka через flink, к сожалению, сообщение kafka использует protobuf в serde, совершенно не знаю, как с этим справиться, вот код из интернета, но я не могу заставить его работать. 

...    
import com.google.protobuf.InvalidProtocolBufferException;
import com.twitter.chill.protobuf.ProtobufSerializer;

public class Protobuf2 {

    public static void main(String[] args) throws Exception {
        String inputTopic = "zz";
        String consumerGroup = "foobar";
        String address = "localhost:9092";
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.getConfig().registerTypeWithKryoSerializer(ExchangeMessage.Order.class, ProtobufSerializer.class);

        FlinkKafkaConsumer<ExchangeMessage.Order> flinkKafkaConsumer = createStringConsumerForTopic(inputTopic, address,
                consumerGroup);
        DataStream<ExchangeMessage.Order> input = environment.addSource(flinkKafkaConsumer);

        input.map(order ->
            {
                System.out.println("foooooooo, " + order.getOid());
                return order;
            });

        environment.execute("kafka02");
    }

    public static FlinkKafkaConsumer<ExchangeMessage.Order> createStringConsumerForTopic(String topic,
            String kafkaAddress, String kafkaGroup) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", kafkaAddress);
        props.setProperty("group.id", kafkaGroup);
        FlinkKafkaConsumer<ExchangeMessage.Order> consumer = new FlinkKafkaConsumer<>(topic,
                new OrderSerDeSchema(), props);
        return consumer;
    }


    @SuppressWarnings("serial")
    static class OrderSerDeSchema
            implements DeserializationSchema<ExchangeMessage.Order>, SerializationSchema<ExchangeMessage.Order> {

        @Override
        public ExchangeMessage.Order deserialize(byte[] message) throws IOException {
            ExchangeMessage.Order order = null;
            try {
                order = ExchangeMessage.Order.parseFrom(message);
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
            return order;
        }

        @Override
        public boolean isEndOfStream(ExchangeMessage.Order nextElement) {
            return false;
        }

        @Override
        public TypeInformation<ExchangeMessage.Order> getProducedType() {
            return null;
        }

        @Override
        public byte[] serialize(ExchangeMessage.Order element) {
            return new byte[0];
        }
    }

}

Это не работает, это дает NPE: 

Caused by: java.lang.NullPointerException
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:577)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)
    at ......

Кто-нибудь знает, что я здесь не так делаю?Это единственный способ бросить протобуф - использовать твиттер ProtobufSerializer?Или есть другой путь?

1 Ответ

0 голосов
/ 30 мая 2019

добавить зависимость 

<dependency>
    <groupId>com.twitter</groupId>
    <artifactId>chill-protobuf</artifactId>
    <version>0.7.6</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>1.8.0</version>
</dependency>

и добавить это в свой код 

env.getConfig().registerTypeWithKryoSerializer(MockMessageProto.MockMessage.class, ProtobufSerializer.class);

возможно, вы столкнетесь с конфликтом версий, измените версию chill-protobuf, чтобы исправить ее

...