单模板文件调度多Pipeline为独立Dataflow作业的模板覆盖问题问询

你遇到的模板覆盖问题核心来自两个错误实现：

• 所有并行Pipeline共用同一个options实例，多线程修改同一个实例的job_name、临时路径等属性存在线程安全问题，你加的100秒sleep只是临时规避方案，没有解决根本冲突
• 生成Pipeline模板时未给每个Pipeline指定独立的模板输出路径，所有Pipeline都往同一个路径写模板文件，自然会出现后生成的覆盖先生成的情况

1. 每个Pipeline使用独立的options实例 #

不要在run方法里全局初始化self.options，改为在run_pipeline函数内部为每个Pipeline单独初始化options，彻底解决多线程属性冲突问题，同时可以去掉所有sleep逻辑。

2. 给每个Pipeline指定独立的模板输出路径 #

在GoogleCloudOptions中新增template_location配置，按table_type区分路径，避免模板文件重名覆盖：

google_cloud_options.view_as(GoogleCloudOptions).template_location = f'gs://dataflow_storage_bq_bt/dataflow_test/templates/{table_type}'

3. 补全写入BigTable的计数指标 #

你现有代码中没有定义Written Row对应的计数器，会导致行数校验时经常取到None值，修改GetRowFn增加计数：

class GetRowFn(beam.DoFn):
    def __init__(self):
        self.bt_write_count = Metrics.counter('BigTable', 'Written Row')
    def process(self, row_list):
        for row in row_list:
            self.bt_write_count.inc()
            yield row

def run_pipeline(self, query_text, schema, table_type):
    # 每个Pipeline单独初始化options，不要复用全局实例
    options = PipelineOptions()
    google_cloud_options = options.view_as(GoogleCloudOptions)
    # 加时分秒避免同类型任务短时间内启动重名
    job_name = table_type.replace('_','') + datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    google_cloud_options.job_name = str(job_name)
    google_cloud_options.temp_location = f'gs://dataflow_storage_bq_bt/dataflow_test/tmp/{table_type}'
    google_cloud_options.staging_location = f'gs://dataflow_storage_bq_bt/dataflow_test/staging/{table_type}'
    # 独立的模板路径，解决覆盖问题
    google_cloud_options.template_location = f'gs://dataflow_storage_bq_bt/dataflow_test/templates/{table_type}_template'
    # 其他自定义options配置也在这里单独设置
    options.view_as(DailyUploadOptions).test = query_text
    
    with beam.Pipeline(options=options) as pipeline:
        data_collection = pipeline | f"Get {table_type} from BigQuery" >> beam.io.Read(ReadFromBigQuery(query=query_text.get(),
                                                                                                    use_standard_sql=True))

        data_collection \
        | f"Get  {table_type} list of direct_row's " >> beam.ParDo(CreateRowFn(schema)) \
        | f"Get  {table_type} single direct row" >> beam.ParDo(GetRowFn()) \
        | f"Write  {table_type} To BT" >> WriteToBigTable(project_id=self.config_data["gcp_config"]["bt_project"],
                                                        instance_id=self.config_data["gcp_config"]["bt_instance_id"],
                                                        table_id=self.config_data["gcp_config"]["bt_table_id"])
    
    bigquery_count = self.getrow_count(pipeline,'bigquery_row') 
    bigtable_count = self.getrow_count(pipeline,'Written Row')
    if bigquery_count is None or bigtable_count is None:
        print(f"No daily upload data for {table_type}")  
    elif bigquery_count == bigtable_count:
        print(f"All daily upload data for {table_type} moved to bigtable from bigquery")
    else:
        raise ValueError(f"Row count mismatch; check the pipeline for {table_type}")

def run(self):
    sql_config = self.config_data["sql_config"]
    # 把任务参数整理成列表，用循环启动线程，避免重复代码
    task_list = [
        (self.options.view_as(DailyUploadOptions).test1, sql_config['test1_config']['schema'], 'feature1'),
        (self.options.view_as(DailyUploadOptions).test2, sql_config['test2_config']['schema'], 'feature2'),
        (self.options.view_as(DailyUploadOptions).test3, sql_config['test3_config']['schema'], 'feature3'),
        (self.options.view_as(DailyUploadOptions).test4, sql_config['test4_config']['schema'], 'feature4'),
        (self.options.view_as(DailyUploadOptions).test5, sql_config['test5_config']['schema'], 'feature5'),
        (self.options.view_as(DailyUploadOptions).test6, sql_config['test6_config']['schema'], 'feature6'),
        (self.options.view_as(DailyUploadOptions).test7, sql_config['test7_config']['schema'], 'feature7'),
    ]
    for query, schema, table_type in task_list:
        Thread(target=self.run_pipeline, args=(query, schema, table_type)).start()
        # 不需要再加sleep，因为每个任务的options独立，不会出现属性冲突

• 生成模板的场景下，不需要在本地启动Pipeline运行，只需要构建Pipeline结构后调用pipeline.run().wait_until_finish()即可完成模板上传
• 如果是直接提交任务到Dataflow运行而不是打模板，不需要配置template_location参数，只要保证options独立就不会出现冲突

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