Airflow XCOM跨任务传参时模板未渲染问题解决方案

Airflow的Jinja模板渲染发生在任务实例实际执行前的瞬间，而非DAG文件被调度器解析加载的阶段。你在定义DAG的Python代码中直接调用base64.b64decode()、os.path.basename()时，传入的{{ task_instance.xcom_pull(...) }}还只是未被替换的原始模板字符串，根本不是上游任务推送的XCom实际值，自然会触发解码、路径处理错误。
这个场景不需要新增中间处理任务，直接利用Jinja模板的内置能力就能完成所有逻辑。

SFTPOperator的remote_filepath、local_filepath本身就是原生支持Jinja渲染的模板字段，你可以直接在模板表达式内完成base64解码、basename提取的操作，不需要在DAG定义层调用Python原生函数。
Airflow内置了b64decode模板过滤器，路径的basename可以直接通过字符串切分获取（Linux环境路径以/为分隔符，取最后一段即可），代码如下：

with DAG('my_dag',
         default_args = dict(
                             # 你的其他默认参数
                             xcom_push = True,
                             )
         ) as dag:

    # 省略已有的get_update_id任务

    get_results  = SSHOperator(task_id = 'get_results',
                               ssh_conn_id = 'my_remote_server',
                               command = """cd ~/path/to/dir && python results.py -t p -u {{ task_instance.xcom_pull(task_ids='get_update_id') }}""",
                               cmd_timeout = -1,
                               )

    download_results = SFTPOperator(task_id = 'download_results',
                                    ssh_conn_id = 'my_remote_server',
                                    # 模板渲染时自动做base64解码，trim用于过滤stdout可能带的多余换行、空白
                                    remote_filepath = "{{ ti.xcom_pull(task_ids='get_results') | trim | b64decode }}",
                                    # 解码后按/切分取最后一段，等价于os.path.basename
                                    local_filepath = "{{ (ti.xcom_pull(task_ids='get_results') | trim | b64decode).split('/') | last }}",
                                    operation = 'get',
                                    )
    
    # 任务依赖
    get_update_id >> get_results >> download_results

如果你不想用字符串切分的写法，想直接复用os.path.basename的逻辑，可以在DAG初始化时将os.path注册为自定义宏，注入到Jinja渲染上下文里，模板内就能直接调用该方法：

import os

with DAG('my_dag',
         default_args = dict(
                             xcom_push = True,
                             ),
         # 注册自定义宏
         user_defined_macros = {
             "os_path": os.path
         }
         ) as dag:
    
    # 其他任务保持不变
    download_results = SFTPOperator(task_id = 'download_results',
                                    ssh_conn_id = 'my_remote_server',
                                    remote_filepath = "{{ ti.xcom_pull(task_ids='get_results') | trim | b64decode }}",
                                    # 直接在模板里调用os.path.basename
                                    local_filepath = "{{ os_path.basename(ti.xcom_pull(task_ids='get_results') | trim | b64decode) }}",
                                    operation = 'get',
                                    )

• 确认上游results.py的stdout只输出base64编码后的文件全路径，不要夹杂其他日志、打印内容，否则会导致base64解码失败。如果存在多余输出，建议修改脚本仅输出目标路径，或者在Jinja中增加字符串截取逻辑过滤无关内容。
• 如果你的远程服务器是Windows环境，路径分隔符为\，将split的参数替换为'\\'即可。
• 只有当你需要对路径做复杂业务逻辑处理（比如路径重命名、规则校验、多路径拼接等），Jinja表达式难以简洁实现时，才需要新增PythonOperator作为中间任务处理XCom值，当前取basename的场景完全不需要。

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