如何在Airflow中使用BigQueryInsertJobOperator执行BQ加载任务

要实现从Google Cloud Storage（GCS）加载数据到BigQuery，BigQueryInsertJobOperator需要通过构造BigQuery Load Job配置来完成，核心是在configuration参数中指定load模块的具体规则。以下是完整实现示例及关键参数说明：

完整代码示例 #

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    dag_id='gcs_to_bq_load_job',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False
) as dag:

    load_gcs_to_bq = BigQueryInsertJobOperator(
        task_id='load_gcs_to_bq',
        project_id='你的GCP项目ID',
        configuration={
            "load": {
                "sourceUris": ["gs://你的存储桶/数据路径/*.csv"],  # GCS文件路径，支持通配符匹配
                "destinationTable": {
                    "projectId": "你的GCP项目ID",
                    "datasetId": "目标BQ数据集ID",
                    "tableId": "目标表名称"
                },
                "schema": {  # 可选：手动定义表结构，无需自动检测时使用
                    "fields": [
                        {"name": "id", "type": "INTEGER"},
                        {"name": "username", "type": "STRING"},
                        {"name": "register_time", "type": "TIMESTAMP"}
                    ]
                },
                "autodetect": True,  # 自动检测源文件结构，与手动schema二选一启用
                "writeDisposition": "WRITE_TRUNCATE",  # 写入模式：覆盖/追加/仅空表写入
                "sourceFormat": "CSV",  # 源文件格式：CSV/NEWLINE_DELIMITED_JSON/PARQUET等
                "skipLeadingRows": 1,  # CSV文件跳过表头行
                "allowQuotedNewlines": True  # 允许CSV中带引号的换行符
            }
        },
        gcp_conn_id='google_cloud_default'  # Airflow中配置的GCP连接ID
    )

    load_gcs_to_bq

关键参数说明 #

• sourceUris：GCS数据文件的路径，支持单个文件、多个文件或通配符（如*.csv）批量匹配
• destinationTable：目标BigQuery表的完整标识，需明确项目ID、数据集ID、表名
• writeDisposition：数据写入策略：
  • WRITE_TRUNCATE：覆盖目标表现有数据
  • WRITE_APPEND：追加数据到目标表
  • WRITE_EMPTY：仅当目标表为空时执行写入
• autodetect：设为True时，BigQuery会自动识别源文件的字段类型与结构，无需手动编写schema
• sourceFormat：指定源文件格式，需与实际文件类型匹配
• skipLeadingRows：CSV文件中需要跳过的行数，通常设为1以跳过表头

与旧版BigQueryExecuteQueryOperator的差异 #

旧版通常通过执行LOAD DATA SQL语句实现数据加载，而BigQueryInsertJobOperator直接调用BigQuery的Job API，更贴合官方最新操作规范，同时支持分区表加载、集群表加载等复杂场景的配置。

内容的提问来源于stack exchange，提问作者Sander van den Oord