Airflow PythonOperator执行BigQuery查询报错 同代码普通环境正常

你贴出的硬编码SQL中不存在"Sedgwick"字符串，且报错指向SQL第4行，和你贴的3行SQL长度不符，核心原因是DAG中实际提交到BigQuery执行的SQL，和你本地测试的硬编码SQL不是同一份内容。本地直接跑Python脚本不会触发Airflow的DAG解析、模板渲染、XCom传参流程，所以能正常运行；DAG执行时额外的流程修改了bq_sql变量的内容，把未做转义的"Sedgwick"（地址/城市字段值）直接拼入了SQL语法位置，被BigQuery识别为非法标识符才抛出400错误。

按以下步骤快速定位具体问题点：

• 在client.query(bq_sql)代码前加一行print(repr(bq_sql))，执行DAG后查看日志输出的完整SQL，就能直接看到"Sedgwick"出现在SQL中的具体位置，确认是拼接错误还是渲染错误。
• 检查代码中bq_sql变量的赋值逻辑：确认你没有在后续代码中拼接从XCom拉取、外部传入的动态字段值，90%以上的同类问题都是直接字符串拼接SQL时，没有给字符串类型的字段值加单引号导致的。
• 检查Jinja渲染冲突：Airflow会自动对PythonOperator中标记为模板字段的字符串做Jinja渲染，如果你的SQL中包含{{ }}类模板标记，会被Airflow自动替换为对应上下文变量，替换出的内容如果是包含"Sedgwick"的地址值且未加引号，就会触发语法错误。

• 禁止直接用字符串拼接的方式构造带动态参数的SQL，改用BigQuery官方客户端支持的参数化查询，从根源避免语法错误和SQL注入风险，参考实现：

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/airflow/airflow/keys/the_JSON.json"
client = bigquery.Client()
geoclient = GeocodioClient('apikey')

bq_sql = """
SELECT ATTOM_ID, ADDRESS, CITY, STATE, ZIP 
FROM `project.dataset.tbl` 
WHERE GEOCODE_DT < DATE_SUB(CURRENT_DATE(), INTERVAL 2 month)  
AND ATTOM_ID IN UNNEST(@target_attom_ids)
"""
job_config = bigquery.QueryJobConfig(
    query_parameters=[
        bigquery.ArrayQueryParameter("target_attom_ids", "INT64", [1, 2]),
    ]
)
query_job = client.query(bq_sql, job_config=job_config)
results = query_job.result()

• 如果确实需要动态拼接SQL片段，拼接完成后必须打印完整SQL做校验，字符串类型的值必须用单引号包裹，同时转义值内部自带的单引号。
• 如果不需要Airflow对PythonOperator内的字符串做Jinja渲染，可以自定义关闭模板渲染的Operator类，避免意外替换：

from airflow.operators.python import PythonOperator

class NoRenderPythonOperator(PythonOperator):
    template_fields = ()

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