如何高效列出指定时间后更新的Amazon S3对象？

问题根源 #

你当前的代码是客户端侧过滤：先用list_objects_v2全量拉取指定前缀下的所有对象，再通过JMESPath筛选符合时间条件的对象。这就导致不管时间范围多大，都要拉取全部100万+对象，所以2010和2023的时间戳耗时一致。

要解决效率问题，核心是减少服务端返回的数据量，而非客户端过滤。以下是针对CDC场景+Glue Bookmark的可行方案：

方案1：按时间前缀命名对象（最优） #

如果能控制S3对象的键名格式（比如yyyy/mm/dd/hh/mm/object-id.json），直接通过Prefix参数精准定位目标时间范围的前缀，让S3只返回该前缀下的对象：

• 比如要筛选2023-07-25之后的对象，就遍历2023/07/25/、2023/07/26/...等前缀，逐个调用list_objects_v2。
• 结合Glue Bookmark：记录上次处理到的时间前缀，下次直接从该前缀开始遍历，完全避免全量拉取。

方案2：使用S3 Inventory批量查询元数据 #

如果无法修改对象键名，用S3 Inventory定期生成包含对象LastModified、Key等元数据的清单文件（支持CSV/Parquet格式）：

1. 在S3控制台配置Inventory，指定生成频率（比如每天）和存储位置。
2. 每次需要筛选时，读取最新的Inventory文件，用Pandas/Spark直接过滤出符合时间条件的对象键。
3. 结合Glue Bookmark：记录上次处理的Inventory生成时间，只读取该时间之后的Inventory文件，避免重复处理。

这种方式比全量list_objects_v2高效得多，尤其适合百万级对象的场景。

方案3：EventBridge+DynamoDB实时跟踪对象变化 #

针对持续CDC的场景，用S3事件通知实时捕获对象的新增/修改操作：

1. 配置S3桶的事件通知，将PutObject、CopyObject等事件发送到EventBridge。
2. 编写EventBridge规则，把事件中的对象键、LastModified时间写入DynamoDB表。
3. 每次处理时，直接从DynamoDB中查询指定时间后的对象键，无需调用list_objects_v2。
4. 结合Glue Bookmark：记录上次查询的时间戳，下次只查询该时间之后的条目。

该方案能实现近实时的对象跟踪，完全避免全量列举。

方案4：优化现有代码+Glue Bookmark（无额外依赖） #

如果以上方案无法落地，可优化现有代码，结合Glue Bookmark减少无效遍历：

注意：S3返回的对象是按键名排序，而非LastModified时间。只有当对象键名和修改时间正相关（比如键名包含时间戳）时，才能提前终止遍历。

改造后的代码 #

import boto3
from datetime import datetime

def get_objects(bucket, prefix, last_processed_time):
    s3 = boto3.client("s3")
    paginator = s3.get_paginator('list_objects_v2')
    # 转换时间格式为S3返回的datetime类型
    target_time = datetime.fromisoformat(last_processed_time.replace(" ", "T"))
    count = 0
    latest_processed_time = target_time

    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        if 'Contents' not in page:
            continue
        for obj in page['Contents']:
            obj_modified = obj['LastModified']
            if obj_modified >= target_time:
                count += 1
                print(f"{count}-->{obj['Key']}")
                # 更新本次处理的最新时间
                if obj_modified > latest_processed_time:
                    latest_processed_time = obj_modified
            else:
                # 仅当键名与修改时间正相关时，可break终止当前页遍历
                # 若键名无时间规律，需删除此break，继续遍历所有对象
                break
    
    print(f"count is {count}")
    # 返回最新时间，用于更新Glue Bookmark
    return latest_processed_time.isoformat().replace("T", " ")

Glue Bookmark集成 #

在Glue Job中，通过以下逻辑读取和更新Bookmark：

from awsglue.context import GlueContext
from pyspark.context import SparkContext
import sys

sc = SparkContext()
glue_context = GlueContext(sc)
args = glue_context.get_resolved_options(sys.argv, ['JOB_NAME'])

# 读取上次的Bookmark
job_bookmark = glue_context.get_job_bookmark()
if job_bookmark:
    last_processed_time = job_bookmark.get('last_processed_time', '2010-07-25 00:00:00+00:00')
else:
    last_processed_time = '2010-07-25 00:00:00+00:00'

# 调用处理函数
latest_time = get_objects('your-bucket', 'your-prefix', last_processed_time)

# 更新Bookmark
glue_context.update_job_bookmark(
    job_name=args['JOB_NAME'],
    bookmark={
        'last_processed_time': latest_time
    }
)

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