使用python将文件从FTPS传输到SFTP

Question

我正在做一些性能测试，以将大文件（~ 4 GB）从 FTPS 传输到 SFTP 服务器。 我做了一些研究并尝试了 python 脚本，看看从 FTPS 获取文件并传输到 SFTP 是否有任何性能改进。

FTPS 连接设置

def create_connection(self):
    print('Creating session..........')
    ftp = ftplib.FTP_TLS()
    # ftp.set_debuglevel(2)
    ftp.connect(self.host, self.port)
    ftp.login(self.user, self.passwd)
    ftp.prot_p()
    # optimize socket params for download task
    print('Optimizing socket..........')
    ftp.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    ftp.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 75)
    ftp.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
    print('Session created successfully')
    return ftp

 def get_file(self, ftp_session, dst_filename, local_filename):
    print('Starting download........', datetime.now())
    myfile = BytesIO()
    print(myfile.tell())
    ftp_session.retrbinary('RETR %s' % dst_filename, myfile.write)
    print(myfile.tell())
    print('Download completed ........', datetime.now())

对于 SFTP 连接，我使用的是 paramiko，

host, port = "abc.com", 22
    transport = paramiko.Transport((host, port))
    username, password = "user", "pwd"
    transport.connect(None, username, password)
    transport.default_window_size =  3 * 1024 * 1024
    sftp = paramiko.SFTPClient.from_transport(transport)
    myfile.seek(0)
    sftp.putfo(fl=myfile, remotepath='remotepath/' + local_filename)
    sftp.close()

我使用的是 BytesIO，这样我就可以将文件保存在内存中并在复制时对其进行流式传输。以下代码可以复制文件，但大约需要 20 分钟。该代码首先将文件复制到内存中，然后进行传输。有没有什么方法可以更有效地传输文件？

Answer 1

经过一些简短的谷歌搜索（是的，谷歌确实有效=）我偶然发现了这个线程：

Paramiko 无法下载大于 1GB 的大文件

import threading, os, time, paramiko

#you could make the number of threads relative to file size
NUM_THREADS = 4
MAX_RETRIES = 10

def make_filepart_path(file_path, part_number):
    """creates filepart path from filepath"""
    return "%s.filepart.%s" % (file_path, part_number+1)

def write_chunks(chunks, tnum, local_file_part, username, password, ftp_server, max_retries):
    ssh_conn = sftp_client = None
    for retry in range(max_retries):
        try:
            ssh_conn = paramiko.Transport((ftp_server, port))
            ssh_conn.connect(username=username, password=password)
            sftp_client = paramiko.SFTPClient.from_transport(ssh_conn)
            with sftp_client.open(sftp_file, "rb") as infile:
                with open(local_file_part, "wb") as outfile:
                    for chunk in infile.readv(chunks):
                        outfile.write(chunk)
            break
        except (EOFError, paramiko.ssh_exception.SSHException, OSError) as x:
            retry += 1
            print("%s %s Thread %s - > retrying %s..." % (type(x), x, tnum, retry))
            time.sleep(abs(retry) * 10)
        finally:
            if hasattr(sftp_client, "close") and callable(sftp_client.close):
                sftp_client.close()
            if hasattr(ssh_conn, "close") and callable(ssh_conn.close):
                ssh_conn.close()



start_time = time.time()

for retry in range(MAX_RETRIES):
    try:
        ssh_conn = paramiko.Transport((ftp_server, port))
        ssh_conn.connect(username=username, password=password)
        sftp_client = paramiko.SFTPClient.from_transport(ssh_conn)
        # connect to get the file's size in order to calculate chunks
        filesize = sftp_client.stat(sftp_file).st_size
        sftp_client.close()
        ssh_conn.close()
        chunksize = pow(4, 12)
        chunks = [(offset, chunksize) for offset in range(0, filesize, chunksize)]
        thread_chunk_size = (len(chunks) // NUM_THREADS) + 1
        # break the chunks into sub lists to hand off to threads
        thread_chunks = [chunks[i:i+thread_chunk_size] for i in range(0, len(chunks) - 1, thread_chunk_size)]
        threads = []
        fileparts = []
        for thread_num in range(len(thread_chunks)):
            local_file_part = make_filepart_path(local_file, thread_num) 
            args = (thread_chunks[thread_num], thread_num, local_file_part, username, password, ftp_server, MAX_RETRIES)
            threads.append(threading.Thread(target=write_chunks, args=args))
            fileparts.append(local_file_part)
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        # join file parts into one file, remove fileparts
        with open(local_file, "wb") as outfile:
            for filepart in fileparts:
                with open(filepart, "rb") as infile:
                    outfile.write(infile.read())
                os.remove(filepart)
        break
    except (EOFError, paramiko.ssh_exception.SSHException, OSError) as x:
        retry += 1
        print("%s %s - > retrying %s..." % (type(x), x, retry))
        time.sleep(abs(retry) * 10)
    finally:
       if hasattr(sftp_client, "close") and callable(sftp_client.close):
           sftp_client.close()
       if hasattr(ssh_conn, "close") and callable(ssh_conn.close):
           ssh_conn.close()


print("Loading File %s Took %d seconds " % (sftp_file, time.time() - start_time))