error-handling - Apache Flink - "keyBy"中的异常处理

Question

进入 Flink job 的数据可能会因为代码中的错误或缺乏验证而触发异常。我的目标是提供一致的异常处理方式，我们的团队可以在 Flink 作业中使用这些方式，而不会导致生产中断。

1. 重启策略似乎不适用于这里:

   • 简单的重启不会解决问题，我们会陷入重启循环
   • 我们不能简单地跳过事件
   • 它们对 OOME 或一些暂时性问题有好处
   • 我们不能添加自定义的

2. “keyBy”函数中的 try/catch block 不能完全帮助:

   • 处理异常后无法跳过“keyBy”中的事件

示例代码:

env.addSource(kafkaConsumer)
    .keyBy(keySelector) // must return one result for one entry
    .flatMap(mapFunction) // we can skip some entries here in case of errors
    .addSink(new PrintSinkFunction<>());
env.execute("Flink Application");

我希望能够跳过在“keyBy”中导致问题的事件处理以及应该只返回一个结果的类似方法。

Answer 1

除了@phanhuy152 的建议(这对我来说似乎完全合法)为什么不在 keyBy 之前使用 filter？

env.addSource(kafkaConsumer)
    .filter(invalidKeys)
    .keyBy(keySelector) // must return one result for one entry
    .flatMap(mapFunction) // we can skip some entries here in case of errors
    .addSink(new PrintSinkFunction<>());
env.execute("Flink Application");