Composer DAG配置BigQuery外部表：Source Data Partitioning与Source URI Prefix

要实现启用源数据分区（Source Data Partitioning）并设置Source URI Prefix，核心是在BigQueryCreateExternalTableOperator的external_table_resource参数中，配置externalDataConfiguration下的对应字段。以下是完整的实现示例：

完整DAG代码 #

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryCreateExternalTableOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'create_partitioned_bq_external_table',
    default_args=default_args,
    description='Create BigQuery external table with source data partitioning',
    schedule_interval=None,  # 按需设置调度规则
    catchup=False,
) as dag:

    create_external_table = BigQueryCreateExternalTableOperator(
        task_id='create_partitioned_external_table',
        project_id='你的GCP项目ID',
        dataset_id='目标数据集ID',
        table_id='外部表名称',
        external_table_resource={
            'configuration': {
                'externalDataConfiguration': {
                    'sourceUris': ['gs://my-gs-bucket/folder1/subfolder/country*'],
                    'sourceFormat': 'PARQUET',
                    # 设置Source URI Prefix，对应控制台配置项
                    'sourceUriPrefix': 'gs://my-gs-bucket/folder1/subfolder',
                    # 启用源数据分区配置
                    'partitioning': {
                        'type': 'FIELD_PARTITIONING',
                        'fieldName': 'country',
                        # 可选：强制查询时必须指定分区过滤条件
                        # 'requirePartitionFilter': True
                    },
                    # 可选：自动推断Parquet文件的表结构
                    'autodetect': True
                }
            }
        },
        gcp_conn_id='google_cloud_default',  # 确保已配置有效的GCP连接
    )

    create_external_table

关键参数说明 #

• sourceUriPrefix：指定分区路径的公共前缀，BigQuery会自动识别该前缀下的key=value格式路径（如country=US）作为分区字段，完全对应控制台的「Source URI Prefix」配置。
• partitioning：设置为FIELD_PARTITIONING类型，fieldName对应路径中的分区键（此处为country），以此启用源数据分区功能，和控制台操作的「启用Source Data Partitioning」效果一致。
• sourceUris：使用通配符匹配所有分区路径下的Parquet文件，逻辑和控制台选择URI模式一致。

注意事项 #

1. 确保GCS路径的分区格式符合BigQuery要求：必须是key=value的层级结构。
2. 如果已有固定表结构，可在externalDataConfiguration中添加schema字段手动定义，无需开启autodetect。
3. 验证Composer服务账号权限：需拥有BigQuery数据集编辑权限和GCS存储桶的读取权限。

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