如何在 Apache flink KeyedBroadCastProcessFunction 中处理异常

Question

我是 Flink 的新手，我正在使用 Flink KeyedBroadCastProcessFunction 进行模式评估，与 (https://flink.apache.org/2019/06/26/broadcast-state.html) 类似，我正在使用 JAVA 开发我的代码，但我不知道如果出现任何故障，我该如何处理异常在处理数据流时发生，我搜索了很多但没有得到我最终在以下两个链接中

Flink: what's the best way to handle exceptions inside Flink jobs

Apache Flink - exception handling in "keyBy"

根据第一个链接，用户说他在 processfn 中使用 sideoutput 来捕获错误，我还在我的程序中使用 sideoutput 发送与模式不匹配的数据，但我不知道如何处理错误和无效数据到同侧输出

根据第二个链接，用户正在尝试使用 null 键和 printsink 函数向 keyby 函数添加接收器，我根本不理解

谁能帮我解决以下问题

1) 任何用于异常处理的文档或小代码 sn-p 我在 flink 文档站点中没有找到任何东西 2）flink异常处理的最佳实践

如果有人可以回答，我在网上没有找到任何有效资源，这对于进一步参考其他人也很有用

Answer 1

您可以从 ProcessFunction 获得任意数量的辅助输出 - 每个输出都有自己独特的 OutputTag。因此，您可以将一个用于不匹配的数据，另一个用于错误。

final OutputTag<T> unmatched = new OutputTag<String>("unmatched-data"){};
final OutputTag<String> errors = new OutputTag<String>("side-output-for-errors"){};

SingleOutputStreamOperator<T> matchedData = ...;

DataStream<T> unmatched = matchedData.getSideOutput(unmatched);
DataStream<String> errors = matchedData.getSideOutput(errors);

如果您最终得到几个不同的操作员，每个操作员都使用侧输出来收集错误，那么您可以将它们合并在一起以进行报告，这可能看起来像这样：

final OutputTag<String> errors = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<T> task1 = ...;
SingleOutputStreamOperator<T> task2 = ...;
SingleOutputStreamOperator<T> task3 = ...;

DataStream<String> exceptions1 = task1.getSideOutput(errors);
DataStream<String> exceptions2 = task2.getSideOutput(errors);
DataStream<String> exceptions3 = task3.getSideOutput(errors);

DataStream<String> exceptions = exceptions1.union(exceptions2, exceptions3);

exceptions.addSink(new FlinkKafkaProducer(...));