问题是在subdag PythonOperator中使用Jinja模板没有被渲染。解决这个问题的方法是在调用subdag的时候,传入要使用的Jinja模板的渲染上下文。以下是一个示例:
from airflow import DAG from airflow.operators.python_operator import PythonOperator, SubDagOperator from datetime import datetime, timedelta
subdag = DAG('sub_dag', default_args=default_args, schedule_interval=None)
def sub_dag_task(**context): # Jinja template to be rendered template = '{{ ds }}'
# Render the Jinja template with the context
rendered = context['templates_dict'][template].render(**context)
print(rendered)
sub_dag_operator = SubDagOperator( task_id='sub_dag_operator', subdag=subdag, op_kwargs={'templates_dict': {'template': '{{ ds }}'}}, dag=dag, )
上面的代码中,我们在SubDagOperator实例化时提供了要使用的Jinja模板的渲染上下文。在子DAG任务内,我们使用传入的context对象中的模板路径来引入渲染的模板。通过运行这个任务,我们可以在Airflow的UI中看到模板被正确地渲染。