Airflow TaskFlow中Jinja模板无法渲染，该功能是否支持？

Jinja模板在Airflow的TaskFlow模式下完全可以正常工作，你遇到的渲染失败是因为代码写法错误，而非模式本身不支持。

问题原因 #

你在write_output任务函数里直接将Jinja模板字符串赋值给unique_id，但Airflow不会自动渲染任务函数内部的普通字符串——TaskFlow模式下，模板渲染默认只作用于任务的参数（比如通过params传递的内容），而非函数内部定义的字符串变量。

解决办法 #

方法1：直接获取Airflow上下文变量（推荐）

不需要依赖Jinja模板，直接从Airflow上下文中提取所需变量进行拼接，更符合Pythonic写法：

from airflow import DAG
from airflow.hooks.filesystem import LocalFilesystemHook
from airflow.operators.python import get_current_context, task
from datetime import days_ago

@dag(default_args={"owner": "airflow"}, schedule_interval=None, start_date=days_ago(1))
def my_dag():
    fs_hook = LocalFilesystemHook()
    
    @task
    def write_output(output_name):
        output = "This is some output."
        # 获取当前任务上下文
        context = get_current_context()
        ds = context["ds"]
        ti = context["ti"]
        # 用Python变量拼接生成唯一ID
        unique_id = f"{ds}_{ti.task_id}_{ti.try_number}"
        file_path = f"/path/to/output/{output_name}_{unique_id}.txt"
        fs_hook.write(output, file_path)
        return file_path

my_dag()

方法2：手动渲染Jinja模板

如果一定要使用Jinja模板语法，可以调用Airflow的render_template方法手动渲染字符串：

from airflow import DAG
from airflow.hooks.filesystem import LocalFilesystemHook
from airflow.operators.python import get_current_context, task
from airflow.templates import render_template
from datetime import days_ago

@dag(default_args={"owner": "airflow"}, schedule_interval=None, start_date=days_ago(1))
def my_dag():
    fs_hook = LocalFilesystemHook()
    
    @task
    def write_output(output_name):
        output = "This is some output."
        context = get_current_context()
        # 定义Jinja模板字符串
        unique_id_template = "{{ ds }}_{{ ti.task_id }}_{{ ti.try_number }}"
        # 手动渲染模板
        unique_id = render_template(unique_id_template, context)
        file_path = f"/path/to/output/{output_name}_{unique_id}.txt"
        fs_hook.write(output, file_path)
        return file_path

my_dag()

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