使用BigQueryOperator执行带参数SQL文件时Jinja模板渲染错误

问题 #

在Airflow DAG任务中，使用BigQueryOperator执行GCS路径/home/airflow/gcs/dags/movies/sql/RRR/下的cast.sql文件，通过params传递参数时出现Jinja模板渲染错误，提示找不到SQL文件。实际需求是无需使用模板，仅传递参数执行GCS中的查询，相关信息如下：

代码 #

sql_path = "/home/airflow/gcs/dags/movies/sql/RRR/cast.sql"

stage_latest_parameters = {
    "project_id": project_id,
    "dataset_name": dataset_name,
    "date": raw_latest_date,
    "source_table": source,
    "dest_table": dest
}

sql_with_parameters_task = BigQueryOperator(
    task_id='sql_with_parameters_task',
    sql=sql_path,
    params=stage_latest_parameters,
    use_legacy_sql=False,
    dag=dag
)

错误信息 #

Exception rendering Jinja template for task 'sql_with_parameters_task', field 'sql'. 
Template: '/home/airflow/gcs/dags/movies/sql/RRR/cast.sql'
...
jinja2.exceptions.TemplateNotFound: /home/airflow/gcs/dags/movies/sql/RRR/cast.sql

SQL文件内容 #

CREATE OR REPLACE TABLE `{project_id}.{dataset_name}.{dest_table}` 
AS
SELECT *, {date} AS date
FROM `{project_id}.{dataset_name}.{source_table}`

原因分析 #

1. Jinja模板渲染机制触发：BigQueryOperator默认会把sql参数的值当作Jinja模板路径处理，尝试从Airflow的模板搜索路径中查找该文件，而非直接读取GCS上的绝对路径。你指定的路径不在Jinja默认搜索范围内，因此抛出TemplateNotFound错误。
2. 参数传递方式不匹配：SQL文件使用Python字符串格式化语法{}，但你通过Jinja模板的params参数传递变量，两者语法不兼容。

解决方案 #

方法一：直接读取SQL内容并做字符串格式化 #

跳过Jinja模板渲染，直接读取文件内容后用Python原生格式化替换参数：

# 读取GCS上的SQL文件内容
with open(sql_path, 'r') as f:
    sql_query = f.read()

# 用Python字符串格式化替换参数
formatted_sql = sql_query.format(**stage_latest_parameters)

sql_with_parameters_task = BigQueryOperator(
    task_id='sql_with_parameters_task',
    sql=formatted_sql,
    use_legacy_sql=False,
    dag=dag
)

方法二：适配Jinja模板语法并配置路径 #

如果想保留params传递参数的方式，需修改SQL语法并确保文件在模板搜索路径内：

1. 修改cast.sql为Jinja语法：

CREATE OR REPLACE TABLE `{{ project_id }}.{{ dataset_name }}.{{ dest_table }}` 
AS
SELECT *, {{ date }} AS date
FROM `{{ project_id }}.{{ dataset_name }}.{{ source_table }}`

2. 调整代码使用相对路径或配置模板搜索路径：

# 假设DAG文件在/home/airflow/gcs/dags/movies/目录下，使用相对路径
sql_path = "sql/RRR/cast.sql"

# 或者在DAG定义时指定模板搜索路径
dag = DAG(
    dag_id='your_dag_id',
    template_searchpath=['/home/airflow/gcs/dags/movies/sql/RRR/'],
    # 其他DAG参数（如schedule_interval、start_date等）
)

sql_with_parameters_task = BigQueryOperator(
    task_id='sql_with_parameters_task',
    sql=sql_path,
    params=stage_latest_parameters,
    use_legacy_sql=False,
    dag=dag
)

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