Apache Beam及Google Dataflow使用全局变量处理大数据集空字典报错问题

Apache Beam/Dataflow 是分布式执行引擎，你定义的全局字典仅在单个Worker进程的内存中生效，无法跨不同Worker、不同机器共享：

• 小数据量测试时，Dataflow 不会启动多Worker，整个流水线的所有步骤都在同一个进程内执行，全局字典可以正常读写
• 大数据量运行时，Dataflow 会自动扩容多个Worker，并且会将流水线的不同步骤调度到不同的Worker/进程执行，你在compute步骤写入的全局字典仅存在于执行compute步骤的Worker中，执行inject步骤的Worker看不到这份数据，自然会出现空字典、KeyError的问题

不要使用全局变量存共享映射，改用Beam原生支持的分布式数据传递能力，以下两种方案按需选择：

方案1：侧输入（Side Input，适合映射数据量≤1G的场景） #

先把你需要的(trans_number, seq)到(ordertxt, upref)的映射预先生成，作为侧输入传入后续的注入步骤，Beam会自动将这份映射同步到所有需要用到的Worker节点。

代码修改示例： #

# 第一步：预生成pre_compute映射的侧输入
pre_compute_side_input = (
    left_join_std_dtdo_raw # 替换为你做compute之前的原始PCollection
    | "提取映射KV" >> beam.Map(lambda x: (
        (x["transaction_number"], x["stdetail_seq"]), 
        (x.get("dorder_ordertxt",""), x.get("dorder_upref",""))
    ))
    | "按Key聚合值" >> beam.CombinePerKey(
        lambda old_val, new_val: (
            new_val[0] if new_val[0] else old_val[0],
            new_val[1] if new_val[1] else old_val[1]
        )
    )
    | "转成字典侧输入" >> beam.combiners.ToDict()
)

# 第二步：处理原始数据流，移除ordertxt和upref字段，保留原有处理逻辑
left_join_std_dtdo = (
    left_join_std_dtdo_raw
    | '移除ordertxt/upref字段' >> beam.Map(lambda x: {k:v for k,v in x.items() if k not in ["dorder_ordertxt", "dorder_upref"]})
    | 'UPDATE PRICE FOR SCCRM01' >> beam.ParDo(update_price_sccrm01())
    | 'REMOVE PRICE from DICTIONARY' >> beam.ParDo(remove_dtdo_price())
)

# 第三步：修改注入步骤为带侧输入的DoFn，不用全局变量
class InjectUprefOrdertxt(beam.DoFn):
    def process(self, element, pre_compute):
        # 原有字符串转字典的逻辑不变
        data = element.strip("\n").replace("'", '"').replace("None", "null")
        data = json.loads(data)
        key = (data.get('transaction_number'), data.get('stdetail_seq'))
        ordertxt, upref = pre_compute[key]
        data['dorder_ordertxt'] = ordertxt
        data['dorder_upref'] = upref
        yield data

# 第四步：流水线注入步骤传入侧输入
rm_left_std_dtdo = (
    left_join_std_dtdo 
    | 'CHANGE JOINED STD DTDO INTO STR' >> beam.Map(lambda x: str(x))
    | 'DISTINCT STD DTDO' >> beam.Distinct()
    | 'EVALUATE AND INJECT AS DICT STD DTDO' >> beam.ParDo(
        InjectUprefOrdertxt(), 
        pre_compute=beam.pvalue.AsSingleton(pre_compute_side_input)
    )
    | 'Adjust STD_NET_PRICE WITH DODT_PRICE' >> beam.ParDo(replaceprice())
)

方案2：关联Join（适合映射数据量很大的场景） #

如果你的映射数据量超过1G，侧输入会占用过多Worker内存，改为直接将预聚合的映射PCollection和去重后的数据流按照(trans_number, seq)做关联Join，补全字段即可，不需要把全量映射加载到单节点内存。

不要在Beam流水线的处理函数中使用全局变量存储业务数据，分布式环境下全局变量的可见性完全无法保证，小数据量的正常运行只是巧合。

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