1. 项目概述：为什么一个资深运维/开发会把Boto3当“日常工具箱”用

我干AWS相关工作快八年了，从最早手动点控制台删EC2实例、清S3桶，到后来写Shell脚本调AWS CLI，再到如今几乎每个自动化任务的第一行代码都是import boto3。这不是跟风，是被现实逼出来的——你不可能靠人眼盯住几十个账户里上百个S3桶的生命周期策略是否生效，也不可能在凌晨三点手动关停测试环境里漏关的RDS实例来止损。Boto3不是什么高大上的“云原生框架”，它就是AWS官方给开发者配的一把瑞士军刀：不花哨，但每一块刀片都磨得够锋利，而且说明书（文档）写得比大多数开源项目都厚实。关键词里只写了“AWS”，但实际用起来你会发现，Boto3真正解决的是“人”的问题：把重复、机械、易出错、必须守着时间点执行的操作，变成一段能放进CI/CD流水线、能定时触发、能加告警、能写进监控大盘的Python逻辑。它不替代架构设计，但能让架构落地时少踩80%的运营坑。适合谁？不是只给DevOps工程师看的——如果你是后端开发，要写一个自动归档日志到S3并设置30天过期的模块；如果你是数据工程师，需要每天凌晨扫描Glue Data Catalog里三个月没被查询过的表并打上标签；甚至如果你是安全合规岗，要定期导出IAM用户MFA启用状态生成审计报告——只要你的工作流里有“登录控制台→点几下→复制粘贴→再点几下”这个动作，Boto3就值得你花半天时间把它焊进自己的技能树里。它不难，但门槛不在语法，而在对AWS服务模型的理解深度。下面我就用自己踩过的真实坑、压测过的参数、上线跑了一年多的脚本，把Boto3怎么真正用起来讲透。

2. 核心设计思路：为什么选Boto3而不是其他方案

2.1 Boto3 vs AWS CLI：不是替代，而是升级

很多人第一次接触Boto3时会困惑：“我用aws cli不是挺好？写个for循环就能批量删S3对象。”这话没错，但问题出在“好”的定义上。CLI本质是命令行封装，它的强项是单次交互、快速验证；而Boto3的强项是状态感知和流程编排。举个真实例子：我们有个日志清理任务，要求“删除所有创建时间早于90天、且文件名包含error关键字的S3对象”。用CLI怎么做？aws s3api list-objects-v2 --bucket my-log-bucket --query 'Contents[?LastModified<2024-01-01&& contains(Key,error)].Key' --output text | xargs -I {} aws s3 rm s3://my-log-bucket/{}。表面看没问题，但实际运行时发现三处硬伤：第一，list-objects-v2默认只返回1000个对象，如果桶里有5000个文件，--query只筛前1000个，漏删4000个；第二，xargs并发删对象时，没有失败重试机制，网络抖动导致某个aws s3 rm失败，整个流程就卡死，还得人工介入；第三，无法记录每次删了多少个、耗时多少、哪些key删失败了——审计日志全靠猜。换成Boto3后，核心逻辑变成：

import boto3 from datetime import datetime, timedelta import logging s3 = boto3.client('s3', region_name='us-east-1') bucket = 'my-log-bucket' cutoff_date = datetime.now() - timedelta(days=90) deleted_count = 0 failed_keys = [] # 使用Paginator确保遍历全部对象 paginator = s3.get_paginator('list_objects_v2') for page in paginator.paginate(Bucket=bucket): for obj in page.get('Contents', []): if obj['LastModified'] < cutoff_date and 'error' in obj['Key']: try: s3.delete_object(Bucket=bucket, Key=obj['Key']) deleted_count += 1 logging.info(f"Deleted {obj['Key']}") except Exception as e: failed_keys.append((obj['Key'], str(e))) logging.error(f"Failed to delete {obj['Key']}: {e}") logging.info(f"Total deleted: {deleted_count}, Failed: {len(failed_keys)}")

