Apache Flink - 如何使用 AWS Kinesis 发送和使用 POJO

Question

我想使用 Flink 使用来自 Kinesis 的 POJO。
有没有关于如何正确发送和反序列化消息的标准？

谢谢

Answer 1

我解决了这个问题：

DataStream<SamplePojo> kinesis = see.addSource(new FlinkKinesisConsumer<>(
        "my-stream",
        new POJODeserializationSchema(),
        kinesisConsumerConfig));

和

public class POJODeserializationSchema extends AbstractDeserializationSchema<SamplePojo> {
    private ObjectMapper mapper;

    @Override
    public SamplePojo deserialize(byte[] message) throws IOException {
        if (mapper == null) {
            mapper = new ObjectMapper();
        }

        SamplePojo retVal = mapper.readValue(message, SamplePojo.class);

        return retVal;
    }

    @Override
    public boolean isEndOfStream(SamplePojo nextElement) {
        return false;
    }
}