如何使用boto3将大文件分块上传至不同区域的S3存储桶？

我刚好有过类似的多区域S3批量上传需求，结合boto3的multipart API，给你整理了一套可以直接复用的方案，涵盖多区域客户端管理、大文件分块、并行上传这几个核心部分。

前置准备 #

• 确保安装了boto3：pip install boto3
• 配置AWS凭证：可以通过环境变量AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY，或者~/.aws/credentials文件，注意你的凭证需要拥有所有目标S3桶的s3:PutObject、s3:CreateMultipartUpload、s3:UploadPart、s3:CompleteMultipartUpload、s3:AbortMultipartUpload权限。

核心实现代码 #

首先，我们需要为每个目标区域创建独立的S3客户端（因为S3客户端是区域绑定的，跨区域用同一个客户端会有性能问题甚至报错）。然后封装分块上传的通用函数，最后实现多桶的并行上传。

import boto3
import os
from multiprocessing.pool import ThreadPool
from typing import Dict, List

# 配置项
TARGET_BUCKETS = [
    {"bucket_name": "bucket-us-east-1", "region": "us-east-1"},
    {"bucket_name": "bucket-eu-west-1", "region": "eu-west-1"},
    {"bucket_name": "bucket-ap-southeast-1", "region": "ap-southeast-1"}
]  # 替换成你的目标桶和对应区域
FILE_PATH = "/path/to/your/500gb-file"  # 替换成你的大文件路径
CHUNK_SIZE = 16 * 1024 * 1024  # 16MB分块，可根据网络调整（建议5MB-5GB）

def create_s3_client(region: str) -> boto3.client:
    """创建指定区域的S3客户端"""
    return boto3.client(
        "s3",
        region_name=region
        # 如果凭证不是默认配置，可以在这里指定aws_access_key_id和aws_secret_access_key
    )

def upload_part(s3_client, bucket_name, upload_id, part_number, chunk, key):
    """上传单个分块，返回分块ETag"""
    response = s3_client.upload_part(
        Bucket=bucket_name,
        Key=key,
        PartNumber=part_number,
        UploadId=upload_id,
        Body=chunk
    )
    return {"PartNumber": part_number, "ETag": response["ETag"]}

def multipart_upload_to_s3(s3_client, bucket_name, file_path, key=None):
    """将单个文件分块上传到指定S3桶"""
    if not key:
        key = os.path.basename(file_path)  # 默认用文件名作为S3对象键

    # 1. 初始化分块上传
    upload_response = s3_client.create_multipart_upload(
        Bucket=bucket_name,
        Key=key
    )
    upload_id = upload_response["UploadId"]

    parts = []
    file_size = os.path.getsize(file_path)
    part_number = 1

    try:
        # 2. 读取文件分块并上传
        with open(file_path, "rb") as f:
            while True:
                chunk = f.read(CHUNK_SIZE)
                if not chunk:
                    break
                # 这里可以用线程池并行上传分块，提升单桶上传速度
                part = upload_part(s3_client, bucket_name, upload_id, part_number, chunk, key)
                parts.append(part)
                part_number += 1

        # 3. 完成分块上传
        s3_client.complete_multipart_upload(
            Bucket=bucket_name,
            Key=key,
            UploadId=upload_id,
            MultipartUpload={"Parts": parts}
        )
        print(f"成功上传文件到 {bucket_name}/{key}")
    except Exception as e:
        # 上传失败时终止分块上传，避免产生无效的分块占用空间
        s3_client.abort_multipart_upload(
            Bucket=bucket_name,
            Key=key,
            UploadId=upload_id
        )
        print(f"上传到 {bucket_name} 失败，已终止分块上传: {str(e)}")
        raise

def batch_upload_to_multiple_regions():
    """批量上传文件到多区域S3桶"""
    # 为每个桶创建对应的客户端
    bucket_clients = [
        (create_s3_client(bucket["region"]), bucket["bucket_name"])
        for bucket in TARGET_BUCKETS
    ]

    # 用线程池并行处理多桶上传（IO密集型任务，线程比进程更高效）
    with ThreadPool(len(bucket_clients)) as pool:
        # 给每个桶分配上传任务
        tasks = [
            pool.apply_async(multipart_upload_to_s3, args=(client, bucket_name, FILE_PATH))
            for client, bucket_name in bucket_clients
        ]
        # 等待所有任务完成
        for task in tasks:
            task.get()

if __name__ == "__main__":
    batch_upload_to_multiple_regions()

关键注意事项 #

• 分块大小选择：CHUNK_SIZE建议设置在8MB-32MB之间，太小会增加API调用次数，太大可能导致单分块上传超时。如果网络不稳定，可以适当调小。
• 并行策略：
  • 单桶内的分块可以再用线程池并行上传（代码里的upload_part可以改成线程池调用，进一步提升单桶速度）；
  • 多桶之间用线程池并行，因为每个桶的上传是独立的IO操作，线程切换开销更低。
• 断点续传：如果需要断点续传，可以记录已上传的分块ETag和upload_id，下次上传时先调用list_parts接口获取已上传的分块，跳过重复上传。
• 权限与区域：确保你的AWS凭证在目标区域有足够权限，并且S3客户端是对应区域的，跨区域用同一个客户端会导致请求路由到错误的端点，影响性能甚至失败。
• 错误处理：代码里已经包含了上传失败时终止分块上传的逻辑，避免S3中残留未完成的分块（未完成的分块会占用存储空间，且不会自动删除，需要手动清理或设置生命周期规则）。

扩展优化 #

• 可以添加进度条监控（用tqdm库），实时查看上传进度；
• 针对超大文件，可以加入分块校验（比如计算MD5），确保上传的分块和源文件一致；
• 如果需要上传多个文件，可以扩展代码为批量文件上传，遍历本地目录后分配任务。

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