AWS S3 Bucket同步优化：批量对比删文件及列表填充提速需求

问题背景 #

需要对比BucketA和BucketB，删除BucketB中存在但BucketA没有的同名文件，两个桶各有300-400万个文件。原代码在获取文件列表时速度极慢，需要优化这一步骤。

原实现的核心瓶颈 #

原代码存在两个关键低效点：

• 分页PageSize设置为2，导致需要发起数百万次API请求，严重拖慢速度（S3 list_objects_v2最大支持一次返回1000个对象）；
• 单线程遍历所有分页结果，且先将所有Key存入列表再转换为集合，既浪费内存又增加额外处理开销。

优化方案 #

1. 基础优化：调大分页参数+直接用集合存储 #

将PageSize改为最大值1000，减少API请求次数；同时直接把Key存入集合，省去列表转集合的步骤。

import boto3

def get_bucket_keys(bucket_name):
    s3_client = boto3.client('s3')
    paginator = s3_client.get_paginator('list_objects_v2')
    # 使用最大PageSize，大幅减少API调用次数
    page_iterator = paginator.paginate(Bucket=bucket_name, PaginationConfig={'PageSize': 1000})
    keys_set = set()
    for page in page_iterator:
        if 'Contents' in page:
            # 批量更新集合，比循环append更高效
            keys_set.update(obj['Key'] for obj in page['Contents'])
    return keys_set

# 获取两个桶的Key集合
bucket_a_keys = get_bucket_keys('BucketA')
bucket_b_keys = get_bucket_keys('BucketB')

# 计算BucketB独有的文件
to_delete = bucket_b_keys - bucket_a_keys

# 批量删除工具函数
def batch_delete_from_bucket(bucket_name, keys, batch_size=1000):
    s3_client = boto3.client('s3')
    batch = []
    for key in keys:
        batch.append({'Key': key})
        if len(batch) >= batch_size:
            s3_client.delete_objects(
                Bucket=bucket_name,
                Delete={'Objects': batch}
            )
            batch = []
    # 处理剩余不足一批的文件
    if batch:
        s3_client.delete_objects(
            Bucket=bucket_name,
            Delete={'Objects': batch}
        )

# 执行删除
batch_delete_from_bucket('BucketB', to_delete)

2. 进阶优化：多线程并行获取文件列表 #

由于S3列表查询是网络IO密集型操作，使用多线程并行处理分页请求，可进一步提升速度。

import boto3
from concurrent.futures import ThreadPoolExecutor

def fetch_single_page(page_params):
    """单个分页查询任务"""
    s3_client = boto3.client('s3')
    bucket_name, continuation_token = page_params
    request_args = {'Bucket': bucket_name, 'MaxKeys': 1000}
    if continuation_token:
        request_args['ContinuationToken'] = continuation_token
    response = s3_client.list_objects_v2(**request_args)
    keys = [obj['Key'] for obj in response.get('Contents', [])]
    next_token = response.get('NextContinuationToken')
    return keys, next_token

def get_bucket_keys_concurrent(bucket_name, max_workers=10):
    """多线程获取桶内所有Key"""
    s3_client = boto3.client('s3')
    # 初始化第一个分页
    initial_response = s3_client.list_objects_v2(Bucket=bucket_name, MaxKeys=1000)
    all_keys = set(obj['Key'] for obj in initial_response.get('Contents', []))
    next_token = initial_response.get('NextContinuationToken')
    
    # 提交所有分页任务
    tasks = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        while next_token:
            tasks.append(executor.submit(fetch_single_page, (bucket_name, next_token)))
            # 获取下一个分页的token，避免重复提交
            temp_resp = s3_client.list_objects_v2(Bucket=bucket_name, ContinuationToken=next_token, MaxKeys=1000)
            next_token = temp_resp.get('NextContinuationToken')
        
        # 汇总所有任务结果
        for future in tasks:
            page_keys, _ = future.result()
            all_keys.update(page_keys)
    
    return all_keys

# 使用多线程版本获取Key集合
bucket_a_keys = get_bucket_keys_concurrent('BucketA')
bucket_b_keys = get_bucket_keys_concurrent('BucketB')
to_delete = bucket_b_keys - bucket_a_keys
batch_delete_from_bucket('BucketB', to_delete)

3. 终极优化：使用S3 Inventory #

对于百万级文件的桶，推荐使用S3 Inventory功能。它会定期（每日或每周）自动生成桶内对象的清单文件（CSV/ORC/Parquet格式），直接下载清单文件处理，无需调用大量list_objects_v2 API，效率提升显著。

操作步骤：

1. 在S3控制台为BucketA和BucketB开启Inventory，配置清单文件输出到一个单独的S3桶，选择CSV格式并包含对象Key字段；
2. 等待Inventory生成最新的清单文件；
3. 读取清单文件中的Key并对比，执行批量删除。

示例代码：

import csv
import boto3

def get_inventory_keys(inventory_bucket, inventory_file_path):
    """从S3 Inventory CSV文件中读取所有Key"""
    s3_client = boto3.client('s3')
    temp_file = '/tmp/s3_inventory.csv'
    # 下载清单文件到本地临时路径
    s3_client.download_file(inventory_bucket, inventory_file_path, temp_file)
    
    keys_set = set()
    with open(temp_file, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            keys_set.add(row['Key'])
    return keys_set

# 替换为你的Inventory存储桶和文件路径
bucket_a_keys = get_inventory_keys('my-inventory-bucket', 'BucketA-inventory/latest.csv')
bucket_b_keys = get_inventory_keys('my-inventory-bucket', 'BucketB-inventory/latest.csv')

to_delete = bucket_b_keys - bucket_a_keys
batch_delete_from_bucket('BucketB', to_delete)

方案选择建议 #

• 若桶内文件数量在百万级且对实时性要求不高，优先选S3 Inventory，效率最高且资源消耗最少；
• 若需要实时对比，使用多线程+大PageSize的方案，比单线程快数倍；
• 绝对避免设置过小的PageSize，这是拖慢速度的核心原因之一。

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