每日批处理场景下如何将GCS Parquet数据写入BigQuery

针对每日1000行Parquet格式数据从GCS写入BigQuery的批处理场景，按运维成本、性价比排序有两个可落地方案，优先选择第一个：

• 首选方案：BigQuery原生LOAD能力 + Cloud Scheduler定时触发。原生支持Parquet格式解析，无需维护额外计算资源，数据加载耗时秒级，成本几乎为0，完全匹配当前无复杂数据转换的需求
• 备选方案：Dataflow Python SDK自定义管道。适合后续需要叠加数据清洗、字段转换、多源数据合并等复杂逻辑的场景，灵活度更高，但运维成本、运行成本显著高于前者

方案1：BigQuery原生加载（推荐，10分钟可完成配置） #

前置准备：给BigQuery默认服务账号绑定目标GCS桶的存储对象查看者权限、目标BigQuery数据集的数据编辑者权限。

1. 提前在BigQuery创建目标业务表，表结构与Parquet文件字段完全对齐：

2. 编写Parquet加载SQL，直接用BigQuery内置的批量加载语法：

-- 每日运行时替换uris里的日期路径，避免重复加载历史数据
LOAD DATA INTO `你的GCP项目ID.你的数据集名.你的目标表名`
FROM FILES (
  format = 'PARQUET',
  uris = ['gs://你的GCS桶名/每日parquet文件路径/*.parquet']
);

提示：建议日常生成Parquet文件时按日期分区存放，比如路径格式为parquet_data/dt=20240520/*.parquet，每日定时任务只加载对应日期路径下的文件，从根源避免重复写入。

3. 定时配置：用Cloud Scheduler创建每日固定时间触发的任务，直接调用BigQuery Jobs接口执行上述SQL即可；如果需要加前置校验逻辑，也可以写一个几十行的Python Cloud Function封装加载逻辑，由Cloud Scheduler定时触发函数运行。

方案2：Dataflow Python SDK实现（适配后续复杂加工需求） #

如果确定要用Dataflow实现，直接基于Apache Beam Python SDK编写管道即可，官方原生支持Parquet读取、BigQuery写入，无需自行开发格式解析逻辑。

1. 安装依赖包

pip install apache-beam[gcp]

2. 编写管道代码

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions

# 基础管道配置
pipeline_opts = PipelineOptions()
gcp_opts = pipeline_opts.view_as(GoogleCloudOptions)
gcp_opts.project = "你的GCP项目ID"
gcp_opts.region = "Dataflow运行区域，例如asia-east1"
gcp_opts.temp_location = "gs://你的GCS桶名/dataflow_temp/"  # 管道临时文件存放路径
gcp_opts.job_name = "gcs-parquet-to-bq-daily"

# 目标BigQuery表结构
BQ_TABLE_SCHEMA = {
    "fields": [
        {"name": "Date", "type": "DATETIME", "mode": "REQUIRED"},
        {"name": "Company", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Tel", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Address", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Staff", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Lank", "type": "INTEGER", "mode": "NULLABLE"}
    ]
}

with beam.Pipeline(options=pipeline_opts) as p:
    (
        p
        # 读取GCS上的Parquet文件，支持通配符批量匹配
        | "ReadParquetFromGCS" >> beam.io.ReadFromParquet(
            file_pattern="gs://你的GCS桶名/parquet文件路径/*.parquet"
        )
        # 如需做数据清洗、字段转换，可在此处追加Beam转换步骤
        # | "DataTransform" >> beam.Map(自定义处理函数)
        # 写入BigQuery表
        | "WriteToBigQuery" >> beam.io.WriteToBigQuery(
            table="你的GCP项目ID:你的数据集名.你的目标表名",
            schema=BQ_TABLE_SCHEMA,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
        )
    )

3. 部署与定时：将代码提交生成Dataflow自定义模板，后续通过Cloud Scheduler每日定时触发模板运行即可。

• 所有运行任务的服务账号遵循最小权限原则，仅分配必要的GCS读取、BigQuery写入权限即可，避免权限溢出带来安全风险
• 你的日数据量仅千行级别，完全不需要启动多worker的Dataflow集群，优先选BigQuery原生加载方案，没有额外组件运维成本，稳定性更高

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