解决AWS Lambda中simple-salesforce处理6.5万条Salesforce记录超时问题

一、simple-salesforce的分页查询支持 #

simple-salesforce完全支持基于nextRecordsUrl的分页查询，同时可通过SOQL的LIMIT参数指定每批次返回的记录数（Salesforce单页最大支持2000条），具体实现方式如下：

1. 初始查询（指定批次大小）
   使用sf.query()执行带LIMIT的SOQL语句，返回结果包含nextRecordsUrl（存在更多数据时）和当前批次记录：

from simple_salesforce import Salesforce

sf = Salesforce(username='your_username', password='your_password', security_token='your_token')
batch_size = 2000  # Salesforce允许的单页最大记录数
soql = f"SELECT Id, ...（你的471个字段） FROM YourCustomObject__c LIMIT {batch_size}"

result = sf.query(soql)
records = result['records']
next_url = result['nextRecordsUrl']

2. 遍历所有分页
   循环调用sf.query_more()，传入上一次返回的nextRecordsUrl，直到done标记为True：

while not result['done']:
    result = sf.query_more(next_url, identifier_is_url=True)
    records.extend(result['records'])
    next_url = result['nextRecordsUrl']
    # 此处插入当前批次的清洗+DataFrame处理逻辑，避免一次性加载全量数据到内存

若不想手动管理分页，sf.query_all_iter()本身就是基于分页的迭代器，可通过SOQL的LIMIT参数控制每次迭代返回的批次大小：

for batch in sf.query_all_iter(f"SELECT ... FROM YourCustomObject__c LIMIT {batch_size}"):
    # 处理当前批次记录
    cleaned_batch = [clean_record(record) for record in batch]
    df_batch = pd.DataFrame(cleaned_batch)
    # 可将批次数据追加到S3的csv.gz文件，或临时存储后合并

二、Lambda超时的批次续接优化 #

你提出的WHERE Id > prevId ORDER BY Id分段思路可行，结合Lambda剩余时间监控可实现断点续传，具体优化点：

1. 记录断点信息
   将每次处理的最后一条记录Id和进度保存到AWS DynamoDB或S3（如JSON文件），作为下一次Lambda触发的起始点。

2. 监控Lambda剩余时间
   在Lambda函数中用context.get_remaining_time_in_millis()监控剩余时间，当剩余时间不足（如小于30秒，留足保存断点的时间）时，停止处理并保存当前断点：

import boto3
import pandas as pd

def lambda_handler(event, context):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('SalesforceProcessingBreakpoint')
    # 读取上次断点Id，默认空字符串
    response = table.get_item(Key={'job_id': 'salesforce_cleanup'})
    last_id = response.get('Item', {}).get('last_id', '')

    batch_size = 1500  # 根据处理速度调整，确保每批处理时间在安全范围
    soql = f"SELECT ... FROM YourCustomObject__c WHERE Id > '{last_id}' ORDER BY Id LIMIT {batch_size}"
    result = sf.query(soql)
    records = result['records']

    while records and context.get_remaining_time_in_millis() > 30000:  # 剩余时间大于30秒
        # 清洗当前批次记录
        cleaned_records = [clean_record(record) for record in records]
        df_batch = pd.DataFrame(cleaned_records)
        # 追加到S3的csv.gz文件（可使用s3fs库直接追加，或临时文件合并）
        append_to_s3_csv(df_batch, 'your-bucket', 'path/to/output.csv.gz')

        # 更新断点Id为当前批次最后一条记录的Id
        last_id = records[-1]['Id']
        table.put_item(Item={'job_id': 'salesforce_cleanup', 'last_id': last_id})

        # 获取下一批次
        if not result['done']:
            result = sf.query_more(result['nextRecordsUrl'], identifier_is_url=True)
            records = result['records']
        else:
            records = []

    # 任务完成后删除断点记录（可选）
    if not records:
        table.delete_item(Key={'job_id': 'salesforce_cleanup'})
    return {'status': 'success', 'last_processed_id': last_id}

3. 其他优化建议

• 提升Lambda内存配置：Lambda内存越高，CPU和网络带宽同步提升，可显著加快数据查询和处理速度，建议从1024MB开始测试调整。
• 批量数据处理：避免逐条清洗记录，改用Pandas向量化操作处理DataFrame，减少循环开销。
• 异步触发后续批次：用CloudWatch Events定时触发Lambda，或在当前Lambda处理完一批后通过SQS发送消息触发下一次处理，实现自动续接。

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