优化多记录类型的DynamoDB操作：插入与更新

针对你遇到的5万条JSON记录批量合并写入DynamoDB的性能问题，结合Lambda的限制，提供以下几个务实的优化方案：

1. 内存预合并+PartiQL批量UPSERT（推荐） #

核心思路 #

• 先在内存中对每个1000条的批次做预合并，把同一process_id的类型A/B记录合并为单条，减少后续DB操作次数。
• 使用DynamoDB PartiQL的ON DUPLICATE KEY UPDATE语法实现UPSERT逻辑，无需提前调用get_item判断记录是否存在，直接通过批量语句提交，大幅降低请求量。
• 利用BatchExecuteStatement接口，每次批量执行25条PartiQL语句（DynamoDB的批量上限），比单条put_item/update_item效率提升数倍。

代码示例 #

import boto3
from botocore.config import Config

# 配置自动重试，处理吞吐量不足的情况
dynamodb_config = Config(
    retries={
        'max_attempts': 5,
        'mode': 'standard'
    }
)
dynamodb = boto3.client('dynamodb', config=dynamodb_config)
TABLE_NAME = "your_target_table"

def merge_batch_records(batch):
    """内存预合并同一process_id的记录"""
    merged_dict = {}
    for rec in batch:
        pid = rec['process_id']
        if pid not in merged_dict:
            merged_dict[pid] = {
                'process_id': pid,
                'client_id': rec['client_id'],
                'start_date': rec.get('start_date'),
                'end_date': rec.get('end_date')
            }
        else:
            # 补充缺失的日期字段
            if 'start_date' in rec:
                merged_dict[pid]['start_date'] = rec['start_date']
            if 'end_date' in rec:
                merged_dict[pid]['end_date'] = rec['end_date']
    return list(merged_dict.values())

def execute_partiql_batch(merged_records):
    """批量执行PartiQL UPSERT语句"""
    statements = []
    for rec in merged_records:
        # 构造PartiQL语句，只更新不存在的日期字段
        params = [
            {'S': rec['process_id']},
            {'S': rec['client_id']}
        ]
        if rec['start_date'] and rec['end_date']:
            stmt = f"""
                INSERT INTO {TABLE_NAME} (process_id, client_id, start_date, end_date)
                VALUES (:pid, :cid, :sd, :ed)
                ON DUPLICATE KEY UPDATE 
                    start_date = IF_NOT_EXISTS(start_date, :sd),
                    end_date = IF_NOT_EXISTS(end_date, :ed)
            """
            params.extend([{'S': rec['start_date']}, {'S': rec['end_date']}])
        elif rec['start_date']:
            stmt = f"""
                INSERT INTO {TABLE_NAME} (process_id, client_id, start_date)
                VALUES (:pid, :cid, :sd)
                ON DUPLICATE KEY UPDATE start_date = IF_NOT_EXISTS(start_date, :sd)
            """
            params.append({'S': rec['start_date']})
        elif rec['end_date']:
            stmt = f"""
                INSERT INTO {TABLE_NAME} (process_id, client_id, end_date)
                VALUES (:pid, :cid, :ed)
                ON DUPLICATE KEY UPDATE end_date = IF_NOT_EXISTS(end_date, :ed)
            """
            params.append({'S': rec['end_date']})
        
        statements.append({
            'Statement': stmt.strip(),
            'Parameters': params
        })
        
        # 每25条执行一次批量语句
        if len(statements) == 25:
            resp = dynamodb.batch_execute_statement(Statements=statements)
            # 处理单条记录的错误（如主键冲突之外的异常）
            for idx, result in enumerate(resp['Responses']):
                if 'Error' in result:
                    print(f"Record {statements[idx]['Parameters'][0]['S']} failed: {result['Error']['Message']}")
            statements = []
    # 处理剩余不足25条的记录
    if statements:
        resp = dynamodb.batch_execute_statement(Statements=statements)
        for idx, result in enumerate(resp['Responses']):
            if 'Error' in result:
                print(f"Record {statements[idx]['Parameters'][0]['S']} failed: {result['Error']['Message']}")

def lambda_handler(event, context):
    # 从S3读取并解析JSON记录（替换为你的实际逻辑）
    raw_records = load_records_from_s3(event)
    # 按1000条拆分大批次，避免Lambda超时
    BATCH_SIZE = 1000
    for i in range(0, len(raw_records), BATCH_SIZE):
        current_batch = raw_records[i:i+BATCH_SIZE]
        # 内存预合并
        merged_records = merge_batch_records(current_batch)
        # 批量写入DB
        execute_partiql_batch(merged_records)

2. 事务批量处理（备选） #

如果需要严格保证操作原子性，可以将25条以内的put_item/update_item操作打包进DynamoDB事务（transact_write_items），每个事务最多支持25条操作。

核心逻辑 #

• 对预合并后的记录，先尝试用Put操作带条件表达式（attribute_not_exists(process_id)）插入新记录。
• 若插入失败（记录已存在），则改用Update操作补充对应日期字段。
• 将这些操作按25条一组打包进事务提交，减少网络往返次数。

3. Lambda配置优化 #

• 提升内存配额：Lambda的CPU、网络带宽与内存正相关，建议设置为2048MB或更高，加快JSON解析和DB操作速度。
• 设置合理超时：将Lambda超时设为15分钟（上限），确保1000条批次的处理时间足够。

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