Airflow 2.1.4中SSHOperator经XCom传值返回异常字符串问题

XCom拉取到的MTcySyBhaXJmbG93CTIuMUcgQXV0b0hpdHVtX05ld为Base64编码后的SSH命令执行输出，修改enable_xcom_pickling参数无法解决该问题：

• Airflow 2.1.4版本内置SSHOperator在开启XCom推送时，默认将SSH通道返回的原始字节流做Base64编码后写入XCom，该设计用于兼容二进制类命令输出、规避特殊字符、跨字符集导致的传输异常
• BashOperator无默认输出编码逻辑，直接存储明文结果，因此相同拉取逻辑下可正常拿到返回值。

方法1：拉取后手动解码（成本最低，推荐） #

在PythonOperator的回调函数中，对拉取到的值做Base64解码即可得到原始文本，修改后的代码如下：

import base64

def decision_function(**context):
    ti = context['ti']
    output_b64 = ti.xcom_pull(task_ids='ssh_task')
    # 增加判空逻辑，避免命令无返回时触发解码异常
    output_ssh = base64.b64decode(output_b64).decode('utf-8').strip() if output_b64 else ""
    print('ssh output is: {}'.format(output_ssh))

ls = SSHOperator(
    task_id="ssh_task",
    command="ls",
    ssh_hook=sshHook,
    dag=mydag
)


get_xcom = PythonOperator(
     task_id='test',
     python_callable=decision_function,
     provide_context=True,
     dag=mydag
)

ls >> get_xcom

示例中的异常字符串经上述逻辑解码后，即可得到正常的ls命令执行结果。

方法2：自定义Operator自动解码 #

如果多个任务需要复用SSHOperator的输出，可继承原生SSHOperator重写执行逻辑，在推送XCom前自动完成解码，避免每个下游任务重复写解码逻辑：

from airflow.providers.ssh.operators.ssh import SSHOperator
import base64

class PlainTextSSHOperator(SSHOperator):
    def execute(self, context):
        raw_result = super().execute(context)
        return base64.b64decode(raw_result).decode('utf-8').strip() if raw_result else ""

后续使用PlainTextSSHOperator替换原生SSHOperator即可，XCom中存储的直接是明文输出，下游拉取时无需额外处理。

• 非必要不要开启enable_xcom_pickling，该参数开启后会允许XCom序列化任意Python对象，存在远程代码执行的安全风险
• 如果执行的SSH命令返回二进制内容（如压缩包、媒体文件），不要直接按utf-8格式解码，根据业务需求处理原始字节即可

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