Airflow 1.10.2中子Dag的PythonOperator无法渲染Jinja模板问题

问题根因 #

• Airflow的Jinja模板渲染只发生在任务实例执行前的准备阶段，你当前在load_sub_dag函数（也就是DAG解析阶段）就直接用params['date']做字符串替换，此时还没到渲染环节，拿到的就是原始模板字符串。
• Airflow 1.10.2版本的原生PythonOperator默认只会渲染templates_dict、op_args两个字段，你传入的op_kwargs默认不在渲染范围内，就算直接把模板放到op_kwargs里也不会自动渲染。

修复方案（推荐） #

利用你已经开启的provide_context=True参数，在任务执行的python_callable中从上下文获取已经渲染完成的params，不要在DAG定义阶段提前处理路径。

第一步：修改子DAG中任务的传参逻辑 #

去掉load_sub_dag中提前替换路径的代码，把路径模板和国家直接传入op_kwargs：

def load_sub_dag(dag_id, start_date, email, params, schedule_interval):
    dag = DAG(
        dag_id=dag_id,
        schedule_interval=schedule_interval,
        start_date=start_date
    )

    start = PythonOperator(
        task_id="start",
        python_callable=get_start_time,
        provide_context=True,
        dag=dag
    )

    for country in params['countries']:
        i += 1
        check_files = PythonOperator(
            task_id="success_file_check_{}_{}".format(source, i),
            python_callable=check_success_file,
            # 只传原始模板和国家，不要提前替换date
            op_kwargs={"path_template": source_detail['path'], 
                       "country": country,
                       "success_file_name": success_file_name,
                       "hourly": hourly},
            provide_context=True,
            retries=0,
            dag=dag
        )
        start >> check_files
    return dag

第二步：修改check_success_file逻辑，从上下文拿渲染后的参数 #

def check_success_file(path_template, country, success_file_name, hourly, **context):
    # 这里从context获取的params已经完成Jinja渲染，date是真实的日期字符串
    rendered_params = context['params']
    rendered_date = rendered_params['date']
    # 此时再做路径替换就能拿到正确的日期
    file_path = path_template.replace('$date', rendered_date).replace('$cc', country)
    # 后续保留你原有的文件校验逻辑即可

可选方案：自定义支持渲染op_kwargs的Operator #

如果不想修改原有python_callable的入参，可以自定义扩展PythonOperator的模板字段：

# 自定义支持op_kwargs渲染的Operator
class RenderablePythonOperator(PythonOperator):
    # 把op_kwargs加入模板渲染字段列表
    template_fields = PythonOperator.template_fields + ('op_kwargs',)

# 子DAG中替换原有PythonOperator，路径直接保留模板变量
file_path_template = source_detail['path'].replace('$date', '{{ params.date }}')
for country in params['countries']:
    i += 1
    final_path_template = file_path_template.replace('$cc', country)
    check_files = RenderablePythonOperator(
        task_id="success_file_check_{}_{}".format(source, i),
        python_callable=check_success_file,
        op_kwargs={"file_path": final_path_template, 
                   "success_file_name": success_file_name,
                   "hourly": hourly},
        provide_context=True,
        retries=0,
        dag=dag
    )

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