Google Cloud Composer DAG向BigQuery插入数据时遇400：Braced constructors不支持

问题背景 #

我是Google Cloud Composer（Apache Airflow）新手，正在创建DAG向跨项目BigQuery表插入数据。DAG目录已配置在gcs/dags/，目标BigQuery表的字段和数据类型也已正确创建，但执行insert_to_bq任务时始终返回上述错误。

尝试过的代码1（BigQueryOperator） #

...
SQL_INSERT = f"INSERT {BQ_TABLE_ID} ({BQ_INSERT_COLS}) VALUES('{VAR_1}', TIMESTAMP({{ts}}), '{VAR_3}')"

# DAG
with models.DAG(
    'dag_name',
    default_args = DEFAULT_ARGS,
    schedule_interval = '@once'
) as dag:
    ...
    insert_to_bq = BigQueryOperator(
        task_id = "insert_to_bq_task",
        write_disposition = 'WRITE_APPEND',
        create_disposition = 'CREATE_NEVER',
        sql = SQL_INSERT,
        use_legacy_sql = False,
        dag = dag
    ...

尝试过的代码2（BigQueryInsertJobOperator） #

...
insert_to_bq = BigQueryInsertJobOperator(
    task_id = "insert_to_bq_task",
    configuration = {
        "query" : {
            "query" : SQL_INSERT,
            "useLegacySql" : False,
            "destinationTable" : {
                "projectId" : "project-id",
                "datasetId" : "dataset_id",
                "tableId" : "table_id"
            },
            "writeDisposition" : "WRITE_APPEND",
        }
    },
    location = "my_location"
)

错误原因 #

问题出在SQL语句的模板渲染冲突上：

• 用Python f-string拼接BQ_TABLE_ID等变量时，同时保留了Airflow的模板变量{{ts}}
• f-string解析时会把{{ts}}转成{ts}，最终传递给BigQuery的SQL变成TIMESTAMP({ts})
• BigQuery会将{ts}识别为大括号构造器（Braced constructors），而标准SQL不支持这种语法，因此抛出400错误

解决方法 #

避免f-string和Airflow模板变量混用，推荐使用Airflow的params参数传递动态变量，让Airflow统一处理模板渲染。

最终生效代码（用户更新） #

SQL_INSERT = f'''
INSERT {{ params.table_id }} {{ params.columns }}
VALUES('{{ params.var_string }}', TIMESTAMP({{ params.timestamp }}))
'''
...
insert_to_bq = BigQueryOperator(
    task_id="insert_to_bq_task",
    write_disposition='WRITE_APPEND',
    create_disposition='CREATE_NEVER',
    sql=SQL_INSERT,
    use_legacy_sql=False,
    params={
        'table_id':'bq_table_id',
        'columns':'bq_columns',
        'var_string':'string_value',
        'timestamp':datetime.now(),
    }
)
...

其他可选方案 #

1. 完全不使用f-string，改用Airflow模板语法传递所有变量：

SQL_INSERT = "INSERT {{ params.table_id }} ({{ params.columns }}) VALUES('{{ params.var1 }}', TIMESTAMP('{{ ts }}'), '{{ params.var3 }}')"

# 在BigQueryOperator中传入params
params={
    'table_id': BQ_TABLE_ID,
    'columns': BQ_INSERT_COLS,
    'var1': VAR_1,
    'var3': VAR_3
}

2. 如果必须用f-string，需要对模板变量的大括号进行双重转义（不推荐，易出错）：

# 用四个大括号让f-string解析后保留{{ts}}
SQL_INSERT = f"INSERT {BQ_TABLE_ID} ({BQ_INSERT_COLS}) VALUES('{VAR_1}', TIMESTAMP({{{{ts}}}}), '{VAR_3}')"

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