如何在Apache Beam中通过键在静态查找表上以流模式连接PCollection (Python)

lookup_table = {
    'key1': 'value1',
    'key2': 'value2',
    'key3': 'value3'
}

import apache_beam as beam

def lookup_value(element, lookup_table):
    key = element['key']  # 获取PCollection中的键
    value = lookup_table.get(key)  # 在查找表中查找对应的值
    return {'key': key, 'value': value}

with beam.Pipeline() as pipeline:
    lookup_table_pcoll = pipeline | 'Create lookup table' >> beam.Create([lookup_table])
    input_pcoll = pipeline | 'Create input PCollection' >> beam.Create([
        {'key': 'key1'},
        {'key': 'key2'},
        {'key': 'key3'}
    ])
    
    # 使用beam.SideInput将查找表作为附加输入传递给beam.Map
    output_pcoll = input_pcoll | 'Lookup values' >> beam.Map(lookup_value, lookup_table=beam.pvalue.AsDict(lookup_table_pcoll))
    
    output_pcoll | 'Print output' >> beam.Map(print)