如何通过 sftp 进行异步文件传输？

Question

我要解决的问题如下： 我有一台包含大量数据（约 5 TB）的台式计算机，我想对其进行分析。数据由 500k 个文件组成，每个文件都可以单独分析。 对于分析，我在大学有一系列可用的服务器，但是，服务器没有空间存储所有这些数据，也没有空间存储分析的输出。

所以我的想法是将数据分段复制到服务器，运行分析，将结果传输回桌面，删除服务器上的输入和输出数据，然后重复。

对于文件传输，我昨天安装了 paramiko，它似乎工作得很好：

remote_get = 'test'
local_deliver = './test'

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.load_host_keys(os.path.expanduser(os.path.join("~", ".ssh", "known_hosts")))
ssh.connect(server, username=username, password=password)
sftp = ssh.open_sftp()

for root, dirs, files in os.walk(local_path):
    for fname in files:
        full_fname = os.path.join(root, fname)
        full_remote = os.path.join(remote_path, fname)
        sftp.put(full_fname, full_remote)
sftp.close()
ssh.close()

但是我唯一的问题是，我需要传输的数据量可能需要几天的时间来回传输，因此我希望尽可能异步启动数据传输，这样我就可以对数据进行分析当前数据集，同时传输下一个要分析的数据集。

但我不知道如何做这样的事情，谁能指出我正确的方向？

Answer 1

此解决方案使用multiprocessing.Pool 创建单独进程的任务池。每次调用apply_async 时，都会传递一个函数指针和一个参数列表。在这种情况下，要执行的函数是copy_file，arg 是文件名：

import os
import paramiko

from multiprocessing import Pool

remote_get = 'test'
local_deliver = './test'

pool = Pool(processes=4)  # Experiment with this number based on your # CPUs
def copy_file(filename):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.load_host_keys(os.path.expanduser(os.path.join("~", ".ssh", "known_hosts")))
    ssh.connect(server, username=username, password=password)
    sftp = ssh.open_sftp()

    full_fname = os.path.join(root, fname)
    full_remote = os.path.join(remote_path, fname)
    sftp.put(full_fname, full_remote)
    sftp.close()
    ssh.close()

for root, dirs, files in os.walk(local_deliver):
    for fname in files:
        pool.apply_async(copy_file, [fname])

您的原始文件中有一些变量没有被考虑在内，所以我使用了我的最佳猜测。 ssh 和 sftp 客户端创建需要移动到 copy_file 中，因为除非它是可序列化的，否则您无法保存并在进程之间共享它。

multiprocessing.Pool 上的 processes 参数可以根据 CPU 的数量进行调整，但请记住，您将在此处与多个瓶颈作斗争：1. CPU，2. NIC 带宽限制，3. 磁盘 I /O 限制。

这里有更多的多处理文档：https://docs.python.org/3/library/multiprocessing.html#using-a-pool-of-workers

编辑：我还记得 paramiko 的 SFTP 与炮击 SFTP 命令的速度要慢得多。可能值得写出批处理文件并使用subprocess.call 执行它们以获得更好的性能。