如何处理apache beam管道中的异常？

Question

 PipelineResult.State state = pipeline.run().waitUntilFinish();
 if(state == PipelineResult.State.FAILED){
            System.out.println("failed");
        }else if(state==PipelineResult.State.DONE){

           

            System.out.println("Done");

        }

Answer 1

我建议研究如何在管道的每一步处理错误，而不是试图处理管道失败。

例如，WriteToBigQuery接收器的默认重试策略是always，因此即使错误的格式不是可以成功写入BigQuery的格式，也将始终重试。您可以考虑实现死信模式来捕获错误并将它们写入不同的源。

我只使用Kafka和PubSub作为管道的输入，所以我不确定从文件读取时如何处理错误，以及您可能会遇到什么问题。

在读取和写入数据之间创建审核或任何其他处理步骤时，您可以使用标记输出，并尝试捕获从一个步骤中为成功和失败的输出创建两个不同的输出，然后将它们写入任何您喜欢的地方