使用Airflow的GoogleCloudStorageToBigQueryOperator加载时Policy Tags不生效

问题原因 #

1. 旧版Operator不支持Policy Tags：你使用的airflow.contrib.operators.gcs_to_bq.GoogleCloudStorageToBigQueryOperator属于已废弃的contrib组件，对BigQuery Schema中的policyTags字段支持不完善，会直接忽略该配置。
2. 表已存在时元数据未更新：如果目标表已经存在，WRITE_TRUNCATE仅会清空表内数据，不会修改表的结构元数据（包括Policy Tags），导致新配置无法覆盖旧的表结构。

解决方案 #

1. 切换到新版官方Operator

替换导入路径为Google Cloud Provider的新版Operator，它完整支持BigQuery的Schema字段配置，包括Policy Tags：

# 替换旧的contrib导入语句
from airflow.providers.google.cloud.operators.bigquery import GoogleCloudStorageToBigQueryOperator

2. 确保表结构被重新创建（针对已存在的表）

如果目标表已经存在，仅通过加载任务无法更新Policy Tags，需要先删除旧表再重新创建：

• 添加BigQueryDeleteTableOperator先删除目标表（忽略表不存在的情况）
• 确保加载任务的create_disposition设置为CREATE_IF_NEEDED（默认值，确保表不存在时自动创建）

3. 修正后的完整代码

import airflow
from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import GoogleCloudStorageToBigQueryOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryDeleteTableOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

with DAG(
        'gcs_to_bq',
        catchup=False,
        default_args=default_args,
        schedule_interval=None) as dag:

    DATASET_NAME = "temp"
    TABLE_NAME = "table"
    FULL_TABLE_ID = f"{DATASET_NAME}.{TABLE_NAME}"

    # 删除已存在的表，确保重新创建时应用新的Policy Tags
    delete_existing_table = BigQueryDeleteTableOperator(
        task_id="delete_existing_table",
        deletion_dataset_table=FULL_TABLE_ID,
        ignore_if_missing=True,
    )

    gcs_to_bq_load = GoogleCloudStorageToBigQueryOperator(
        task_id='gcs_to_bq_load',
        bucket="temp-bucket",
        source_objects=['dummy_data/data.csv'],
        source_format='CSV',
        skip_leading_rows=1,
        write_disposition='WRITE_TRUNCATE',
        destination_project_dataset_table=FULL_TABLE_ID,
        create_disposition="CREATE_IF_NEEDED",
        schema_fields=[
            {
                "name": "id",
                "mode": "NULLABLE",
                "type": "INT64",
                "fields": []
            },
            {
                "name": "email",
                "mode": "REQUIRED",
                "type": "STRING",
                "description": "test policy tags",
                "policyTags": {
                    "names": ["projects/project-id/locations/location/taxonomies/taxonomy-id/policyTags/policytag-id"]
                }
            },
            {
                "name": "created_at",
                "mode": "NULLABLE",
                "type": "DATE",
                "fields": []
            }
        ],
        dag=dag
    )

    # 设置任务依赖：先删表再加载数据
    delete_existing_table >> gcs_to_bq_load

4. 额外检查项

• 确认Airflow的apache-airflow-providers-google版本不低于3.0.0，旧版本可能仍存在Policy Tags支持缺陷。
• 验证执行任务的服务账号拥有bigquery.tables.create和bigquery.tables.update权限，确保能修改表的元数据。

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