如何通过Boto3高效获取700万条Security Hub的Inspector检查结果

咱先揪个你现有代码里的隐形bug——你看return inspector_findings的缩进错了！它卡在try块的内部、while循环的第一次迭代里，导致代码跑一次循环就直接返回了，根本没遍历所有分页！先把这个缩进移到while循环和try块的外面，不然你连全量数据都拿不到，更别说快慢的问题了。

回到核心问题：700万条数据确实是个量级，单靠单线程一页一页蹭肯定慢到离谱。给你两个方案，按优先级排序：

方案一：用Security Hub官方导出功能（强烈推荐，效率拉满） #

对于百万级别的结果，AWS官方早就给了最优解——用StartExportFindings API把结果异步导出到S3桶里。这个方法完全不需要你自己处理分页、并发、限流，全由AWS后台批量处理，速度比你自己调API快N倍，还不会占你本地的内存和CPU。

具体代码示例 #

import boto3
import time

def export_inspector_findings_to_s3(s3_bucket, s3_prefix):
    securityhub = boto3.client('securityhub')
    
    # 启动导出任务
    try:
        export_resp = securityhub.start_export_findings(
            Filters={
                'ProductName': [{'Value': 'Inspector', 'Comparison': 'EQUALS'}]
            },
            ExportDestination={
                'S3Bucket': s3_bucket,
                'S3Prefix': s3_prefix,
                'Format': 'JSON_LINE'  # 也可以选CSV，看你后续需求
            }
        )
    except Exception as e:
        print(f"启动导出任务失败: {e}")
        return

    job_id = export_resp['JobId']
    print(f"导出任务已启动，Job ID: {job_id}")

    # 轮询任务状态，直到完成
    while True:
        job_status_resp = securityhub.describe_export_findings_job(JobId=job_id)
        status = job_status_resp['JobStatus']
        
        if status == 'COMPLETED':
            dest = job_status_resp['ExportDestination']
            print(f"导出完成！文件路径: s3://{dest['S3Bucket']}/{dest['S3Prefix']}")
            # 可选：从S3下载到本地
            s3 = boto3.client('s3')
            for obj in s3.list_objects_v2(Bucket=s3_bucket, Prefix=s3_prefix)['Contents']:
                local_file = f"./{obj['Key'].split('/')[-1]}"
                s3.download_file(s3_bucket, obj['Key'], local_file)
                print(f"已下载文件到本地: {local_file}")
            break
        elif status in ['FAILED', 'CANCELLED']:
            print(f"导出失败！状态: {status}，原因: {job_status_resp.get('FailureReason', '未知')}")
            break
        else:
            print(f"导出中，当前状态: {status}，10秒后重试...")
            time.sleep(10)

# 替换成你的S3桶名和前缀
export_inspector_findings_to_s3("your-security-hub-export-bucket", "inspector-findings-2024/")

这个方案的优势 #

1. 速度快：AWS内部批量导出，比你自己调API快几个数量级
2. 不占本地资源：不需要你维护分页逻辑，也不用把700万条数据塞进内存（你原来的代码把所有数据存列表，分分钟内存溢出）
3. 后续处理方便：导出的文件可以直接用Athena做分析，或者导入数据库，不用自己写解析逻辑

方案二：优化API分页代码（适合不能用S3的场景） #

如果一定要通过API直接获取，那得从这几个点优化：

1. 先把单线程效率拉满 #

• 把MaxResults设为1000（这是get_findings允许的最大值），直接把请求次数从7万次降到7000次，立竿见影
• 不要把所有数据存到内存列表里！改成流式写入本地文件，避免内存爆炸

修改后的单线程代码：

import boto3
import json

securityhub_client = boto3.client('securityhub')

def stream_inspector_findings(output_file):
    next_token = ""
    page_num = 0
    # 用JSON Lines格式逐行写入，避免内存溢出
    with open(output_file, 'w', encoding='utf-8') as f:
        while True:
            try:
                request_kwargs = {
                    'MaxResults': 1000,
                    'Filters': {'ProductName': [{'Value': 'Inspector', 'Comparison': 'EQUALS'}]}
                }
                if next_token:
                    request_kwargs['NextToken'] = next_token
                
                resp = securityhub_client.get_findings(**request_kwargs)
                # 逐行写入每条结果
                for finding in resp['Findings']:
                    json.dump(finding, f)
                    f.write('\n')
                
                page_num += 1
                print(f"已处理第 {page_num} 页，新增 {len(resp['Findings'])} 条结果")
                
                next_token = resp.get('NextToken')
                if not next_token:
                    print("所有结果处理完成！")
                    break
            except Exception as e:
                print(f"处理第 {page_num+1} 页时出错: {e}")
                break

# 执行流式写入
stream_inspector_findings('inspector_findings.jsonl')

2. 用多线程/异步进一步提速 #

单线程7000次请求还是要等很久，你可以用concurrent.futures.ThreadPoolExecutor做并行请求（注意控制并发数，比如10-20个线程，避免触发AWS的API限流）。不过因为NextToken是依赖前一个请求的，没法直接并行所有分页，但可以用“预取”的方式，或者用aioboto3做异步IO（比线程更高效）。不过说实话，700万条数据的话，还是方案一更靠谱。

最后提醒 #

不管用哪种方案，都别把700万条数据全塞到内存列表里！轻则内存告警，重则直接OOM崩溃，一定要用流式处理或者导出到外部存储。

备注：内容来源于stack exchange，提问作者ThatDevOpsGuy