如何通过boto3筛选S3存储桶中特定存储类的对象？

核心说明 #

AWS S3的ListObjectsV2 API本身不支持通过请求参数直接过滤存储类，无论使用Java SDK还是boto3都一样——你提到的Java方案大概率是借助了S3 Inventory或其他间接方式实现的过滤，而非直接在List请求中添加过滤条件。针对你的场景，提供以下几种解决思路：

1. 长期最优方案：使用S3 Inventory生成对象清单 #

如果需要频繁处理这类筛选需求，S3 Inventory是最高效的方式：

• 配置S3 Inventory定期（每日/每周）生成包含对象存储类的CSV/Parquet格式清单
• 直接从清单文件中筛选出StorageClass不为GLACIER/DEEP_ARCHIVE的对象Key
• 批量下载筛选后的对象，无需遍历整个前缀下的5万条记录

操作要点 #

• 在S3控制台为目标桶开启Inventory，选择包含StorageClass字段
• 当Inventory生成后，读取清单文件（通常存储在指定的目标桶中），用pandas或普通文件读取工具筛选符合条件的Key
• 直接批量下载这些Key对应的对象

2. 短期优化：提升现有遍历效率 #

如果只是临时需求，可通过以下方式优化现有代码的执行速度：

2.1 减少返回字段，降低数据传输量 #

使用ListObjectsV2的Projection参数，只请求需要的Key和StorageClass字段，减少每个分页的数据包大小：

paginator = s3.get_paginator("list_objects_v2")
pages = paginator.paginate(
    Bucket=aws_s3_credentials["s3_logs_bucket_name"],
    Prefix=s3_folder_name,
    Projection="keys"  # 仅返回Key与StorageClass，默认返回所有字段
)

2.2 多线程并行下载 #

由于本地过滤后需要下载的对象占比约30%，使用多线程可以大幅提升下载速度，避免单线程串行等待：

from concurrent.futures import ThreadPoolExecutor

def download_single_object(s3_client, bucket_name, obj_key, local_path, logger):
    try:
        s3_client.download_file(bucket_name, obj_key, local_path)
    except Exception as e:
        logger.error(f"下载失败 {obj_key}: {str(e)}")

# 遍历到符合条件的对象时，提交到线程池
with ThreadPoolExecutor(max_workers=10) as executor:
    for page in pages:
        for obj in page["Contents"]:
            if obj["StorageClass"] in ["GLACIER", "DEEP_ARCHIVE"]:
                continue
            local_file_path = local_download_folder / obj["Key"].split("/")[-1]
            executor.submit(
                download_single_object,
                s3,
                aws_s3_credentials["s3_logs_bucket_name"],
                obj["Key"],
                str(local_file_path),
                logger
            )

2.3 复用S3客户端 #

当前代码在每个log_file_code循环中都创建新的S3客户端，会产生不必要的开销，建议将客户端创建移到循环外部：

# 移到循环外，复用客户端
s3 = boto3.client(
    "s3",
    aws_access_key_id=aws_s3_credentials["s3_user_id"],
    aws_secret_access_key=aws_s3_credentials["s3_user_secret"],
)

for log_file_code in log_file_codes:
    logger.debug(f"正在下载日志文件: {log_file_code}")
    s3_folder_name = str(log_file_code).zfill(4)
    # 后续代码复用已创建的s3客户端

3. 备选方案：使用S3 Batch Operations（批量操作） #

如果需要一次性处理大量对象，可使用S3 Batch Operations：

• 基于S3 Inventory清单创建任务，筛选出非归档类对象
• 直接通过Batch Operations将这些对象下载到指定的EC2实例或本地（需借助AWS DataSync或其他传输工具）

优化后的完整代码示例 #

import logging
from typing import Dict, List
from pathlib import Path
import os
import boto3
from concurrent.futures import ThreadPoolExecutor

def download_single_object(
    logger: logging.Logger,
    s3_client,
    bucket_name: str,
    obj_key: str,
    local_file_path: str
):
    try:
        s3_client.download_file(bucket_name, obj_key, local_file_path)
        logger.debug(f"成功下载 {obj_key}")
    except Exception as e:
        logger.error(f"下载失败 {obj_key}: {str(e)}")

def download_log_files(
    logger: logging.Logger,
    log_file_codes: List[int],
    aws_s3_credentials: Dict[str, str],
    local_download_dir: str,
):
    """从S3存储桶下载所有日志文件到本地磁盘。

    参数:
        logger (logging.Logger): 日志记录器。
        log_file_codes (List[int]): 日志文件编码列表。
        aws_s3_credentials (Dict[str, str]): AWS S3认证信息。
        local_download_dir (str): 本地下载目录路径。
    """
    # 复用S3客户端，避免重复创建
    s3 = boto3.client(
        "s3",
        aws_access_key_id=aws_s3_credentials["s3_user_id"],
        aws_secret_access_key=aws_s3_credentials["s3_user_secret"],
    )
    bucket_name = aws_s3_credentials["s3_logs_bucket_name"]

    for log_file_code in log_file_codes:
        logger.debug(f"正在下载日志文件: {log_file_code}")
        s3_folder_name = str(log_file_code).zfill(4)

        paginator = s3.get_paginator("list_objects_v2")
        pages = paginator.paginate(
            Bucket=bucket_name,
            Prefix=s3_folder_name,
            Projection="keys"  # 仅返回必要字段，减少数据传输
        )

        local_download_folder = Path(f"{local_download_dir}/{s3_folder_name}")
        if not os.path.exists(local_download_folder):
            os.makedirs(local_download_folder)

        # 多线程并行下载
        with ThreadPoolExecutor(max_workers=10) as executor:
            for page in pages:
                # 处理空页面的情况
                if "Contents" not in page:
                    continue
                for obj in page["Contents"]:
                    if obj["StorageClass"] in ["GLACIER", "DEEP_ARCHIVE"]:
                        continue
                    local_file_path = local_download_folder / obj["Key"].split("/")[-1]
                    executor.submit(
                        download_single_object,
                        logger,
                        s3,
                        bucket_name,
                        obj["Key"],
                        str(local_file_path)
                    )

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