Python Apache Beam使用AsDict侧输入报不可下标错误如何解决？

beam.pvalue.AsDict()是Beam用于标记侧输入的特殊包装类，仅用于告知运行时需要将对应PCollection作为字典类型的侧输入注入，在流水线构建阶段不会生成实际字典数据。
你当前的错误是因为在DoFn的__init__构造方法中就尝试操作AsDict包装对象，而实际的字典值要等到DoFn的process方法执行时，才会由Beam运行时解析注入，构造方法执行时侧输入数据还未生成，自然无法按下标访问。

1. 调整自定义DoFn的定义 #

将侧输入参数从__init__移到process方法中接收，此时拿到的参数就是解析完成的普通字典：

class Locations(beam.DoFn):  # Location_ID
    def process(self, element: str, operations: dict):
        # 此时operations已经是实际可操作的字典
        locations: list = operations["lookup_location_id"]
        # 此处继续写原有的业务代码

2. 调整ParDo调用逻辑 #

侧输入不要传入DoFn的构造方法，改为在beam.ParDo的参数中传入，参数名要和process方法中定义的侧输入参数名保持一致：

locations = (
            csv_data
            | beam.ParDo(Locations(), operations=beam.pvalue.AsDict(operations))
            | "Dedup locations" >> beam.Distinct()
        )

补充说明 #

你现有的operations PCollection已经经过GroupByKey处理，是标准的KV结构PCollection，完全符合AsDict的格式要求，无需调整生成逻辑。如果后续需要新增多个侧输入，只需按照相同规则在ParDo中传入对应AsList/AsDict/AsSingleton包装的PCollection，process方法新增对应同名参数即可直接使用。

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