迁移至Composer2.4.1与Airflow2.5.3时BigQuery外部表创建报错求助

核心原因 #

Airflow 2.5.3版本的BigQueryCreateExternalTableOperator中，source_format参数要求传入大写的官方枚举值（如PARQUET、CSV），你的代码通过format.lower()将其转为小写的parquet，不符合BigQuery API对该字段的格式要求，因此报错；不传该参数时，Operator默认采用CSV格式解析Parquet文件，自然无法识别文件自带的Schema，导致后续查询报错。

修复步骤 #

1. 修正source_format的大小写
   修改Operator中的source_format赋值，将格式值统一转为大写：

   source_format=format.upper(),
   

   无论传入的format是小写（如parquet）还是大写（如PARQUET），通过upper()转换后都能匹配BigQuery API的要求。

2. 对齐autodetect参数逻辑
   你的函数定义了autodetect参数，但返回Operator时硬编码为autodetect=True，改为使用传入的参数保持逻辑一致性：

   autodetect=autodetect,
   

3. 优化非CSV格式的参数设置
   对于Parquet这类非CSV格式，skip_leading_rows应设为None而非0，让BigQuery使用默认规则处理：

   skip_leading_rows=skip_rows if format.upper() == 'CSV' else None,
   

修复后的完整函数片段 #

def operator_bq_create_external_table(self
    , task_name: str
    , bq_project_dataset_table_destination: str
    , bucket: str
    , prefix: str
    , format: str
    , schema: list = []
    , autodetect: bool = True
    , skip_rows: int = 0
    , delimiter: str = ',') -> BigQueryCreateExternalTableOperator:

    project, _, __ = self.split_project_dataset_table(bq_project_dataset_table_destination)

    return BigQueryCreateExternalTableOperator(
        task_id=f"create_external_table_{task_name}",
        bucket=bucket,
        destination_project_dataset_table='_'.join([
            bq_project_dataset_table_destination,
            "external",
            self.get_logical_date_template('%Y%m%d%H')
        ]),
        source_objects=[
            f"{self.get_logical_date_template(prefix)}*.{format.lower()}"
        ],
        schema_fields=schema,
        autodetect=autodetect,
        source_format=format.upper(),
        skip_leading_rows=skip_rows if format.upper() == 'CSV' else None,
        field_delimiter=delimiter if format.upper() == 'CSV' else None
    )

额外验证点 #

• 确认GCS中Parquet文件的路径与source_objects模板渲染后的路径完全匹配；
• 若自动识别Schema失败，可手动传入schema_fields参数，同时确保Parquet文件本身包含有效的元数据Schema。

内容的提问来源于stack exchange，提问作者Gustavo Lima Monteiro