开发者社区> 问答> 正文

Flink:处理Flink工作中的异常的最佳方法是什么

flink小助手 2018-12-10 11:27:52 776

我有一个flink工作,接受Kafaka主题,并通过一堆运营商。我想知道处理中间发生的异常的最佳方法是什么。

我的目标是有一个集中的地方来处理可能从不同运营商抛出的异常,这是我目前的解决方案:

使用ProcessFunction和输出sideOutput,以context在catch块中,假设有一个例外,并有一个独立的装置功能为sideOutput在结束的地方调用外部服务来更新另一个相关作业的状态

但是,我的问题是,通过这样做,似乎我仍然需要调用collector.collect()并传入一个空值,以便继续跟随运算符并点击最后一个阶段sideOutput将流入单独的接收器函数。这是正确的方法吗?

此外,我不确定如果我不在collector.collect()操作员内部调用会发生什么,它会挂起并导致内存泄漏吗?

分享到
取消 提交回答
全部回答(1)
  • flink小助手
    2019-07-17 23:19:08

    最好不调用collector.collect()。collect()当您使用侧输出捕获异常时,您不需要使用空值调用- 每个运算符都可以有自己的侧输出。最后,如果您有多个这样的运算符,其侧面输出用于异常,则可以union()在将该流发送到接收器之前将侧输出放在一起。

    如果由于某种原因下游运营商需要知道存在异常,那么一种方法是输出一个Either,但是每个下游运营商当然需要有代码来检查它接收的内容。

    0 0