Airflow能否调用指定远程服务器上的可执行任务？

你的需求完全可以实现，以下是几种贴合场景的方案：

1. 使用SSHOperator（推荐） #

这是Airflow官方推荐的跨服务器执行任务的方式，直接通过SSH连接目标服务器并执行脚本，能精准指定运行服务器。

步骤： #

• 先在Airflow UI的「Admin > Connections」中创建一个SSH类型的连接：
  • 填写目标服务器的IP、SSH端口（默认22）、登录用户名
  • 选择认证方式：密码或SSH密钥（密钥需放在Airflow服务器可访问的路径）
• 在DAG中使用SSHOperator调用脚本：

from airflow.providers.ssh.operators.ssh import SSHOperator
from airflow import DAG
from datetime import datetime

with DAG(
    dag_id='remote_script_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
) as dag:
    run_remote_task = SSHOperator(
        task_id='execute_remote_script',
        ssh_conn_id='your_remote_server_conn',  # 对应UI中创建的连接ID
        command='/absolute/path/to/your/script.sh',  # 目标服务器上的脚本绝对路径
        timeout=300  # 脚本执行超时时间，按需调整
    )

2. 使用PythonOperator结合SSH库 #

如果偏好用PythonOperator封装逻辑，可以借助paramiko库在Python函数中建立SSH连接并执行脚本：

步骤： #

• 在Airflow服务器安装paramiko：

pip install paramiko

• 编写DAG代码：

from airflow.operators.python import PythonOperator
from airflow import DAG
from datetime import datetime
import paramiko

def run_remote_script():
    # 初始化SSH客户端
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        # 连接目标服务器
        ssh.connect(
            hostname='目标服务器IP',
            username='登录用户名',
            password='登录密码',  # 或用key_filename='/path/to/private/key'指定密钥
            port=22
        )
        # 执行脚本命令
        stdin, stdout, stderr = ssh.exec_command('/absolute/path/to/your/script.sh')
        # 捕获输出和错误
        output = stdout.read().decode('utf-8')
        error = stderr.read().decode('utf-8')
        if error:
            raise RuntimeError(f"脚本执行失败: {error}")
        print(f"脚本输出: {output}")
    finally:
        ssh.close()

with DAG(
    dag_id='python_remote_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
) as dag:
    python_remote_task = PythonOperator(
        task_id='run_script_via_python',
        python_callable=run_remote_script
    )

3. 消息队列触发（适合复杂场景） #

如果需要更解耦的架构，可以在目标服务器部署一个监听服务（比如基于Redis/RabbitMQ），Airflow通过PythonOperator发送触发消息，监听服务收到消息后执行指定脚本。这种方式适合脚本执行逻辑复杂、需要异步处理的场景，但配置成本较高。

关键说明 #

Airflow的多节点部署（如CeleryExecutor）是将任务分配到任意可用Worker节点，无法强制指定某台固定服务器，因此上述方案更符合你「必须在特定服务器运行」的需求。

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