使用BQ Hook Insert Job为BigQuery加载任务添加标签未生效排查

通过Python脚本使用Airflow的BigQuery Hook的insert_job方法将CSV数据加载到BigQuery表，功能运行正常，但添加的标签未出现在region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT的label列中。当前配置如下：

job_configuration = {
            "load": {
                "destinationTable": {
                    "project_id": PROJECT_NAME,
                    "datasetId": GCP_DATASET_NAME,
                    "tableId": table_name,
                },
                "sourceUris": [f"gs://{BUCKET}/{file_name}"],
                "writeDisposition": "WRITE_TRUNCATE",
                "skipLeadingRows": 1,
                "allowJaggedRows": True,
                "allowQuotedNewlines": True,
                "labels": { "airflow_dag": "load_job" },
            }
        }
try:
            bq_hook.insert_job(
                configuration=job_configuration,
                project_id=PROJECT_NAME,
            )

使用版本：Composer 2.0.25、Airflow 2.2.5

问题原因 #

你把labels配置放在了load子对象内部，但BigQuery的任务标签是顶层配置项，不属于load配置的一部分。这种情况下标签不会被绑定到整个加载任务上，自然不会出现在信息架构的任务记录中。

解决方法 #

将labels从load配置中移到job_configuration的根级别，修改后的配置如下：

job_configuration = {
    "load": {
        "destinationTable": {
            "project_id": PROJECT_NAME,
            "datasetId": GCP_DATASET_NAME,
            "tableId": table_name,
        },
        "sourceUris": [f"gs://{BUCKET}/{file_name}"],
        "writeDisposition": "WRITE_TRUNCATE",
        "skipLeadingRows": 1,
        "allowJaggedRows": True,
        "allowQuotedNewlines": True,
    },
    "labels": { "airflow_dag": "load_job" }  # 移至顶层配置
}

try:
    bq_hook.insert_job(
        configuration=job_configuration,
        project_id=PROJECT_NAME,
    )

验证说明 #

修改后重新运行任务，即可在region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT的label_key和label_value列中看到对应的标签内容。

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