使用200万行文件作为侧输入时，Apache Beam未向BigQuery写入数据

问题场景 #

使用Apache Beam Python SDK 2.48.0构建数据流：

• 有界数据源：通过beam.io.ReadFromText读取35MB、200万行的json.gz格式文件；
• 无界数据源：通过beam.io.ReadFromPubSub读取，配置会话窗口（3600秒）+ 按计数触发（每1条触发）+ 丢弃模式累积。

将有界数据作为Side Input注入LeftJoiner ParDo与无界数据关联，小样本文件（35KB）可正常运行，但大文件关联时LeftJoiner从未执行，数据无法写入BigQuery；反转流程（无界数据作为Side Input）时，大文件场景下关联函数无限阻塞。

核心关联代码：

enriched_unbounded_pcol = (
    unbounded_pcol
    | beam.ParDo(PreprocessUnbounded()) # 返回(v1, v2)元组
    | beam.ParDo(LeftJoiner(), side_input_file=beam.pvalue.AsDict(side_input_file)))

LeftJoiner实现：

class LeftJoiner(beam.DoFn):
    def process(self, element, **kwargs):
        v1, v2 = element
        side_input_dict = kwargs["side_input_file"]
        v3 = side_input_dict.get(v1)
        yield compute_stuff(v3)

核心原因 #

1. 大Side Input内存过载：200万行数据构建的字典会占用大量内存（保守估计1-2GB），超出Dataflow默认Worker的内存配额，导致Worker加载Side Input时内存溢出或超时，ParDo因依赖未就绪无法执行。
2. 无界流触发依赖Side Input就绪：无界流的触发逻辑需要等待Side Input完全分发到Worker节点，若Side Input加载失败，无界元素会被阻塞在ParDo之前，表现为ParDo无执行日志。
3. 无界数据作为Side Input的设计缺陷：无界数据流的水印默认是Timestamp.MAX_VALUE，Beam会持续等待水印推进以确定Side Input的完整数据，导致依赖它的有界流ParDo永久阻塞。

解决方案 #

1. 分片Side Input减少单节点内存压力 #

避免直接加载全量字典，将有界数据按关联键分片后分发，让每个Worker仅加载对应分片的数据：

# 预处理有界数据，按v1分组
side_input_sharded = (
    side_input_file
    | beam.ParDo(ParseSideInput())  # 解析为(v1, v3)键值对
    | beam.GroupByKey()
)

# 关联时使用AsIter获取对应分片数据
enriched_unbounded_pcol = (
    unbounded_pcol
    | beam.ParDo(PreprocessUnbounded())
    | beam.ParDo(LeftJoiner(), side_input=beam.pvalue.AsIter(side_input_sharded))
)

# 适配分片逻辑的LeftJoiner
class LeftJoiner(beam.DoFn):
    def process(self, element, **kwargs):
        v1, v2 = element
        # 获取当前v1对应的分片迭代器
        side_input_iter = kwargs["side_input"]
        v3 = next(side_input_iter, None)
        yield compute_stuff(v3)

2. 调整Worker资源配置 #

• 提升Worker内存：提交作业时指定高内存实例，例如：

  --worker_machine_type=n1-standard-8
  

  或直接指定内存配额：

  --worker_memory_gb=8
  

• 调整Side Input缓存：增大缓存阈值，减少分发压力：

  --side_input_cache_size=500  # 单位：MB
  

3. 分布式缓存预加载大文件 #

将大文件上传至GCS，通过Dataflow分布式缓存让Worker提前加载到本地磁盘，再在ParDo中构建字典：

class LeftJoiner(beam.DoFn):
    def setup(self):
        # Worker启动时加载本地缓存的文件
        self.side_input_dict = {}
        import gzip, json
        with gzip.open('/tmp/side_input.json.gz', 'rt') as f:
            for line in f:
                data = json.loads(line)
                self.side_input_dict[data['v1']] = data['v3']

    def process(self, element):
        v1, v2 = element
        v3 = self.side_input_dict.get(v1)
        yield compute_stuff(v3)

提交作业时指定分布式缓存文件：

python your_pipeline.py \
    --runner=DataflowRunner \
    --project=your-gcp-project \
    --temp_location=gs://your-bucket/temp \
    --staging_location=gs://your-bucket/staging \
    --files_to_stage=gs://your-bucket/side_input.json.gz

4. 禁止无界数据作为Side Input #

无界数据的水印特性导致其无法作为可靠的Side Input，此场景下必须将有界数据作为关联基准，无界数据作为主数据流。

验证步骤 #

• 查看Worker stderr日志：在GCP控制台Dataflow作业详情中，检查是否存在内存溢出、Side Input加载超时等错误；
• 监控Worker内存使用率：通过Dataflow监控面板确认内存是否接近阈值；
• 单独测试大文件加载：编写小Pipeline仅读取并统计大文件行数，验证文件本身无损坏且加载时间合理。

内容的提问来源于stack exchange，提问作者Grégore Borel