Lambda执行报错：ClientError对象无法JSON序列化的解决求助

一、核心错误原因：异常返回格式不正确 #

你的lambda_handler捕获异常后直接返回了Exception对象：

except Exception as e:
    return e

Lambda运行时要求返回值必须是可JSON序列化的对象，但ClientError这类异常对象无法直接被JSON序列化，这就是触发Runtime.MarshalError的直接原因。

修复方案： #

修改异常处理逻辑，返回标准的可序列化字典：

def lambda_handler(event,context):
    try:
        print("Event Received", event)
        dynamo_bulk_writer(event)
        # 执行成功时返回正常响应
        return {"statusCode": 200, "message": "数据同步完成"}
    except Exception as e:
        # 将异常信息转为字符串后封装成字典返回
        return {
            "statusCode": 500,
            "errorMessage": str(e),
            "errorType": type(e).__name__
        }

二、处理DynamoDB的Datetime字段（预防后续序列化问题） #

源表存在Datetime类型属性，boto3从DynamoDB读取这类数据时会返回datetime对象，若后续逻辑需要使用或返回这些字段，同样会触发JSON序列化失败，需提前转换为字符串格式：

转换方法： #

新增一个递归转换函数，在读取数据时处理datetime对象：

from datetime import datetime

def convert_datetime(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    elif isinstance(obj, dict):
        return {k: convert_datetime(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_datetime(item) for item in obj]
    return obj

# 在dynamo_bulk_reader的循环中调用转换函数
data.append(convert_datetime(response['Items'][i]))

三、其他代码优化建议 #

1. 批量写入效率优化：当前dynamo_bulk_writer每次循环都创建batch_writer，会频繁建立/关闭连接，效率极低，应将with table.batch_writer()放在循环外部：

def dynamo_bulk_writer(event):
    dynamodb = boto3.resource('dynamodb', region_name=event['TargetRegion'])
    table = dynamodb.Table(event['EnvironmentId']+'-xyz-dynamo2')

    print("Importing items into: " + str(table))
    items = dynamo_bulk_reader(event)
    
    # 批量写入上下文管理器放在循环外
    with table.batch_writer() as batch:
        for table_item in items:
            batch.put_item(
                Item = {
                    'SubsId'  : table_item['SubsId'],
                    'ItemId'  : table_item['ItemId'],
                    'RecType' : table_item['RecType'],
                    'BuId'    : int(table_item['SubsId'].split("_")[0])
                }
            )
    print("Finished importing items...")

2. 处理Query分页数据：当前dynamo_bulk_reader只读取了Query接口返回的第一页数据，若数据量超过1MB或1000条，会丢失后续数据，需处理LastEvaluatedKey实现完整分页读取：

def dynamo_bulk_reader(event):
    dynamodb = boto3.resource('dynamodb', region_name=event['TargetRegion'])
    table = dynamodb.Table(event['EnvironmentId']+'-abc-dynamo')

    print("Exporting items from: " + str(table))
    data = []
    last_evaluated_key = None

    while True:
        query_params = {
            'IndexName': 'idx-RecType-SubsId',
            'KeyConditionExpression': Key('RecType').eq('CreditNote') & Key('SubsId').begins_with(str(event['BuisnessUnitId'])+'_'),
            'FilterExpression': Attr('Split').eq(0)
        }
        if last_evaluated_key:
            query_params['ExclusiveStartKey'] = last_evaluated_key
        
        response = table.query(**query_params)
        
        for item in response['Items']:
            if abs(item['CNAttr']['RemBal']) > 0:
                data.append(convert_datetime(item))  # 同时处理datetime字段
        
        last_evaluated_key = response.get('LastEvaluatedKey')
        if not last_evaluated_key:
            break

    print("Finished exporting: " + str(len(data)) + " items.")
    print("Data Exported: ")
    pprint.pprint(data)
    return data

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