如何使用boto3加速不同凭证的AWS S3间直接传输？

问题背景 #

需要加速两个使用不同凭证的AWS S3兼容存储之间的文件直接传输，当前使用参考代码传输126MB文件耗时超30秒，效果未达预期，需排查代码问题并提供优化方案。

原代码 #

def raw_from_prod_to_nonprod():
    import boto3
    import boto3.s3.transfer as s3transfer
    import botocore

    import urllib3
    urllib3.disable_warnings()

    from io import BytesIO

    service = "s3"
    verify_bool = False
    endpoint_url_with_https = "https://xx-xxxx.xx.xxx.xx:1234"

    src_access_key = ""
    src_secret_key = ""
    src_bucket = ""
    src_initial_path = "upload/2024/2024-05-17/"

    dst_access_key = ""
    dst_secret_key = ""
    dst_bucket = ""
    dst_initial_path = "download/2024/2024-05-17/"

    # Set the desired multipart threshold value (50MB)
    mb = 1024 ** 2
    min_size_of_file_for_multipart_threshold = 50

    workers = 20
    botocore_config = botocore.config.Config(max_pool_connections=workers)

    transfer_config = s3transfer.TransferConfig(
        use_threads=True,
        max_concurrency=workers,
        multipart_threshold=min_size_of_file_for_multipart_threshold * mb
    )

    src = boto3.client(
        service_name=service,
        verify=verify_bool,
        endpoint_url=endpoint_url_with_https,
        aws_access_key_id=src_access_key,
        aws_secret_access_key=src_secret_key,
        config=botocore_config
    )

    dst = boto3.client(
        service_name=service,
        verify=verify_bool,
        endpoint_url=endpoint_url_with_https,
        aws_access_key_id=dst_access_key,
        aws_secret_access_key=dst_secret_key,
        config=botocore_config
    )  

    paginator = src.get_paginator('list_objects')
    response_iterator = paginator.paginate(
        Bucket=src_bucket,
        Prefix=src_initial_path,
        PaginationConfig={
            'PageSize': 1000,
            'MaxItems': 1000
        }
    )

    objs = response_iterator.build_full_result()['Contents']
    keys_to_copy = [o['Key'] for o in objs]  # or use a generator (o['Key'] for o in objs)

    s3t = s3transfer.create_transfer_manager(dst, transfer_config)

    for key in keys_to_copy:
        print(key)
        copy_source = {
            'Bucket': src_bucket,
            'Key': key
        }

        copy_to = {
            'Bucket': dst_bucket,
            'Key': dst_initial_path + key.rsplit('/', 1)[-1]
       }

        src_response = src.get_object(Bucket=src_bucket, Key=key)
        src_data = src_response['Body'].read()
        future = s3t.upload(
            fileobj=BytesIO(src_data),
            bucket=dst_bucket,
            key=dst_initial_path + key.rsplit('/', 1)[-1]
        )
        future.result()
        s3t.shutdown()  # wait for all the upload tasks to finish

raw_from_prod_to_nonprod()

代码问题分析 #

1. 本地中转冗余：先调用get_object把整个文件读入内存再上传，绕开了S3的服务器端复制能力，所有数据需经过本地网络中转，大幅增加传输耗时。
2. 传输管理器滥用：循环内每次调用s3t.shutdown()，导致传输管理器反复销毁重建，无法复用并发连接池，完全浪费多线程配置的优势。
3. 内存负载过高：src_response['Body'].read()将整个文件加载到内存，大文件会引发内存压力，拖慢传输速度。
4. 同步阻塞上传：每次上传都调用future.result()等待单个文件完成，无法实现多文件并行传输。

优化方案 #

核心优化：使用S3服务器端复制 #

服务器端复制（Server-Side Copy）是同S3兼容存储间最快的传输方式，数据直接在存储服务之间流转，无需本地中转。只要目标凭证对源桶有GetObject权限，源凭证对源桶有读权限，即可直接触发复制。

修正后的代码 #

def raw_from_prod_to_nonprod():
    import boto3
    import boto3.s3.transfer as s3transfer
    import botocore
    import urllib3

    urllib3.disable_warnings()

    service = "s3"
    verify_bool = False
    endpoint_url_with_https = "https://xx-xxxx.xx.xxx.xx:1234"

    src_access_key = ""
    src_secret_key = ""
    src_bucket = ""
    src_initial_path = "upload/2024/2024-05-17/"

    dst_access_key = ""
    dst_secret_key = ""
    dst_bucket = ""
    dst_initial_path = "download/2024/2024-05-17/"

    # 调整多部分配置：阈值50MB，分块大小16MB（减少请求次数）
    mb = 1024 ** 2
    multipart_threshold = 50 * mb
    multipart_chunksize = 16 * mb

    # 并发配置：根据网络带宽调整，建议30-50
    workers = 30
    botocore_config = botocore.config.Config(
        max_pool_connections=workers,
        connect_timeout=10,
        read_timeout=30
    )

    transfer_config = s3transfer.TransferConfig(
        use_threads=True,
        max_concurrency=workers,
        multipart_threshold=multipart_threshold,
        multipart_chunksize=multipart_chunksize
    )

    # 初始化客户端
    src = boto3.client(
        service_name=service,
        verify=verify_bool,
        endpoint_url=endpoint_url_with_https,
        aws_access_key_id=src_access_key,
        aws_secret_access_key=src_secret_key,
        config=botocore_config
    )

    dst = boto3.client(
        service_name=service,
        verify=verify_bool,
        endpoint_url=endpoint_url_with_https,
        aws_access_key_id=dst_access_key,
        aws_secret_access_key=dst_secret_key,
        config=botocore_config
    )  

    # 分页获取源桶对象
    paginator = src.get_paginator('list_objects')
    response_iterator = paginator.paginate(
        Bucket=src_bucket,
        Prefix=src_initial_path,
        PaginationConfig={'PageSize': 1000}
    )

    objs = response_iterator.build_full_result()['Contents']
    keys_to_copy = [o['Key'] for o in objs]

    # 初始化传输管理器，仅在所有任务完成后关闭
    s3t = s3transfer.create_transfer_manager(dst, transfer_config)
    futures = []

    for key in keys_to_copy:
        print(key)
        dst_key = f"{dst_initial_path}{key.rsplit('/', 1)[-1]}"
        # 使用传输管理器的copy方法，直接触发服务器端复制
        future = s3t.copy(
            copy_source={'Bucket': src_bucket, 'Key': key},
            bucket=dst_bucket,
            key=dst_key
        )
        futures.append(future)

    # 等待所有复制任务完成
    for future in futures:
        future.result()

    # 所有任务完成后关闭传输管理器
    s3t.shutdown()

raw_from_prod_to_nonprod()

额外优化建议 #

1. 权限配置检查：确保目标凭证拥有源桶的GetObject权限，源桶的Bucket Policy需允许目标账户/凭证访问（跨账户场景）。
2. 调整分块大小：根据文件大小和网络带宽调整multipart_chunksize，大带宽环境下可增大到32MB或64MB，减少HTTP请求次数。
3. 批量异步处理：文件数量较多时，可优化为批量提交任务，避免一次性加载所有对象到内存。
4. 启用TCP复用：通过botocore的tcp_keepalive=True配置，保持长连接，减少连接建立开销。
5. 网络环境优化：确保本地网络到S3端点的带宽充足，优先选择同区域存储进行传输。

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