如何用Boto并行列出S3中含10万+对象的前缀/目录？

默认的S3分页查询依赖前一页返回的NextKeyMarker才能发起下一页请求，确实没法直接并行所有分页请求。下面给两种可行的并行方案：

一、前缀拆分法（你提到的字符范围拆分思路） #

S3对象键按字典序存储，我们可以把原前缀拆分成多个子前缀，每个子前缀独立发起查询，完全并行处理。比如原前缀是my_prefix/，可以按对象键的首字符（数字、大小写字母）拆分成多个子前缀，每个子前缀对应一批对象。

代码示例 #

import boto3
from concurrent.futures import ThreadPoolExecutor

def fetch_objects_for_prefix(bucket, prefix):
    """获取指定前缀下的所有对象"""
    s3_client = boto3.client('s3')
    paginator = s3_client.get_paginator('list_objects_v2')
    objects = []
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        objects.extend(page.get('Contents', []))
    return objects

def main():
    bucket_name = '你的桶名'
    base_prefix = '你的前缀/'
    
    # 生成要并行扫描的子前缀列表
    sub_prefixes = []
    # 覆盖数字开头的对象
    for num in '0123456789':
        sub_prefixes.append(f"{base_prefix}{num}")
    # 覆盖小写字母开头的对象
    for char in 'abcdefghijklmnopqrstuvwxyz':
        sub_prefixes.append(f"{base_prefix}{char}")
    # 如果有大写字母开头的对象，加上这部分
    # for char in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ':
    #     sub_prefixes.append(f"{base_prefix}{char}")
    
    # 用线程池并行处理
    all_objects = []
    # 线程数根据S3请求配额调整，默认每秒1500次请求，别设太满
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 提交所有子前缀的查询任务
        futures = [executor.submit(fetch_objects_for_prefix, bucket_name, p) for p in sub_prefixes]
        # 收集所有结果
        for future in futures:
            all_objects.extend(future.result())
    
    print(f"总计获取到 {len(all_objects)} 个对象")

if __name__ == "__main__":
    main()

二、字典序范围划分法 #

如果不想拆分子前缀，可以先获取一批对象键作为分界点，把整个前缀的对象范围拆分成多个区间，每个区间用StartAfter和EndingBefore限定，并行查询每个区间的对象。这种方法适合对象键分布不均匀的场景。

代码示例 #

import boto3
from concurrent.futures import ThreadPoolExecutor

def fetch_objects_in_range(bucket, prefix, start_after=None, ending_before=None):
    """获取指定前缀下某个字典序范围内的对象"""
    s3_client = boto3.client('s3')
    paginator = s3_client.get_paginator('list_objects_v2')
    paginate_args = {'Bucket': bucket, 'Prefix': prefix}
    if start_after:
        paginate_args['StartAfter'] = start_after
    if ending_before:
        paginate_args['EndingBefore'] = ending_before
    
    objects = []
    for page in paginator.paginate(**paginate_args):
        objects.extend(page.get('Contents', []))
    return objects

def main():
    bucket_name = '你的桶名'
    base_prefix = '你的前缀/'
    
    # 先获取少量对象作为分界点
    s3_client = boto3.client('s3')
    initial_response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=base_prefix, MaxKeys=10)
    if not initial_response.get('Contents'):
        print("前缀下无对象")
        return
    
    # 提取分界键（去掉首尾，避免重复查询）
    split_keys = [obj['Key'] for obj in initial_response['Contents']][1:-1]
    if not split_keys:
        # 如果分界键不足，直接全量查询
        objects = fetch_objects_in_range(bucket_name, base_prefix)
        print(f"总计获取到 {len(objects)} 个对象")
        return
    
    # 构造查询区间
    query_ranges = []
    # 第一个区间：从前缀开始到第一个分界键之前
    query_ranges.append({'start_after': None, 'ending_before': split_keys[0]})
    # 中间区间
    for i in range(len(split_keys)-1):
        query_ranges.append({'start_after': split_keys[i], 'ending_before': split_keys[i+1]})
    # 最后一个区间：从最后一个分界键到结尾
    query_ranges.append({'start_after': split_keys[-1], 'ending_before': None})
    
    # 并行查询所有区间
    all_objects = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(fetch_objects_in_range, bucket_name, base_prefix, r['start_after'], r['ending_before']) for r in query_ranges]
        for future in futures:
            all_objects.extend(future.result())
    
    print(f"总计获取到 {len(all_objects)} 个对象")

if __name__ == "__main__":
    main()

注意事项 #

• 线程池的max_workers不要设置过大，避免触发S3的请求频率限制（默认每秒允许1500次请求），可根据实际情况调整。
• 如果对象键分布极不均匀（比如90%的对象都以'a'开头），前缀拆分法会出现负载不均的情况，此时优先用范围划分法。
• 两种方案都摆脱了对NextKeyMarker的依赖，每个查询任务完全独立，实现真正的并行。

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