Airflow中SQLExecuteQueryOperator参数Jinja渲染问题及多SQL文件优化问询

针对你遇到的问题，以下是几种可维护性更高的优化方案，避免修改多个SQL文件的麻烦：

方案1：在DAG中定义全局用户自定义宏 #

直接在DAG里注册一个名为refdate的宏，所有SQL文件统一引用这个宏，后续修改逻辑只需要改DAG文件：

DAG文件代码示例 #

from airflow import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from datetime import datetime

with DAG(
    dag_id="multi_sql_task_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    # 注册全局宏，定义refdate的计算逻辑
    user_defined_macros={
        "refdate": lambda ds: macros.ds_add(ds, -2)
    }
) as dag:
    # 任务1：执行第一个SQL文件
    run_sql_task1 = SQLExecuteQueryOperator(
        task_id="run_sql_file1",
        sql="sql/file1.sql",
        conn_id="your_database_connection"
    )

    # 任务2：执行第二个SQL文件
    run_sql_task2 = SQLExecuteQueryOperator(
        task_id="run_sql_file2",
        sql="sql/file2.sql",
        conn_id="your_database_connection"
    )

    # 设置任务依赖
    run_sql_task1 >> run_sql_task2

SQL文件写法 #

所有SQL文件里直接用{{ refdate(ds) }}调用宏即可：

-- file1.sql
SELECT * FROM user_activity WHERE activity_date = '{{ refdate(ds) }}';

-- file2.sql
INSERT INTO daily_summary (summary_date, total_users)
SELECT '{{ refdate(ds) }}', COUNT(DISTINCT user_id) FROM user_activity WHERE activity_date = '{{ refdate(ds) }}';

修改逻辑时：只需要修改DAG中user_defined_macros里的lambda表达式，比如改成lambda ds: macros.ds_add(ds, -1)，所有SQL任务自动生效。

方案2：用Jinja模板包含统一变量定义 #

把refdate的定义放到单独的Jinja模板文件里，所有SQL文件通过include引用，修改时只需要改这个模板文件：

步骤1：创建统一变量模板 #

在Airflow的模板目录（默认是dags/templates）下新建refdate_def.jinja文件：

{% set refdate = macros.ds_add(ds, -2) %}

步骤2：SQL文件引用模板 #

每个SQL文件开头加入包含语句，之后直接使用{{ refdate }}：

{% include 'refdate_def.jinja' %}
-- file1.sql
SELECT * FROM user_activity WHERE activity_date = '{{ refdate }}';

-- file2.sql
INSERT INTO daily_summary (summary_date, total_users)
SELECT '{{ refdate }}', COUNT(DISTINCT user_id) FROM user_activity WHERE activity_date = '{{ refdate }}';

修改逻辑时：只需要更新refdate_def.jinja里的表达式，所有引用该模板的SQL文件自动同步变更。

方案3：修复SQLExecuteQueryOperator的参数渲染问题 #

如果坚持用parameters传递变量，需要确保参数值被Airflow正确渲染。注意不要直接传Jinja字符串，而是在DAG上下文里直接计算或使用模板渲染：

正确写法示例 #

run_sql_task1 = SQLExecuteQueryOperator(
    task_id="run_sql_file1",
    sql="sql/file1.sql",
    conn_id="your_database_connection",
    # 直接在参数中调用宏计算refdate，Airflow会自动渲染
    parameters={"refdate": macros.ds_add(ds, -2)}
)

SQL文件写法 #

SQL文件里用{{ params.refdate }}引用参数：

SELECT * FROM user_activity WHERE activity_date = '{{ params.refdate }}';

注意：这个方案需要每个任务都配置parameters，适合需要针对单个任务调整refdate的场景，但维护效率不如前两个方案。

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