Airflow中无法正确读取SSHOperator输出的技术问询

你的核心误区在于错误地在TaskFlow的@task函数中仅创建SSHOperator对象并返回其output属性，而非实际执行该Operator，导致下游任务拿到的是未渲染的XCom引用模板字符串，而非真实的命令输出。

具体原因 #

• SSHOperator的output属性本质是一个XCom引用的模板标记（即{{ task_instance.xcom_pull(...) }}），只有在Airflow的Operator执行上下文里才会被渲染为真实值。
• 你用@task装饰的ssh函数是一个Python任务，它的逻辑只是创建了Operator对象，并没有触发执行，返回的模板标记不会被Python任务自动解析，因此下游read任务直接打印出了原始模板字符串。

解决方案1：直接将SSHOperator作为TaskFlow任务节点 #

TaskFlow API支持直接将传统Operator作为任务节点，其output可以直接传递给下游@task函数，Airflow会自动处理XCom的渲染与传递：

import base64
import pendulum
from airflow.decorators import dag, task
from airflow.providers.ssh.operators.ssh import SSHOperator


@dag(
    start_date=pendulum.datetime(2024, 3, 1, tz="UTC"),
    tags=["test"],
)
def run_ssh():
    # 直接实例化SSHOperator作为独立任务，无需@task装饰
    ssh_task = SSHOperator(task_id="ssh", remote_host="myhost", command="ls -l")

    @task
    def read(value):
        # 解码SSHOperator返回的base64编码输出
        decoded_value = base64.b64decode(value).decode("utf-8")
        print("*********", decoded_value)

    # 将SSHOperator的输出传递给read任务
    read(ssh_task.output)


run_ssh()

解决方案2：在@task函数内部执行SSHOperator #

如果需要在@task中封装Operator逻辑，需调用Operator的execute方法并传入上下文参数，主动触发执行并获取真实输出：

import base64
import pendulum
from airflow.decorators import dag, task
from airflow.providers.ssh.operators.ssh import SSHOperator
from airflow.utils.context import Context


@dag(
    start_date=pendulum.datetime(2024, 3, 1, tz="UTC"),
    tags=["test"],
)
def run_ssh():
    @task
    def ssh(remote_host, context: Context):
        op = SSHOperator(task_id="ssh", remote_host=remote_host, command="ls -l")
        # 执行Operator并获取返回的base64编码输出
        result = op.execute(context=context)
        return result

    @task
    def read(value):
        decoded_value = base64.b64decode(value).decode("utf-8")
        print("*********", decoded_value)

    read(ssh("myhost"))


run_ssh()

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