apache-flink - apache flink - 错误处理的正确方法

Question

我想知道 Flink 中是否有内置错误处理的选项。
可能有2种情况:

来自 Kafka 的当前消息(在我的情况下)无效，继续下一个

未捕获的异常 - 从我所见，它可以完全停止流聚合。

我可以处理这两种情况吗？ (Java代码)

Answer 1

1)这是使用 flatMap 惯用的方式完成的:如果您的消息有效，则继续使用包含有效元素的列表(可能已在同一步骤中处理)。如果它无效，您只需返回一个空列表，以便该步骤不会产生任何元素。我可以提供 Scala 代码，但我不熟悉 Java API，所以我不想让您偏离轨道。只需检查 flatMap 调用。

2)这取决于异常的类型:如果它是由您自己的代码引起的，只需捕获它并在运算符内部处理它，或者简单地记录它并继续。没有关于特定案例的任何进一步信息，这是我所知道的最好的，但同样，来自 Scala 我没有遇到运行时异常。