如何过滤坏和好的 json 事件，然后增加坏 json 记录的指标计数，并使用 java [关闭] 将这些记录存储在 apache Beam 中

Posted 2023-02-16

【中文标题】如何过滤坏和好的 json 事件，然后增加坏 json 记录的指标计数，并使用 java [关闭] 将这些记录存储在 apache Beam 中

【英文标题】：How to filter bad and good json events and then increment metrics count for bad json record and store those record in apache beam using java [closed]

【发布时间】：2021-05-22 17:25:47

【问题描述】：

我有包含 json 原始消息事件的 PubSub 主题，我想过滤好的 json 记录/事件和坏的 json 记录/事件并存储在不同的 PCollection 中。对于每个不良记录计数器指标，应增加并将日志存储在另一个 PCollection 中，以便稍后我可以检查日志中是否存在不良 json 记录。我需要使用哪种 Apache Beam 变换以及如何使用 Java 使用这些变换。

【参考方案1】：

您可以阅读beam programming guide。你会为你的用例找到很好的解决方案和模式。例如，要过滤好 JSON 和坏 JSON，您需要创建一个转换，其中包含标准输出（假设是正确的 JSON）和坏 JSON 的an addition output。

因此，从那里，您有 2 个 PCollections。然后独立处理它们。您可以在文件中、BigQuery 中接收错误的 JSON，或者只需创建一个转换，在 Cloud Logging 中写入一个特殊的日志跟踪，以便稍后在另一个进程中获取和处理此日志跟踪。