Apache Beam未从订阅中删除无效元素

Question

刚意识到我的管道是错误的，当涉及到错误事件时，它们会不断被处理，并且从未从订阅中删除。

基本上，我有一个简单的管道，其中包含一个触发器，可以将这些事件提取到文件中。

在其中一个阶段中，它会处理通过PubSub接收到的消息的有效负载，并将其转发到下一个阶段。然而，在某些情况下，这将失败。

        pipeline
        .apply("Read PubSub Events",
            PubsubIO.readMessagesWithAttributes().fromSubscription(options.getSubscription()))
        .apply("Map to MyClass",
            ParDo.of(new PubSubMessageToMyClass())) // Exception thrown in this stage.
        .apply("Apply Timestamps", WithTimestamps.of(new SetTimestampFn()).withAllowedTimestampSkew(new Duration(Long.MAX_VALUE)))
        ...
        );

当错误发生时，我会在管道中一遍又一遍地看到同一事件，就像它永远不会结束处理一样。

是否有任何方法可以明确告诉Apache Beam使给定消息无效并防止进一步失败的处理？

Answer 1

数据流处理任意捆绑包中的元素，并在该捆绑包中的任何元素出现错误时重试整个捆绑包。在批处理模式下运行时，包含失败项的捆绑包将重试4次。当单个捆绑包发生4次故障时，管道将完全失效。在流模式下运行时，包含失败项目的捆绑包将无限期重试，这可能会导致管道永久暂停。

考虑通过添加异常处理程序来防止代码中的错误。例如，如果您想删除在ParDo中完成的某些自定义输入验证失败的元素，请在ParDo中使用try/catch块来处理异常并删除元素。