带Side Input的Stateful DoFn在DirectRunner下偶发报错问询

问题背景 #

你实现的有状态DoFn使用了beam.pvalue.AsSingleton侧输入，主输入明确为键值对，但在DirectRunner下测试时偶发触发以下错误：

ValueError: Input elements to the transform <ParDo(PTransform) label=[ParDo(StatefulWithSideInput)] side_inputs=[AsSingleton(PCollection[side/Map(decode).None])]> with stateful DoFn must be key-value pairs.

代码多数时候能正常输出预期结果，且在Dataflow上运行完全正常，仅DirectRunner下偶现此问题。

问题本质 #

这是DirectRunner的偶发Bug，和你的代码写法无关：

• 有状态DoFn仅要求主输入为键值对，侧输入无需满足此要求，你的代码完全符合规范，Dataflow的正常运行也验证了这一点。
• 报错源于DirectRunner在Pipeline转换为Runner API的阶段，内部逻辑偶发错误地将侧输入纳入了"是否为键值对"的检查范围，而非仅检查主输入。
• 随机性是因为Pipeline转换阶段的内部处理顺序存在偶发异常，导致检查逻辑误判输入来源。

临时解决方案 #

若需在DirectRunner下稳定完成测试，可尝试以下方案：

1. 给侧输入临时添加虚拟键：
   修改侧输入的生成逻辑，先给元素加上虚拟键，再提取value作为Singleton输入：

   side_input = (
     pipeline
     | "side" >> beam.Create(['test'])
     | beam.Map(lambda x: ('dummy_key', x))  # 添加虚拟键
     | beam.Map(lambda x: x[1])  # 提取实际需要的value
   )
   

2. 升级Apache Beam版本：该问题在Beam 2.30及以上版本中已被官方修复。
3. 更换测试用Runner：使用FlinkRunner或DataflowRunner的本地模式替代DirectRunner进行测试，可避免此问题。

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