Airflow中BigQueryExecuteQueryOperator执行含DECLARE的SQL报400错误怎么解决？

该报错是BigQuery本身的作业配置限制导致，和Airflow无关：

• 之前可正常运行的SQL属于单语句SELECT查询，BigQuery会识别为普通查询作业，支持配置destination_dataset_table、create_disposition、write_disposition参数，直接将查询结果写入指定目标表
• 新增了DECLARE、SET语句的SQL属于BigQuery多语句脚本，这类作业可能存在多个输出节点，因此BigQuery不允许在全局作业配置中指定单个目标表及对应的写入、建表策略，你在BigQueryExecuteQueryOperator中传入的相关参数会被直接拒绝，抛出400错误。

方案1：将写表逻辑移入SQL脚本（改动最小） #

移除Airflow运算符中目标表相关的配置，直接在SQL脚本内部完成写表操作，匹配脚本模式的要求：

1. 修改Airflow运算符配置，删除三个参数：

execute_query_job = BigQueryExecuteQueryOperator(
                        task_id = "execute_query_job_{}".format(destination_table),
                        use_legacy_sql = False,
                        sql = sql_query,
                        # 移除以下三个配置项
                        # destination_dataset_table = destination_table,
                        # create_disposition = "CREATE_IF_NEEDED",
                        # write_disposition = 'WRITE_TRUNCATE',
                        dag = dag
                    )

2. 修改SQL脚本，在核心查询前添加写表语句，对应原来的写入策略：

DECLARE
  temp string DEFAULT 'D';
SET temp = 'M';
-- 新增写表逻辑：CREATE OR REPLACE对应WRITE_TRUNCATE，自动匹配CREATE_IF_NEEDED的建表逻辑
CREATE OR REPLACE TABLE `你的目标表完整路径(project_id.dataset_id.table_id)` AS
WITH
  BASE_DATA AS (
  -- 原有SQL逻辑保持不变
  SELECT
    CASE
      WHEN temp = 'M' THEN DATE_TRUNC(EventDate,MONTH)
      WHEN temp =  'Q'THEN DATE_TRUNC(EventDate,QUARTER) 
  END
    ed,
    SUM(CASE
            WHEN temp = 'M' THEN tl
            WHEN temp = 'Q' THEN tl
     END) AS tl_count
  FROM
    `project_id.dataset.data_table`
  WHERE
    CASE
      WHEN temp = 'M' THEN (DATE(EventDate) BETWEEN DATE_ADD(DATE_TRUNC(DATE(CURRENT_DATE()), MONTH), INTERVAL -2 MONTH) AND DATE_ADD(DATE_TRUNC(CURRENT_DATE(), MONTH), INTERVAL -1 DAY))
      WHEN temp = 'Q' THEN (DATE(EventDate) BETWEEN DATE_ADD(DATE_TRUNC(DATE(CURRENT_DATE()), QUARTER), INTERVAL -2 QUARTER)
      AND DATE_ADD(DATE_TRUNC(CURRENT_DATE(), QUARTER), INTERVAL -1 DAY))
  END
  GROUP BY
    1
  ORDER BY
    1 DESC)
SELECT
  ed,
  tl_count
FROM
  BASE_DATA
ORDER BY
  ed DESC;

方案2：变量逻辑移到Airflow侧，恢复单语句查询 #

如果不想修改SQL写表逻辑，可以把DECLARE、SET的变量逻辑迁移到Airflow的Python代码中，拼接生成单语句SQL，即可继续使用原来的运算符配置：

# 在Airflow侧定义变量，替换SQL中的DECLARE/SET逻辑
temp = 'M'
if temp == 'M':
    date_trunc_rule = "DATE_TRUNC(EventDate,MONTH)"
    date_filter = "(DATE(EventDate) BETWEEN DATE_ADD(DATE_TRUNC(DATE(CURRENT_DATE()), MONTH), INTERVAL -2 MONTH) AND DATE_ADD(DATE_TRUNC(CURRENT_DATE(), MONTH), INTERVAL -1 DAY))"
elif temp == 'Q':
    date_trunc_rule = "DATE_TRUNC(EventDate,QUARTER)"
    date_filter = "(DATE(EventDate) BETWEEN DATE_ADD(DATE_TRUNC(DATE(CURRENT_DATE()), QUARTER), INTERVAL -2 QUARTER) AND DATE_ADD(DATE_TRUNC(CURRENT_DATE(), QUARTER), INTERVAL -1 DAY))"

# 拼接生成单语句SQL
sql_query = f"""
WITH
  BASE_DATA AS (
  SELECT
    {date_trunc_rule} AS ed,
    SUM(tl) AS tl_count
  FROM
    `project_id.dataset.data_table`
  WHERE
    {date_filter}
  GROUP BY
    1
  ORDER BY
    1 DESC)
SELECT
  ed,
  tl_count
FROM
  BASE_DATA
ORDER BY
  ed DESC;
"""
# 原有BigQueryExecuteQueryOperator配置无需修改

方案3：更换为BigQueryInsertJobOperator #

如果脚本逻辑复杂无法拆分，可以换用BigQueryInsertJobOperator，自定义构造全量作业配置，适配多语句脚本的运行要求，该方案适合复杂脚本场景，改动量相对更大。

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