从 Apache Beam 读取 PubSub 时，如何避免 KeyError 的困扰？

2024-03-14 11:25:58

从 Apache Beam 中读取 PubSub 时排除 KeyError 的全面指南

作为一名经验丰富的程序员，我经常使用 Apache Beam 来创建强大的流处理管道。最近，我在从 PubSub 读取数据时遇到了一个恼人的 KeyError，错误信息为：“ref_PCollection_PCollection_6”。经过一番深入的调查，我发现了导致此错误的潜在原因并制定了有效的解决方案。

KeyError 的原因

此错误通常是由以下原因之一导致：

管道配置错误： 管道中的组件可能未正确配置，导致键引用无效。
数据不一致： 读取的 PubSub 消息可能不符合预期的模式或格式，导致 Beam 无法正确处理数据。

解决 KeyError 的步骤

1. 检查管道配置

仔细检查 beam.io.ReadFromPubSub 转换的配置，确保主题名称正确且窗口设置适当。此外，验证管道中的其他组件是否已正确连接，并且键引用有效。

2. 验证数据

检查从 PubSub 读取的实际数据，确保其符合预期的模式和格式。考虑使用 Beam 的 JsonCoder 或 PicklerCoder 来解码消息，以更好地控制数据结构。

3. 使用 Beam Shell 调试

Beam Shell 是一种交互式工具，可帮助调试 Beam 流处理管道。你可以使用 Beam Shell 来检查数据管道和数据本身。有关如何使用 Beam Shell 的更多信息，请参阅 Beam 文档。

4. 升级 Beam 版本

有时，Beam 的早期版本可能会遇到此类错误。尝试将 Beam 升级到最新版本以解决潜在问题。

代码示例

以下是一个经过修改的代码示例，它解决了之前代码中的潜在问题：

import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib
from apache_beam.options import pipeline_options
from apache_beam.options.pipeline_options import GoogleCloudOptions
import google.auth

ib.options.recording_duration = '10m'
ib.options.recording_size_limit = 1e9

options = pipeline_options.PipelineOptions()

options.view_as(pipeline_options.StandardOptions).streaming = True

with beam.Pipeline(InteractiveRunner(), options=options) as p:
  words = p|"read" >> beam.io.ReadFromPubSub(topic="projects/PROJECT_ID/topics/Test")| beam.WindowInto(beam.window.FixedWindows(10, 0)) | "count" >> beam.combiners.Count.PerElement() 

ib.show(words, include_window_info=True)