如何在SQLExecuteQueryOperator的params/parameters中使用Jinja模板生成动态表名？

方案1：使用Airflow用户自定义宏（推荐） #

在DAG定义时添加user_defined_macros，把表名模板封装成全局可复用的宏，所有SQL脚本直接引用该宏即可。后续修改表名规则时，只需改动DAG中的宏定义，无需修改所有SQL文件。

示例代码：

from airflow import DAG
from airflow.providers.postgres.operators.postgres import SQLExecuteQueryOperator
from datetime import datetime

def get_staging_table_name(**context):
    # 自定义表名生成逻辑，可随时修改规则
    return f"staging_{context['ts_nodash']}"

with DAG(
    dag_id='dynamic_table_dag',
    start_date=datetime(2024,1,1),
    schedule_interval='@daily',
    user_defined_macros={
        'staging_table_name': get_staging_table_name
    }
) as dag:
    t1 = SQLExecuteQueryOperator(
        task_id='create_table',
        conn_id='pg_conn_id',
        sql='file1.sql'
    )

    t2 = SQLExecuteQueryOperator(
        task_id='query_table',
        conn_id='pg_conn_id',
        sql='file2.sql'
    )

    t1 >> t2

对应的SQL文件：

-- file1.sql
CREATE TABLE {{ staging_table_name() }} (col1 int);

-- file2.sql
SELECT * FROM {{ staging_table_name() }};

方案2：提前渲染表名并通过XCom传递 #

先用PythonOperator生成表名并推送到XCom，后续SQL任务从XCom拉取该值，SQL脚本中直接引用params变量即可。

示例代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import SQLExecuteQueryOperator
from datetime import datetime

def generate_table_name(**context):
    table_name = f"staging_{context['ts_nodash']}"
    context['ti'].xcom_push(key='staging_table', value=table_name)
    return table_name

with DAG(
    dag_id='dynamic_table_xcom_dag',
    start_date=datetime(2024,1,1),
    schedule_interval='@daily'
) as dag:
    generate_table = PythonOperator(
        task_id='generate_table_name',
        python_callable=generate_table_name,
        provide_context=True
    )

    t1 = SQLExecuteQueryOperator(
        task_id='create_table',
        conn_id='pg_conn_id',
        sql='file1.sql',
        params={'table_name': "{{ ti.xcom_pull(task_ids='generate_table_name', key='staging_table') }}"}
    )

    t2 = SQLExecuteQueryOperator(
        task_id='query_table',
        conn_id='pg_conn_id',
        sql='file2.sql',
        params={'table_name': "{{ ti.xcom_pull(task_ids='generate_table_name', key='staging_table') }}"}
    )

    generate_table >> t1 >> t2

SQL文件保持你原来的写法：

-- file1.sql
CREATE TABLE {{ params.table_name }} (col1 int);

-- file2.sql
SELECT * FROM {{ params.table_name }};

方案3：自定义Operator解析params中的模板 #

继承SQLExecuteQueryOperator，重写execute方法，提前用Airflow的模板环境渲染params中的值，让params支持Jinja模板解析，同时避免parameters字段自动加引号的问题。

示例代码：

from airflow.providers.postgres.operators.postgres import SQLExecuteQueryOperator
from airflow.utils.context import Context

class TemplatedParamsSQLExecuteQueryOperator(SQLExecuteQueryOperator):
    def execute(self, context: Context):
        # 渲染params中的模板内容
        if self.params:
            rendered_params = {}
            for key, value in self.params.items():
                if isinstance(value, str):
                    rendered_params[key] = self.render_template(value, context)
                else:
                    rendered_params[key] = value
            self.params = rendered_params
        return super().execute(context)

# 使用自定义Operator
t1 = TemplatedParamsSQLExecuteQueryOperator(
    task_id='t1',
    conn_id='pg_conn_id',
    sql='file1.sql',
    params={'table_name': 'staging_{{ ts_nodash }}'}
)

t2 = TemplatedParamsSQLExecuteQueryOperator(
    task_id='t2',
    conn_id='pg_conn_id',
    sql='file2.sql',
    params={'table_name': 'staging_{{ ts_nodash }}'}
)

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