这段代码解决了CLI方案的所有痛点：Paginator自动处理分页，try/except提供细粒度错误捕获，logging输出可审计的结构化日志。关键在于，Boto3让你能把“业务逻辑”（时间判断+关键字匹配）和“基础设施操作”（分页遍历+并发控制+错误处理）彻底解耦。CLI做不到这点，因为它没有“状态”概念——每次调用都是无状态的独立进程。

2.2 Boto3 vs Cloud Custodian：场景决定工具选型

原文提到Cloud Custodian用Boto3做资源清理，这没错，但必须说清楚：Cloud Custodian是Boto3之上的策略引擎，不是替代品。它的优势在于声明式配置（YAML写规则）、跨账户管理、内置大量合规检查模板。但代价是学习成本陡增，且灵活性受限。比如我们曾遇到一个需求：清理EBS快照时，要保留“最近7天内创建的快照”，同时“跳过所有Tag为BackupPolicy=retain的快照”。Cloud Custodian的age过滤器只能按创建时间绝对值筛选，无法动态计算“7天内”，而tag过滤器又不支持!=逻辑（它只支持=或present）。最后我们放弃Custodian，直接用Boto3写了一个200行的脚本，核心逻辑就两行：

# 获取当前时间戳用于计算7天窗口 now = datetime.now(timezone.utc) seven_days_ago = now - timedelta(days=7) # 遍历快照，双重条件过滤 ec2 = boto3.client('ec2', region_name='us-west-2') snapshots = ec2.describe_snapshots(OwnerIds=['self'])['Snapshots'] for snap in snapshots: # 条件1：创建时间在7天内 if snap['StartTime'] < seven_days_ago: continue # 条件2：跳过带特定Tag的快照 if any(tag['Key'] == 'BackupPolicy' and tag['Value'] == 'retain' for tag in snap.get('Tags', [])): continue # 执行删除 ec2.delete_snapshot(SnapshotId=snap['SnapshotId'])

这里的关键洞察是：Boto3给你的是原始API能力，你可以用任何Python逻辑组合条件；而Cloud Custodian给你的是预设的积木块，拼不出新形状就得绕路。所以我的经验法则是：如果任务规则简单、变化频繁、需要嵌入现有Python工程（比如Django后台定时任务），直接用Boto3；如果任务复杂、需跨多账户、要满足等保/ISO27001审计要求，再上Cloud Custodian这类策略引擎。两者不是二选一，而是Boto3打底，Custodian建模。

2.3 Boto3 vs Terraform：别混淆“管理”和“编排”

常有人问：“Terraform也能删资源，为啥不用？”这是根本性误解。Terraform是基础设施即代码（IaC）工具，核心目标是让云资源状态与代码定义保持一致。它删资源，是因为代码里删了那行resource "aws_s3_bucket"，属于“声明式驱逐”；而Boto3删资源，是因为业务逻辑触发了delete_object()，属于“命令式操作”。举个反例：我们有个批处理系统，每天生成临时S3桶存中间结果，处理完必须立刻清空。如果用Terraform管理，就得为每个临时桶写一份HCL代码，执行terraform apply——这完全违背了“临时”二字，且Terraform的state文件会疯狂膨胀。而Boto3一行boto3.resource('s3').Bucket('temp-bucket-20240520').objects.all().delete()就搞定。更关键的是，Terraform无法处理“数据层操作”：比如遍历S3桶里所有JSON文件，解析内容，根据字段值决定是否删除；或者调用Rekognition API分析S3图片，再根据识别结果打标签。这些是Boto3的主场，Terraform连门都摸不到。记住一句话：Terraform管“有没有”，Boto3管“怎么用”。

3. 核心细节解析：Boto3的底层机制与避坑指南

3.1 Session、Client、Resource：三层抽象到底怎么选

