基于标记输出数的动态分叉梁（数据流）管道

Question

我目前正试图根据数据中包含的特定键，将运行在Google Dataflow上的Beam管道分叉到多个目的地。当使用TaggedOutput标记对“fork”的每个endpoint进行硬编码时，我能够实现这一点。但是，在将来，我并不总是知道底层数据中存在哪些键，因此我希望使用类似于以下的for循环动态创建流程中的后续步骤：

p = beam.Pipeline(options=pipeline_options)

pipe = p | 'ReadFromGCS' >> ReadFromText(args['input']) \
         | 'TagAllLines' >> beam.ParDo(produce_tagged_output_keys).with_outputs()

for client in pipe:
  client = pipe.client | client+'MapLinesToDicts' >> beam.Map(lambda line: dict(record=line)) \
                         | client+'WriteToBQTable' >> WriteToBigQuery(client+'_test', dataset=bq_dataset, project=project_id, schema='record:string')

我的理解是，<代码>的结果。with\u outputs（）应该是可编辑的，不是吗？当我运行上述命令时，它会执行管道，而不会出现问题，但会完全忽略for循环。有没有一种方法可以动态地做到这一点，而我却没有？

Answer 1

的结果。with\u outputs（）仅在指定标记时才可使用

的结果。不指定标记的with\u outputs（）是不可使用的，因为在管道构建时输出还未知
的结果。使用\u输出（tag1，tag2，main=tag0）但是可以：
在此处测试差异

请记住，使用apache束，您首先构建管道，然后让数据流过它。因此，您不能让您的管道架构在运行时依赖于数据。

不过，我会提出另一种解决方案：

假设您有一个pcollection集合，其元素的结构如下：

(key, bq_row)

其中，键是决定要写入哪个表的功能，以及要写入bigquery行的数据的bq\u行。

然后，您可以按键对元素进行分组：

grouped = collection | beam.GroupByKey()

现在，每个不同的键都有一个元素。因此，一个元素中的所有bq_row都应该写入同一个大查询表，这取决于元素键。

然后可以定义一个DoFn，它将每个元素中的所有行写入相应的表。（示例代码）

class WriteToBigQueryFn(beam.DoFn):

    def __init__(self, dataset_name):
        super(BigQueryWriter, self).__init__()
        self.client = bigquery.Client()
        self.dataset = client.dataset(dataset_name)

    def process(self, (key, data)):
        table_name = get_table_name(key) # get table name based on key
        table = self.dataset.table(table_name)
        table.reload(self.client)

        table.insert_data(data)

grouped | beam.ParDo(WriteToBigQueryFn)