如何将Airflow的S3Hook改为SSHook/SFTPHook？SFTP下载及连接配置疑问

一、代码替换实现 #

1. 使用SFTPHook（官方推荐，SFTP专用Hook） #

SFTPHook是Airflow专门用于SFTP操作的Hook，替换后的代码如下：

def download_csv_from_sftp(remote_path: str, local_path: str, conn_id: str = "sftp_conn") -> str:
    from airflow.providers.sftp.hooks.sftp import SFTPHook
    
    hook = SFTPHook(conn_id=conn_id)
    # 从SFTP远程路径下载文件到本地指定路径
    hook.retrieve_file(remote_full_path=remote_path, local_full_path=local_path)
    
    return local_path

说明：

• 原S3的bucket_name和key参数合并为remote_path（SFTP的完整远程文件路径，比如/remote_dir/data.csv）
• retrieve_file方法负责文件下载，直接指定远程和本地完整路径即可

2. 使用SSHHook（通过SSH连接实现SFTP下载） #

如果需要通过SSH连接间接操作SFTP，可使用SSHHook：

def download_csv_via_ssh(remote_path: str, local_path: str, conn_id: str = "ssh_conn") -> str:
    from airflow.providers.ssh.hooks.ssh import SSHHook
    
    hook = SSHHook(conn_id=conn_id)
    # 建立SSH连接并打开SFTP会话
    with hook.get_conn() as ssh_client:
        sftp_client = ssh_client.open_sftp()
        sftp_client.get(remote_path, local_path)
        sftp_client.close()
    
    return local_path

二、Airflow连接配置 #

SFTP连接配置（对应SFTPHook） #

在Airflow UI的Admin -> Connections页面新建连接：

• Conn Id：自定义（比如sftp_conn，需与代码中conn_id参数一致）
• Conn Type：选择SFTP
• Host：SFTP服务器的IP地址或域名
• Login：SFTP登录用户名
• Password：SFTP登录密码（密码认证时填写）
• Port：SFTP服务端口，默认22，若修改过则填对应端口
• Extra：密钥认证时填写JSON格式配置，示例：

  {"key_file": "/opt/airflow/keys/sftp_private_key", "key_passphrase": "your_key_passphrase"}
  

  （若密钥无密码，可省略key_passphrase字段）

SSH连接配置（对应SSHHook） #

同样在Admin -> Connections页面新建：

• Conn Id：自定义（比如ssh_conn）
• Conn Type：选择SSH
• Host：SSH服务器的IP地址或域名
• Login：SSH登录用户名
• Password：SSH登录密码（密码认证时填写）
• Port：SSH服务端口，默认22
• Extra：密钥认证时填写JSON格式配置，示例：

  {"key_file": "/opt/airflow/keys/ssh_private_key", "key_passphrase": "your_key_passphrase"}
  

注意事项 #

• 确保Airflow Worker节点能访问SFTP/SSH服务器的对应端口（22或自定义端口）
• 私钥文件需放置在Airflow Worker可读取的路径，且权限设置为600（避免权限过大报错）
• Airflow 2.x版本需提前安装对应Provider包：

  pip install apache-airflow-providers-sftp apache-airflow-providers-ssh
  

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