如何使用Apache Beam（Python）Dataflow在BigQuery中关联两表列

核心思路 #

无需借助Apache Beam Dataflow，直接用BigQuery原生的MERGE语句就能高效完成同主键表的列关联更新，该方案比数据流作业更节省成本、执行效率更高。

具体实现 #

假设两张表的主键为id，目标表是project.dataset.target_table，需从源表project.dataset.source_table同步指定列：

MERGE INTO `project.dataset.target_table` AS target
USING `project.dataset.source_table` AS source
ON target.id = source.id
WHEN MATCHED THEN
  UPDATE SET
    target.new_column1 = source.column1,
    target.new_column2 = source.column2 -- 按需列出所有需要添加/更新的列

方案优势 #

• 无需额外移动数据：BigQuery原生操作在数据仓库层面完成，避免了Dataflow作业的资源开销
• 逻辑简洁直观：省去编写复杂数据流转换逻辑的成本
• 执行效率更高：BigQuery会自动优化关联与更新操作的执行计划

若坚持使用Dataflow的实现方式（不推荐） #

如果一定要通过Apache Beam Python Dataflow完成，可按以下步骤操作：

1. 分别读取两张表的数据：

   target_data = pipeline | 'Read Target Table' >> beam.io.ReadFromBigQuery(table='project.dataset.target_table')
   source_data = pipeline | 'Read Source Table' >> beam.io.ReadFromBigQuery(table='project.dataset.source_table')
   

2. 将数据转换为键值对（键为主键）：

   target_kv = target_data | 'Target to KV' >> beam.Map(lambda x: (x['id'], x))
   source_kv = source_data | 'Source to KV' >> beam.Map(lambda x: (x['id'], x))
   

3. 按主键关联数据：

   joined = ({'target': target_kv, 'source': source_kv} | beam.CoGroupByKey())
   

4. 合并记录，添加源表列：

   def merge_records(element):
       id, data = element
       target = data['target'][0]
       source = data['source'][0]
       # 将源表列合并到目标记录中
       target['new_column1'] = source['column1']
       target['new_column2'] = source['column2']
       return target
   
   merged_data = joined | 'Merge Records' >> beam.Map(merge_records)
   

5. 写入目标表（注意：WRITE_TRUNCATE会覆盖全表，需根据业务需求选择写入策略）：

   merged_data | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
       table='project.dataset.target_table',
       write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
       create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER
   )
   

反馈：参考Guillaume的回复，该方案已解决我的问题，且比使用Apache Beam和Dataflow的方法更优！

内容的提问来源于stack exchange，提问作者Divan Vermeulen