Airflow中SnowflakeOperator的params传Jinja变量渲染失败问题

你的代码有3处核心问题导致渲染失败：

• 基础语法错误：params字典的键值对错误使用等号=赋值，Python字典要求键值对用冒号:分隔，这段代码在DAG解析阶段就会抛出语法错误；同时任务变量名前后不一致，前定义task1后调用task_2会触发名称错误。
• 渲染逻辑错误：Jinja模板的渲染发生在任务实例运行时，你在DAG解析的顶层代码里用f-string拼接{{ execution_date.int_timestamp | int }}字符串，此时宏还未被执行，拼入的只是未渲染的字面量模板字符串；且默认配置下Airflow不会递归渲染params字典内的Jinja语法，只有被标记为模板字段的sql参数会被渲染。
• 模板写法不规范：SQL模板里{{params.table}}漏写了Jinja标记和变量名之间的空格，部分Jinja版本会出现解析异常。

不需要提前在顶层定义带Jinja语法的execution_date变量，直接把宏逻辑写在被渲染的sql字段、op_kwargs字段内部即可，同时修正语法错误，参考代码如下：

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

# 替换为你实际的文件名
FILE_NAME = "biz_data.csv"

with DAG(
    dag_id="s3_load_to_snowflake",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
    # 若需要params内部支持模板渲染可开启下方配置
    # render_template_as_native_obj=True
) as dag:
    task1 = PythonOperator(
        task_id="s3_create",
        op_kwargs={"s3_key": "{{ execution_date.int_timestamp | int }}_" + FILE_NAME}
    )

    task2 = SnowflakeOperator(
        task_id="load_data",
        sql="""
            -- 可在此处插入你需要的额外表操作逻辑，比如删表、建表、权限校验等
            TRUNCATE TABLE IF EXISTS {{ params.target_table }};
            COPY INTO {{ params.target_table }}
            FROM {{ params.s3_stage }}
            FILES = ('{{ execution_date.int_timestamp | int }}_' || '{{ params.target_file }}')
            FILE_FORMAT = (TYPE = CSV, SKIP_HEADER = 1);
        """,
        params={
            "target_table": "test_table",
            "s3_stage": "@your_defined_s3_stage",
            "target_file": FILE_NAME
        }
    )

    task1 >> task2

如果你需要在多个任务里复用生成的S3文件名，可以通过XCom在任务间传递值，不要在DAG解析阶段硬拼Jinja字符串。

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