Airflow 2.7.1使用SFTP/SSH Operator批量复制远程文件至本地失败求助

错误原因分析 #

你遇到的[Errno 21] Is a directory报错，核心是SFTPOperator对路径的识别逻辑不匹配：当你指定批量远程文件时，本地路径必须明确标记为目录，否则Operator会把它当成单个文件名处理，导致冲突。

方案一：修复SFTPOperator批量复制配置 #

SFTPOperator支持批量文件传输，但需要严格配置参数：

1. 通配符批量复制（推荐） #

如果远程文件有统一命名规则，用通配符匹配，同时确保本地路径以斜杠结尾（明确是目录）：

from airflow.providers.ssh.operators.sftp import SFTPOperator
from airflow.providers.ssh.hooks.ssh import SSHHook

# 替换为你的SSH连接ID
ssh_hook = SSHHook(ssh_conn_id="your_remote_ssh_conn")

sftp_batch_task = SFTPOperator(
    task_id="sftp_copy_batch_files",
    ssh_hook=ssh_hook,
    remote_filepath="/remote/server/path/*.txt",  # 用通配符匹配批量文件
    local_filepath="/mnt/d/New_folder/",  # 本地目录必须以/结尾
    operation="get",  # 明确是从远程拉取到本地（默认是put，本地推远程）
    create_intermediate_dirs=True,  # 自动创建不存在的本地目录
    dag=dag
)

2. 指定多个具体文件复制 #

如果需要复制的文件无统一规则，用PythonOperator结合SSHHook手动处理：

from airflow.operators.python import PythonOperator
from airflow.providers.ssh.hooks.ssh import SSHHook

def copy_specific_files():
    ssh_hook = SSHHook(ssh_conn_id="your_remote_ssh_conn")
    # 定义需要复制的远程文件列表
    remote_files = [
        "/remote/path/file1.csv",
        "/remote/path/subdir/file2.log",
        "/remote/path/file3.txt"
    ]
    local_dir = "/mnt/d/New_folder/"

    with ssh_hook.get_conn() as ssh_client:
        sftp_client = ssh_client.open_sftp()
        for remote_file in remote_files:
            # 提取文件名，拼接本地路径
            filename = remote_file.split("/")[-1]
            local_file_path = f"{local_dir}{filename}"
            sftp_client.get(remote_file, local_file_path)
        sftp_client.close()

python_copy_task = PythonOperator(
    task_id="copy_specific_remote_files",
    python_callable=copy_specific_files,
    dag=dag
)

方案二：用SSHOperator/BashOperator执行scp命令 #

之前scp失败大概率是命令参数错误，以下是正确写法：

1. 用BashOperator直接执行scp（Airflow运行在本地时推荐） #

from airflow.operators.bash import BashOperator

bash_scp_task = BashOperator(
    task_id="bash_scp_batch",
    # 替换为你的远程SSH信息、文件路径
    bash_command="scp -i /path/to/your/private_key user@remote_host:/remote/path/*.txt /mnt/d/New_folder/",
    dag=dag
)

2. 用SSHOperator执行rsync（更稳定的批量复制） #

rsync比scp更适合批量文件，支持断点续传：

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

rsync_task = SSHOperator(
    task_id="rsync_batch_copy",
    ssh_conn_id="your_remote_ssh_conn",
    # 远程执行rsync，将文件推送到本地（需本地开启SSH服务）
    command="rsync -avz /remote/path/*.txt user@local_host:/mnt/d/New_folder/",
    dag=dag
)

必做排查项 #

• 权限检查：确认Airflow运行用户对/mnt/d/New_folder有读写权限，远程用户对目标文件有读取权限
• 路径格式：Linux路径用/，Windows路径用D:\New_folder（若Airflow运行在原生Windows）
• SSH连接验证：在Airflow UI中测试你的SSH连接ID，确保主机、端口、密钥/密码配置正确
• 通配符有效性：在远程主机上手动执行ls /remote/path/*.txt，确认有匹配的文件

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