通过相关管道处理DataFlow/Apache Beam中的拒绝

Question

我有一个从BigQuery获取数据并将其写入GCS的管道，但是，如果我发现任何拒绝，我希望将它们正确地添加到一个BigQuery表中。我将拒绝收集到一个全局列表变量中，然后将列表加载到BigQuery表中。当我在本地运行它时，这个过程工作得很好，因为管道以正确的顺序运行。当我使用dataflowrunner运行它时，它不能保证顺序（我希望pipeline1在Pipeline2之前运行。有没有一种方法可以使用Python在Dataflow中使用依赖的管道？或者也请建议是否可以用更好的方法解决这个问题。提前谢谢。

with beam.Pipeline(options=PipelineOptions(pipeline_args)) as pipeline1:
 
    data = (pipeline1
               | 'get data' >> beam.io.Read(beam.io.BigQuerySource(query=...,use_standard_sql=True))
               | 'combine output to list' >> beam.combiners.ToList()
               | 'tranform' >> beam.Map(lambda x: somefunction)  # Collecting rejects in the except block of this method to a global list variable
               ....etc
               | 'to gcs' >> beam.io.WriteToText(output)
               )

# Loading the rejects gathered in the above pipeline to Biquery
with beam.Pipeline(options=PipelineOptions(pipeline_args)) as pipeline2:
    rejects = (pipeline2
                    | 'create pipeline' >> beam.Create(reject_list)
                    | 'to json format' >> beam.Map(lambda data: {.....})
                    | 'to bq' >> beam.io.WriteToBigQuery(....)
                    )

Answer 1

您可以这样做，但只需要1个管道和转换中的一些附加代码。

beam.map(lambda x:somefunction)应该有两个输出：写入GCS的输出，以及最终写入BigQuery的被拒绝的元素。

为此，转换函数必须返回taggedoutput。

with beam.Pipeline(options=PipelineOptions(pipeline_args)) as pipeline1:
 
    data = (pipeline1
               | 'get data' >> beam.io.Read(beam.io.BigQuerySource(query=...,use_standard_sql=True))
               | 'combine output to list' >> beam.combiners.ToList()
               | 'tranform' >> beam.Map(transform)  # Tagged output produced here

    pcoll_to_gcs = data.gcs_output
    pcoll_to_bq  = data.rejected

    pcoll_to_gcs | "to gcs" >> beam.io.WriteToText(output)
    pcoll_to_bq  | "to bq" >> beam.io.WriteToBigQuery(....)

那么transform函数将如下所示

def transform(element):
  if something_is_wrong_with_element:
    yield pvalue.TaggedOutput('rejected', element)

  transformed_element = ....

  yield pvalue.TaggedOutput('gcs_output', transformed_element)