Boto3文档里这三个概念常让人晕头转向。我用最直白的比喻解释：Session是你的“AWS身份凭证容器”，Client是“直连AWS API的扳手”，Resource是“带智能包装的螺丝刀”。三者关系是Session → Client/Resource，但选择哪层取决于你要拧多紧的螺丝。

• Client层：最底层，1:1映射AWS API。比如ec2.describe_instances()直接对应EC2的DescribeInstancesAPI。优势是完全可控：你能看到每个请求的HTTP状态码、原始响应体、重试次数；劣势是代码冗长。例如启动EC2实例，Client写法：

  ec2_client = boto3.client('ec2') response = ec2_client.run_instances( ImageId='ami-0c55b159cbfafe1f0', MinCount=1, MaxCount=1, InstanceType='t3.micro', TagSpecifications=[{ 'ResourceType': 'instance', 'Tags': [{'Key': 'Name', 'Value': 'prod-app'}] }] ) instance_id = response['Instances'][0]['InstanceId'] # 需手动提取ID

• Resource层：Client之上的面向对象封装。它把API响应自动转成Python对象，并提供链式方法。同样启动EC2：

  ec2_resource = boto3.resource('ec2') instances = ec2_resource.create_instances( ImageId='ami-0c55b159cbfafe1f0', MinCount=1, MaxCount=1, InstanceType='t3.micro', TagSpecifications=[{ 'ResourceType': 'instance', 'Tags': [{'Key': 'Name', 'Value': 'prod-app'}] }] ) instance_id = instances[0].id # 直接取属性，不用解析字典

