Apache Flink NoSerialization 用于业务逻辑类答案

【问题标题】：Apache Flink NoSerialization for business logic classApache Flink NoSerialization 用于业务逻辑类
【发布时间】：2022-01-31 18:23:44
【问题描述】：

我正在使用带有 Kakfa 的 Apache Flink，我可以在其中正确地使用来自主题的消息并对其进行反序列化。

我正在使用 flink 数据流 Map 函数来处理收到的 kafka 消息，但 flink 期望我所有的业务对象都是可序列化的。

下面是伪代码

TestRequestListener listener = this.context.getBean(TestRequestListener.class);
KafkaConsumer<KafkaTestBatch> kafkaConsumer = KafkaSource<>.builder()...build();
StreamExecutionEnviroment env = ...
DataStream<KafkaTestBatch> kafkaSource = env.fromSource(kafkaConsumer,...);
kafkaSource.map(kafkaTestBatch -> {
listener.listen(kafkaTestBatch);
});

这里 TestRequestListener 有业务逻辑，它甚至可以查询数据库。

Flink 期望 TestRequestListener 及其所有依赖项都是可序列化的。我的用例是否正确，我使用的是 flink correclty 吗？

【问题讨论】：

你在 Flink 中使用 Springboot 库吗？
问题不在于 Flink。这是它运行分布式进程的事实。您可以将context.getBean 方法调用移动到kafkaSource.map，但是您的上下文变量需要可序列化。基本上，您的侦听器需要与 mapper 函数在同一台机器上运行，这需要从驱动程序中序列化，here。为什么不直接把listen函数代码放在map body中呢？
@Ramachandran 是的
@onecric 我做不到有很多逻辑

标签： java spring-boot apache-kafka apache-flink flink-streaming

【解决方案1】：

如果您使用RichMapFunction，您可以在其open 方法中实例化侦听器；那么你就不用担心序列化了。

不过，你应该小心，不要在 Flink 函数中阻塞 i/o。最好使用 Flink 的async i/o operator。对于查询数据库，您最好以table lookup source 的身份访问外部数据库或将其数据作为 CDC 源流式传输。

【讨论】：

我正在使用带有 flink 的 spring-boot，我无法在 open 方法中获取 TestRequestListener bean。当我在我的类中自动装配应用程序上下文并将其发送到这里时，它会抛出我的类不是可序列化的异常，或者它变为 null。