使用Apache Beam从PubSub读取时出现KeyError: 'ref_PCollection_PCollection_6'

错误原因分析 #

KeyError: 'ref_PCollection_PCollection_6' 是Interactive Beam运行时的内部引用错误，通常由以下场景触发：

• Apache Beam及Interactive扩展包版本不兼容（Colab预装版本可能滞后）
• 流式管道中窗口操作与ib.show()的参数交互存在冲突
• PCollection的链式调用写法导致Runner无法正确追踪数据节点

解决方案 #

1. 升级依赖包到兼容版本 #

Colab预装的Apache Beam版本可能存在兼容性问题，先升级到支持交互式流式处理的最新稳定版：

!pip install --upgrade apache-beam[interactive,google]

2. 完善管道构造与认证配置 #

• 补充Google Cloud认证步骤，确保有权限访问PubSub资源
• 明确配置项目ID，避免隐式参数缺失
• 拆分管道步骤并添加清晰标签，帮助Runner正确识别PCollection节点

修改后的完整代码示例：

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
from google.colab import auth

# 完成Google Cloud身份认证
auth.authenticate_user()

# 配置交互式运行参数
ib.options.recording_duration = '10m'
ib.options.recording_size_limit = 1e9

# 初始化管道选项
options = pipeline_options.PipelineOptions()
options.view_as(pipeline_options.StandardOptions).streaming = True

# 配置Google Cloud项目ID（替换为你的项目ID）
google_options = options.view_as(GoogleCloudOptions)
google_options.project = "YOUR_PROJECT_ID"

with beam.Pipeline(InteractiveRunner(), options=options) as p:
    # 拆分管道步骤，每个步骤添加明确标签
    raw_messages = p | "读取PubSub消息" >> beam.io.ReadFromPubSub(topic="projects/YOUR_PROJECT_ID/topics/Test")
    windowed_data = raw_messages | "10秒固定窗口" >> beam.WindowInto(beam.window.FixedWindows(10))
    element_counts = windowed_data | "按元素计数" >> beam.combiners.Count.PerElement()

# 先测试基础展示，确认正常后再添加窗口信息
ib.show(element_counts)
# 若需窗口信息，确保版本兼容后启用：ib.show(element_counts, include_window_info=True)

3. 逐步排查问题 #

• 先移除窗口操作和计数，仅测试PubSub读取功能，确认能正常获取数据
• 确认基础功能正常后，再逐步添加窗口、计数等变换，定位触发错误的节点
• 若include_window_info=True仍导致错误，暂时禁用该参数，待版本更新后再尝试

内容的提问来源于stack exchange，提问作者Amarjeet