如何使用boto3递归复制同S3存储桶跨目录海量文件

S3本身不存在真实的目录结构，所有对象都以Key作为唯一标识，路径s3://mybucket/abc/process/本质是前缀为abc/process/的所有对象集合。同桶内调用copy接口是S3服务端直接执行复制，不会占用本地公网/内网带宽，单账号默认单Region复制带宽可达10Gbps+，完全适配50TB级别的海量文件传输场景。
你之前使用aws s3 sync仅拷贝了大文件，大概率是默认并发数过低、小文件量级过大时列表分页处理异常、或请求超时导致进程提前退出，以下boto3实现方案可以规避这类问题。

核心实现逻辑 #

• 基于S3 list接口的分页器（Paginator）遍历源前缀下的所有对象，自动处理单次list请求最多返回1000条结果的截断问题，天然支持递归遍历所有嵌套层级的对象，无需手动处理目录逻辑
• 对每个遍历到的对象，将Key中的源前缀替换为目标前缀，构造服务端复制请求
• 配置自动重试、断点续传逻辑，避免脚本中断后重复传输
• 用线程池并发提交复制请求，搭配大文件自动分片复制配置，最大化传输效率

可直接运行的代码 #

import boto3
from boto3.s3.transfer import TransferConfig
from concurrent.futures import ThreadPoolExecutor, as_completed
from botocore.config import Config

# 基础配置
SOURCE_BUCKET = "mybucket"
SOURCE_PREFIX = "abc/process/"  # 源路径前缀，末尾必须带斜杠
DEST_PREFIX = "cde/process/"    # 目标路径前缀，末尾必须带斜杠
CONCURRENCY = 20  # 并发请求数，建议设置10-30，过高容易触发S3流控

# S3客户端配置，增加异常重试
s3_client = boto3.client(
    "s3",
    config=Config(
        retries={"max_attempts": 10, "mode": "standard"},
        connect_timeout=10,
        read_timeout=30
    )
)

# 传输配置，自动处理大文件分片复制
transfer_config = TransferConfig(
    multipart_threshold=256 * 1024 * 1024,  # 超过256MB的文件自动走分片复制
    max_concurrency=10,
    multipart_chunksize=256 * 1024 * 1024,
    use_threads=True
)

def copy_single_object(source_key: str):
    # 跳过根路径空对象
    if source_key == SOURCE_PREFIX:
        return
    # 生成目标对象Key
    dest_key = source_key.replace(SOURCE_PREFIX, DEST_PREFIX, 1)
    # 检查目标对象是否已存在，存在则跳过实现断点续传
    try:
        s3_client.head_object(Bucket=SOURCE_BUCKET, Key=dest_key)
        print(f"跳过已存在对象: {dest_key}")
        return
    except s3_client.exceptions.ClientError as e:
        if e.response["Error"]["Code"] != "404":
            print(f"检查对象{dest_key}状态失败: {str(e)}")
            return
    # 发起服务端复制请求
    try:
        s3_client.copy(
            {"Bucket": SOURCE_BUCKET, "Key": source_key},
            SOURCE_BUCKET,
            dest_key,
            Config=transfer_config
        )
        print(f"复制完成: {source_key} -> {dest_key}")
    except Exception as e:
        print(f"复制失败{source_key}: {str(e)}")
        raise

if __name__ == "__main__":
    # 分页遍历所有源前缀下的对象
    paginator = s3_client.get_paginator("list_objects_v2")
    all_source_keys = []
    for page in paginator.paginate(Bucket=SOURCE_BUCKET, Prefix=SOURCE_PREFIX):
        if "Contents" not in page:
            continue
        for obj in page["Contents"]:
            all_source_keys.append(obj["Key"])
    print(f"扫描完成，共找到{len(all_source_keys)}个待复制对象")

    # 并发执行复制任务
    with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor:
        task_list = [executor.submit(copy_single_object, key) for key in all_source_keys]
        for task in as_completed(task_list):
            try:
                task.result()
            except Exception as e:
                print(f"任务异常: {str(e)}")

海量场景注意事项 #

• 全程无本地数据中转：同桶复制所有数据拷贝动作都在S3服务端完成，运行脚本的设备只需要发送控制请求，不会占用下载/上传带宽，哪怕用本地个人电脑运行也可以跑满S3服务端带宽
• 不要盲目调高并发数：S3默认单账号单Region的PUT/COPY请求配额是3500次/秒，并发数设置超过30后容易触发503 SlowDown流控，反而降低传输效率
• 权限要求：运行脚本的身份需要拥有源路径的s3:GetObject权限、目标路径的s3:PutObject权限，以及桶的s3:ListBucket权限
• 成本说明：同桶复制不会产生跨区域流量费，仅会产生COPY请求费用和目标对象的存储费用，成本极低

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