无法在Python中的单个数据流作业中动态加载多个流管道(N到N管道)(使用运行时值提供程序

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class DynamicPipeline(beam.DoFn):
    def process(self, element):
        # 假设element包含了管道类型的信息
        pipeline_type = element['type']
        
        if pipeline_type == 'typeA':
            yield from self.create_pipeline_A(element)
        elif pipeline_type == 'typeB':
            yield from self.create_pipeline_B(element)
        # 可以继续添加更多的类型

    def create_pipeline_A(self, element):
        # 创建并处理管道A的逻辑
        return [element['data'] * 2]

    def create_pipeline_B(self, element):
        # 创建并处理管道B的逻辑
        return [element['data'] + 10]

def run():
    options = PipelineOptions()
    p = beam.Pipeline(options=options)

    (p
     | 'ReadInput' >> beam.io.ReadFromText('input.json')
     | 'ParseJSON' >> beam.Map(lambda line: json.loads(line))
     | 'DynamicPipeline' >> beam.ParDo(DynamicPipeline())
     | 'WriteOutput' >> beam.io.WriteToText('output'))

    result = p.run()
    result.wait_until_finish()

if __name__ == '__main__':
    run()