如何将Google Dataflow WordCount输出保存至BigQuery表？

原apache_beam.examples.wordcount示例默认仅支持将结果输出到文本文件（Cloud Storage），要改为写入BigQuery，需要修改代码逻辑并调整运行命令，具体步骤如下：

1. 修改WordCount代码 #

找到原代码中写入Cloud Storage的部分，替换为BigQuery写入逻辑：

原输出代码（写入GCS） #

result | 'Write' >> beam.io.WriteToText(known_args.output)

替换为BigQuery写入代码 #

# 定义BigQuery表的Schema
bq_table_schema = {
    'fields': [
        {'name': 'word', 'type': 'STRING', 'mode': 'REQUIRED'},
        {'name': 'count', 'type': 'INTEGER', 'mode': 'REQUIRED'}
    ]
}

# 将(word, count)元组转换为BigQuery兼容的TableRow格式
def convert_to_bq_row(word_count):
    word, count = word_count
    return {'word': word, 'count': int(count)}

# 写入BigQuery
result | 'Convert to BQ Rows' >> beam.Map(convert_to_bq_row)
       | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
           table='your-project-id:your-dataset-id.your-table-name',  # 替换为你的BQ表路径
           schema=bq_table_schema,
           create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
           write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE
       )

2. 调整运行命令 #

去掉原命令中的--output参数（不再需要指定GCS路径），确保项目、临时存储等参数正确：

python3 -m apache_beam.examples.wordcount --region us-central1 --input gs://example/shak/doc.txt --runner DataflowRunner --project your-project-name --temp_location gs://example/temp

3. 权限注意事项 #

• 确保Dataflow服务账号拥有BigQuery数据编辑权限（roles/bigquery.dataEditor），用于写入表数据
• 授予BigQuery作业提交权限（roles/bigquery.jobUser），用于提交BigQuery写入作业
• 保留Cloud Storage临时存储的读写权限，用于Dataflow运行过程中的临时文件处理

内容的提问来源于stack exchange，提问作者Rohit Gupta