How to do asyncronous file transfers over sftp?

user_5197329 0
Viewed 902

The problem I'm trying to solve is the following: I have a desktop computer with large amounts of data (~5 TB), that I want to analyze. The data consist of 500k files, and each file can be analyzed individually. For the analysis I have a series of servers at the university available, however, the server does not have space for all this data, nor does it have space to store the output of the analysis.

So my idea is to copy the data over to the server in segments, run the analysis, transfer the results back to the desktop, delete both input and output data on the server, and repeat.

For the file transfer I installed paramiko yesterday and it seems to work great:

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()

However my only problem is that the amount of data I will need to transfer will likely take days to get back and forth, and hence I would love to start the data transfer asynchronously if possible, such that I can do analysis on the current dataset while transferring the next dataset to be analyzed.

But I don't have any clue how to do such a thing, can anyone point me in the right direction?

paramiko sftp parallel-processing asynchronous python
1 Answers

This solution uses multiprocessing.Pool to create a task pool of separate processes. Each time you call apply_async, you pass a function pointer and a list of args. In this case, the function to perform is copy_file and the arg is the filename:

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])

There were a few variables from your original which weren't accounted for, so I used my best guess. The ssh and sftp client creation needs to be moved inside copy_file since you can't save it and share it between processes unless it's serializable.

The processes arg on multiprocessing.Pool can be tweaked based on number of CPUs, but keep in mind that you'll be battling against multiple bottlenecks here: 1. CPU, 2. NIC bandwidth limit, 3. Disk I/O limits.

Here's some more multiprocessing documentation: https://docs.python.org/3/library/multiprocessing.html#using-a-pool-of-workers

EDIT: I just remembered as well that paramiko's SFTP vs shelling out SFTP commands was significantly slower. It may be worth writing out batch files and executing them with subprocess.call to get even better performance.

user_7699710
Related