通过跳板机用Python实现Dask端口转发至AWS服务器的问题

一、最简替代方案：复用ssh/config的subprocess调用 #

既然你已经在终端通过ssh/config实现了正常的端口转发，直接在Jupyter Notebook里用subprocess调用是最省心的方案，完全不需要折腾底层Paramiko逻辑：

import subprocess
import time

# 启动SSH端口转发，后台运行（-fN参数让SSH在后台驻留，不打开交互shell）
proc = subprocess.Popen(
    ["ssh", "-F", ".ssh/config", "-fN", "aws-dask"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# 等待几秒让端口转发链路建立
time.sleep(3)

# 检查启动状态
stdout, stderr = proc.communicate()
if proc.returncode != 0:
    print(f"端口转发启动失败: {stderr.decode()}")
else:
    print("端口转发已建立，本地8786/8787端口已映射到AWS服务器对应端口")

# 后续如果需要停止转发，执行以下代码
# proc.terminate()

这个方案完全复用你已经调试好的ssh/config配置，和终端操作逻辑完全一致，没有额外学习成本，是当前场景下的最优解。

二、Paramiko端口转发的修复方案 #

如果一定要用Paramiko实现，核心问题是你只建立了到AWS服务器的SSH连接，但没有配置本地端口到远程端口的双向转发映射。以下是完整的可运行代码：

import paramiko
import threading
import socket
import time

def forward_local_port(local_port, remote_host, remote_port, transport):
    """创建本地端口到远程端口的双向数据转发"""
    def handle_channel(channel, src_addr, dest_addr):
        # 建立到远程服务器目标端口的连接
        try:
            remote_sock = socket.socket()
            remote_sock.connect((remote_host, remote_port))
        except Exception as e:
            print(f"无法连接远程端口 {remote_port}: {str(e)}")
            channel.close()
            return
        
        # 双向转发数据的辅助函数
        def transfer_data(source, destination):
            while True:
                data = source.recv(1024)
                if not data:
                    break
                destination.send(data)
            source.close()
            destination.close()
        
        # 启动两个线程分别处理双向数据传输
        threading.Thread(target=transfer_data, args=(channel, remote_sock)).start()
        threading.Thread(target=transfer_data, args=(remote_sock, channel)).start()
    
    # 在本地监听指定端口，绑定转发处理逻辑
    transport.request_port_forward("", local_port, handler=handle_channel)

# SSH配置参数
jump_host = "跳板机IP"
jump_user = "跳板机用户名"
jump_password = "跳板机密码"
target_host = "AWS服务器IP"
target_user = "ec2-user"
private_key_path = "aws-provided.pem"

try:
    # 连接跳板机
    jump_client = paramiko.SSHClient()
    jump_client.load_system_host_keys()
    jump_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    jump_client.connect(jump_host, username=jump_user, password=jump_password)

    # 通过跳板机建立到AWS服务器的SSH通道
    jump_transport = jump_client.get_transport()
    target_session = jump_transport.open_channel(
        "direct-tcpip",
        (target_host, 22),
        ("127.0.0.1", 0)
    )

    # 连接AWS服务器
    target_client = paramiko.SSHClient()
    target_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    target_client.connect(
        target_host,
        username=target_user,
        key_filename=private_key_path,
        sock=target_session
    )
    target_transport = target_client.get_transport()

    # 配置端口转发：本地8786→AWS 8786，本地8787→AWS 8787
    forward_local_port(8786, target_host, 8786, target_transport)
    forward_local_port(8787, target_host, 8787, target_transport)

    print("端口转发已建立，按Ctrl+C停止服务")
    # 保持程序运行以维持转发链路
    while True:
        time.sleep(1)

except Exception as e:
    print(f"执行错误: {str(e)}")
finally:
    # 关闭SSH连接
    if 'target_client' in locals():
        target_client.close()
    if 'jump_client' in locals():
        jump_client.close()

关键修复说明： #

• 新增forward_local_port函数，实现本地端口与远程端口的双向数据转发逻辑
• 基于AWS服务器的SSH传输对象（target_transport）配置端口转发，而非跳板机的传输对象
• 使用线程处理双向数据传输，保证端口转发的持续运行

三、Dask客户端配置 #

端口转发建立后，Dask客户端直接连接本地端口即可：

from dask.distributed import Client

# 连接本地转发的Dask调度器端口
client = Client("localhost:8786")
print(client)

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