Apache Flink - 如何实现自定义 Deserializer 实现 DeserializationSchema

Question

我正在使用 Flink，并且正在使用 Kafka 连接器。我从 flink 收到的消息是逗号分隔的项目列表。 “'a','b','c',1,0.1 ....'12:01:00.000'” 其中一个包含事件时间，我想将此事件时间用于每个分区的水印（在 kafka 源中），然后将此事件时间用于会话窗口。 我的情况与平常有些不同，因为据我了解，人们通常使用“kafka Timestamps”和 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);
    }
}