Apache Flink-如何实现自定义反序列化器实现DeserializationSchema

Question

我用的是Flink我用的是Kafka连接器。我从flink收到的消息是一个逗号分隔的项目列表。“'a','b','c',1,0.1...'12:01:00.000'”其中一个包含事件时间，我想将此事件时间用于每个分区的水印（在kafka源代码中），然后将此事件时间用于会话窗口化。我的情况与通常情况有点不同，因为根据我的理解，人们通常使用“Kafka时间戳”和SimpleStringSchema（）。在我的例子中，我不得不编写自己的反序列化器，实现DeserializationSchema并返回元组或POJO。所以基本上用我自己的函数替换SimpleStringSchema（）。Flink提供了一些开箱即用的反序列化程序，但我真的不知道如何创建一个自定义的反序列化逻辑。

查看flink网站我发现：

https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/connectors/kafka.html

https://github.com/apache/flink-playgrounds/blob/master/docker/ops-playground-image/java/flink-playground-clickcountjob/src/main/java/org/apache/flink/playgrounds/ops/clickcount/records/clickeventdeserializationschema.java

我真的需要一个例子，我可以如何做一个列表。上面提到的是JSON，所以给了我理论和概念，但我被困在那里了。

Answer 1

您应该像这样介绍POJO

public class Event implements Serializable {
    ...
    private Long timestamp;
}

并实现类似于链接中的简单反序列化器--您可以通过逗号手动拆分消息字符串来解析行,或者使用开箱即用的csv阅读器（如opencsv）将行解析为POJO：

public class EventDeserializationSchema implements DeserializationSchema<Event> {

    private static final long serialVersionUID = 1L;

    @Override
    public ClickEvent deserialize(byte[] message) throws IOException {
        String line = new String(message, StandardCharsets.UTF_8);
        String[] parts = line.split(",");
        
        Event event = new Event();
        // TODO: parts to event here
        return event;
    }

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

    @Override
    public TypeInformation<Event> getProducedType() {
        return TypeInformation.of(Event.class);
    }
}