• Session层：当你需要跨服务共享凭证或配置时才用。比如同时操作S3和SQS，且要用同一套区域、重试策略：

  session = boto3.Session( region_name='us-east-1', profile_name='prod-admin' # 读取~/.aws/credentials里的profile ) s3 = session.client('s3') sqs = session.resource('sqs') # 混用Client和Resource也OK

我的实操建议：90%的场景用Client。为什么？因为Resource层看似简洁，但隐藏了太多细节。比如S3的Bucket.objects.all().delete()方法，它内部会先发list_objects_v2请求获取所有key，再批量发delete_objects——但如果桶里有10万个对象，这个all()会一次性加载全部key到内存，OOM风险极高。而Client层的list_objects_v2配合Paginator，你能精确控制每次拉多少key、何时停止。Resource层适合快速原型或简单脚本；Client层才是生产环境的标配。至于Session，除非你明确需要多账户切换或自定义凭证链，否则直接boto3.client()更清爽。

3.2 分页机制深度剖析：为什么ContinuationToken比PageNumber更可靠

原文提到S3的ContinuationToken，但没说透它为什么比传统分页更关键。AWS几乎所有列表API（EC2的describe_instances、RDS的describe_db_instances、CloudWatch的describe_alarms）都用Token分页，而非page_number，原因只有一个：服务端状态不可信。想象一下：你调用describe_instances时，AWS后端要从分布式数据库查所有实例，这个过程可能耗时几百毫秒。如果用PageNumber=1，服务端得先算出总条数（比如10万），再跳过前1000条返回第1001-2000条——这要求服务端在查询时锁定全部数据，性能灾难。而ContinuationToken本质是“游标”，它记录的是上次查询的最后一条数据的位置指针（比如ES的search_after）。下次请求时，服务端直接从那个位置往后扫，无需知道总数，也不用跳过前面的数据。这就是为什么IsTruncated=True时，你必须用ContinuationToken，而不是简单地page_number += 1。

但实操中最大的坑是：Token不是永久有效的。AWS文档明确说，Token有效期通常为15分钟。如果你的分页循环里某次请求耗时超过15分钟（比如处理每个对象要调用Lambda函数），Token就失效了，下次请求会报InvalidTokenException。我踩过的最惨一次是清理一个含200万对象的S3桶，脚本跑了40分钟，到第30页时Token过期，整个流程中断。解决方案是：永远用Paginator，且设置合理的PageSize。Paginator内部会自动刷新Token，但你需要告诉它“每次别拉太多”。比如：

paginator = s3_client.get_paginator('list_objects_v2') # 关键：显式设置PageSize，避免默认的1000导致单次请求过大 page_iterator = paginator.paginate( Bucket='huge-bucket', PaginationConfig={'PageSize': 100} # 每页只取100个key ) for page in page_iterator: # 处理page['Contents']，这里可以加sleep或限速 time.sleep(0.1) # 防止QPS超限

PageSize=100意味着每次HTTP请求只拉100个对象，虽然总请求数变多，但单次响应快、内存占用低、Token不易过期。这是用时间换稳定性的经典trade-off。

3.3 凭证管理：为什么硬编码Access Key是自杀行为

新手最容易犯的错，就是在代码里写：

# ❌ 绝对禁止！ s3 = boto3.client( 's3', aws_access_key_id='AKIA...', aws_secret_access_key='...' )

这等于把家门钥匙刻在玻璃门上。AWS凭证泄露是云上最常见、后果最严重的安全事件。正确姿势是依赖Boto3的凭证链自动发现机制。Boto3会按固定顺序查找凭证：

1. 代码中显式传入（仅调试用，生产禁用）
2. 环境变量：AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY
3. AWS配置文件：~/.aws/credentials（Linux/macOS）或%USERPROFILE%\.aws\credentials（Windows）
4. IAM角色：如果代码运行在EC2/ECS/Lambda上，自动获取实例角色的临时凭证

生产环境必须用第3或第4种。比如在EC2上部署脚本，只需给实例附加一个AmazonS3FullAccess策略的角色，代码里就写最简的：

# ✅ 安全且标准 s3 = boto3.client('s3') # 自动使用实例角色凭证

更进一步，我们强制所有团队用命名配置文件。在~/.aws/credentials里：

[prod-admin] aws_access_key_id = AKIA... aws_secret_access_key = ... [dev-s3-reader] aws_access_key_id = AKIA... aws_secret_access_key = ...

然后代码里指定：

s3_prod = boto3.client('s3', profile_name='prod-admin') s3_dev = boto3.client('s3', profile_name='dev-s3-reader')

这样既能隔离环境权限，又避免密钥硬编码。额外提醒：.aws/credentials文件权限必须是600（Linux/macOS），否则Boto3会拒绝读取——这是它的安全保护机制，别怪它“不工作”。

4. 实操全流程：从零写一个生产级S3生命周期清理脚本

4.1 需求拆解与架构设计

我们以一个真实需求为例：某客户每天上传10万+日志文件到S3桶customer-logs-prod，要求“所有创建时间早于30天的文件自动删除，但保留/archive/目录下的所有文件”。这不是简单的DeleteObjects，涉及路径过滤、时间计算、错误重试、执行审计。完整脚本需包含：

• 配置层：桶名、保留天数、排除路径、AWS区域等可配置参数
• 发现层：分页遍历桶内所有对象，应用时间+路径双重过滤
• 执行层：批量删除（最多1000个key/次），失败时记录并重试
• 审计层：输出JSON格式报告，含删除总数、耗时、失败列表
• 防护层：Dry Run模式（只打印不删）、速率限制、内存控制

整个流程不依赖外部库，纯Boto3+标准库，确保最小依赖。

4.2 核心代码实现与逐行注释

#!/usr/bin/env python3 """ S3 Lifecycle Cleaner: 生产级日志清理脚本 功能：删除指定S3桶中过期文件，支持路径排除和Dry Run模式 作者：一线运维工程师 版本：v2.1 (2024-05-20) """ import boto3 import logging import json import time from datetime import datetime, timedelta, timezone from typing import List, Dict, Tuple, Optional import argparse import sys # 配置日志，输出到stdout和文件 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler('/var/log/s3-cleaner.log') ] ) logger = logging.getLogger(__name__) class S3LifecycleCleaner: def __init__( self, bucket_name: str, retention_days: int = 30, exclude_prefixes: List[str] = None, region_name: str = 'us-east-1', dry_run: bool = False, max_delete_batch: int = 1000, page_size: int = 100 ): """ 初始化清理器 :param bucket_name: S3桶名 :param retention_days: 保留天数，早于此时间的对象将被删除 :param exclude_prefixes: 排除的路径前缀列表，如 ['/archive/', '/important/'] :param region_name: AWS区域 :param dry_run: True则只打印将删除的对象，不实际删除 :param max_delete_batch: 单次DeleteObjects最多删除的key数量（AWS上限1000） :param page_size: Paginator每次请求的对象数量，控制内存和Token有效期 """ self.bucket_name = bucket_name self.retention_days = retention_days self.exclude_prefixes = exclude_prefixes or [] self.region_name = region_name self.dry_run = dry_run self.max_delete_batch = max_delete_batch self.page_size = page_size # 创建S3客户端，自动使用环境凭证 self.s3_client = boto3.client('s3', region_name=self.region_name) self.cutoff_time = datetime.now(timezone.utc) - timedelta(days=self.retention_days) logger.info(f"初始化完成：桶={self.bucket_name}, 保留天数={self.retention_days}, " f"截止时间={self.cutoff_time.isoformat()}, DryRun={self.dry_run}") def _should_exclude(self, key: str) -> bool: """判断对象key是否应被排除（不删除）""" if not self.exclude_prefixes: return False return any(key.startswith(prefix) for prefix in self.exclude_prefixes) def _list_expired_objects(self) -> List[str]: """分页遍历桶，返回所有过期且未被排除的对象key列表""" expired_keys = [] paginator = self.s3_client.get_paginator('list_objects_v2') # 配置分页参数 pagination_config = { 'PageSize': self.page_size, 'MaxItems': 1000000 # 防止无限循环，设个合理上限 } try: page_iterator = paginator.paginate( Bucket=self.bucket_name, PaginationConfig=pagination_config ) for i, page in enumerate(page_iterator): logger.debug(f"处理第{i+1}页，共{len(page.get('Contents', []))}个对象") # 遍历当前页所有对象 for obj in page.get('Contents', []): # 跳过排除路径 if self._should_exclude(obj['Key']): continue # 时间判断：对象最后修改时间早于截止时间 if obj['LastModified'] < self.cutoff_time: expired_keys.append(obj['Key']) # 内存保护：如果keys过多，及时yield或分批处理 # 这里我们累积到一定量就处理，避免内存爆炸 if len(expired_keys) >= self.max_delete_batch * 10: logger.warning(f"已发现{len(expired_keys)}个过期对象，开始分批处理...") break except Exception as e: logger.error(f"遍历对象时出错: {e}") raise logger.info(f"共发现{len(expired_keys)}个待删除对象") return expired_keys def _delete_in_batches(self, keys_to_delete: List[str]) -> Dict: """分批删除对象，返回统计信息""" total_deleted = 0 total_failed = 0 failed_details = [] # 将keys分组，每组最多max_delete_batch个 for i in range(0, len(keys_to_delete), self.max_delete_batch): batch = keys_to_delete[i:i + self.max_delete_batch] logger.info(f"处理第{i//self.max_delete_batch + 1}批，共{len(batch)}个对象") if self.dry_run: logger.info(f"[DRY RUN] 将删除以下{len(batch)}个对象: {batch[:5]}{'...' if len(batch)>5 else ''}") total_deleted += len(batch) continue # 构造DeleteObjects请求体 delete_request = { 'Objects': [{'Key': key} for key in batch], 'Quiet': True # True则只返回失败项，减少响应体积 } try: # 执行删除 response = self.s3_client.delete_objects( Bucket=self.bucket_name, Delete=delete_request ) deleted_count = len(batch) - len(response.get('Errors', [])) total_deleted += deleted_count logger.info(f"第{i//self.max_delete_batch + 1}批删除成功{deleted_count}个") # 记录失败详情 for error in response.get('Errors', []): failed_details.append({ 'Key': error['Key'], 'Code': error['Code'], 'Message': error['Message'] }) total_failed += 1 except Exception as e: logger.error(f"第{i//self.max_delete_batch + 1}批删除失败: {e}") total_failed += len(batch) # 将整批标记为失败 for key in batch: failed_details.append({ 'Key': key, 'Code': 'CLIENT_ERROR', 'Message': str(e) }) # 速率控制：避免QPS超限 time.sleep(0.05) return { 'total_processed': len(keys_to_delete), 'total_deleted': total_deleted, 'total_failed': total_failed, 'failed_details': failed_details } def run(self) -> Dict: """执行完整清理流程""" start_time = time.time() logger.info("=== S3生命周期清理任务开始 ===") try: # 步骤1：发现过期对象 expired_keys = self._list_expired_objects() if not expired_keys: logger.info("未发现过期对象，任务结束") result = { 'status': 'success', 'message': 'No objects to delete', 'summary': {'total_processed': 0, 'total_deleted': 0, 'total_failed': 0} } return result # 步骤2：分批删除 summary = self._delete_in_batches(expired_keys) # 步骤3：生成审计报告 duration = time.time() - start_time report = { 'timestamp': datetime.now(timezone.utc).isoformat(), 'bucket': self.bucket_name, 'retention_days': self.retention_days, 'dry_run': self.dry_run, 'execution_time_seconds': round(duration, 2), 'summary': summary } logger.info(f"=== 任务完成，耗时{duration:.2f}秒 ===") logger.info(f"总计处理{summary['total_processed']}个对象，成功删除{summary['total_deleted']}个，" f"失败{summary['total_failed']}个") # 输出JSON报告到stdout，方便管道处理 print(json.dumps(report, indent=2)) return { 'status': 'success', 'report': report } except Exception as e: logger.error(f"任务执行异常: {e}") return { 'status': 'error', 'error': str(e) } # 命令行入口 def main(): parser = argparse.ArgumentParser(description='S3生命周期清理工具') parser.add_argument('--bucket', required=True, help='S3桶名') parser.add_argument('--days', type=int, default=30, help='保留天数，默认30') parser.add_argument('--exclude', nargs='*', default=[], help='排除的路径前缀，如 /archive/ /important/') parser.add_argument('--region', default='us-east-1', help='AWS区域') parser.add_argument('--dry-run', action='store_true', help='仅预览，不执行删除') parser.add_argument('--page-size', type=int, default=100, help='Paginator每页大小，默认100') args = parser.parse_args() cleaner = S3LifecycleCleaner( bucket_name=args.bucket, retention_days=args.days, exclude_prefixes=args.exclude, region_name=args.region, dry_run=args.dry_run, page_size=args.page_size ) result = cleaner.run() sys.exit(0 if result['status'] == 'success' else 1) if __name__ == '__main__': main()

4.3 部署与执行：如何让它真正跑在生产环境

光有脚本不够，必须配套运维机制。我们团队的标准部署流程：

1. 打包为可执行文件：用PyInstaller打包，避免目标服务器Python环境差异。

   pip install pyinstaller pyinstaller --onefile --name s3-cleaner s3_cleaner.py # 生成 ./dist/s3-cleaner

2. 配置Systemd服务（Linux）：

   # /etc/systemd/system/s3-cleaner.service [Unit] Description=S3 Lifecycle Cleaner After=network.target [Service] Type=oneshot User=aws-ops WorkingDirectory=/opt/s3-cleaner ExecStart=/opt/s3-cleaner/dist/s3-cleaner \ --bucket customer-logs-prod \ --days 30 \ --exclude /archive/ /important/ \ --region us-east-1 \ --page-size 50 StandardOutput=journal StandardError=journal # 添加内存限制防OOM MemoryLimit=512M [Install] WantedBy=multi-user.target

   启用：systemctl daemon-reload && systemctl enable s3-cleaner.service

3. 配置Cron定时（作为备选）：

   # 每天凌晨2点执行 0 2 * * * /opt/s3-cleaner/dist/s3-cleaner --bucket customer-logs-prod --days 30 --dry-run >> /var/log/s3-cleaner-dryrun.log 2>&1 # 每周日凌晨3点执行真实清理 0 3 * * 0 /opt/s3-cleaner/dist/s3-cleaner --bucket customer-logs-prod --days 30 >> /var/log/s3-cleaner.log 2>&1

4. 监控与告警：脚本输出JSON报告，用Filebeat采集日志，推送到ELK。设置告警规则：

   • summary.total_failed > 0：立即告警，人工介入
   • execution_time_seconds > 300：耗时超5分钟，可能桶过大或网络问题
   • summary.total_deleted == 0连续3天：检查日志上传是否中断

这套方案已在我们12个客户环境稳定运行14个月，单次最大处理对象数达870万，从未因内存或Token问题中断。

5. 常见问题与排查技巧实录

5.1 典型问题速查表

5.2 我踩过的三个深坑及独家修复技巧

坑一：S3 ListObjectsV2的Prefix陷阱
初学者常以为list_objects_v2(Prefix='/logs/')能列出/logs/2024/05/下的所有文件，但实际它只匹配路径前缀，不会递归。比如桶里有/logs/app1/error.log和/logs/app2/info.log，Prefix='/logs/'能列出两者；但如果有/logs_archive/old.log，它也会被列出（因为/logs_archive/以/logs/开头）。真正的递归过滤必须在代码里做：

# ❌ 错误：以为Prefix能递归 paginator = s3.get_paginator('list_objects_v2') for page in paginator.paginate(Bucket='my-bucket', Prefix='/logs/'): for obj in page['Contents']: # obj['Key']可能是'/logs_archive/old.log'！ # ✅ 正确：用startswith严格匹配 for obj in page['Contents']: if obj['Key'].startswith('/logs/') and not obj['Key'].startswith('/logs_archive/'): process(obj)

坑二：Lambda中Boto3内存泄漏
在Lambda里反复创建boto3.client()会导致冷启动变慢，且可能内存泄漏。我们曾有个Lambda函数，每分钟触发，运行30分钟后内存占用从128MB涨到512MB。根源是Boto3的HTTP连接池未释放。修复方案：全局复用Client。

# ❌ Lambda handler内创建（错误） def lambda_handler(event, context): s3 = boto3.client('s3') # 每次调用都新建 return s3.list_buckets() # ✅ 正确：模块级全局变量 s3_client = boto3.client('s3') # 冷启动时创建一次 def lambda_handler(event, context): return s3_client.list_buckets() # 复用连接池

坑三：跨区域S3操作的隐式失败
S3是全局服务，但list_objects_v2必须指定桶所在区域，否则可能返回空结果而不报错。比如桶在us-west-2，你用boto3.client('s3')（默认us-east-1）调用，AWS会静默返回{}。排查技巧：在list_objects_v2后加断言：

response = s3_client.list_objects_v2(Bucket='my-bucket') if 'Contents' not in response: # 检查是否区域错误 bucket_region = s3_client.get_bucket_location(Bucket='my-bucket')['LocationConstraint'] logger.error(f"桶位于{bucket_region}，但客户端区域为{s3_client.meta.region_name}") raise ValueError("Region mismatch")

5.3 性能调优实战：如何把100万对象清理从2小时降到12分钟

我们有个客户桶含120万对象，初始脚本用默认PageSize=1000，耗时1小时45分钟。