如何将DynamoDB流的类型化JSON转换后推送至OpenSearch

不需要手动编写全量递归ETL逻辑处理每条记录的类型标记，Boto3 SDK本身内置了专门的类型转换工具，改造成本极低。

最优实现：使用Boto3内置反序列化工具 #

Boto3的boto3.dynamodb.types模块自带TypeDeserializer类，专门用于将DynamoDB返回的带类型标记的结构，转换为原生Python/JSON格式，自动识别所有DynamoDB原生类型：

• S/N/BOOL/NULL等基础类型自动转为对应的字符串、数字、布尔值、空值
• L类型自动转为普通列表，你提到的{"L": [{"S": "AWS"}]}会直接转换为["AWS"]的纯数组格式
• M类型自动转为普通字典，不会保留类型嵌套层级
• SS/NS/BS等集合类型自动转为Python集合，可按需转列表后推送

核心代码示例 #

在你的Lambda函数中添加如下转换逻辑即可：

import json
from decimal import Decimal
from boto3.dynamodb.types import TypeDeserializer

# 初始化反序列化器，全局复用避免重复初始化
deserializer = TypeDeserializer()

def ddb_item_to_plain_dict(ddb_item):
    """将DynamoDB带类型标记的Item转为普通JSON可序列化字典"""
    plain_item = {}
    for key, typed_value in ddb_item.items():
        plain_item[key] = deserializer.deserialize(typed_value)
    return plain_item

def json_default_handler(obj):
    """处理DynamoDB数字类型反序列化后的Decimal类型，避免JSON序列化报错"""
    if isinstance(obj, Decimal):
        # 整数场景可转int，浮点场景转float，按需调整
        return int(obj) if obj % 1 == 0 else float(obj)
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

# 处理DynamoDB流事件示例
def lambda_handler(event, context):
    documents_to_push = []
    for record in event['Records']:
        # 新增/修改事件取NewImage，删除事件取OldImage
        if record['eventName'] in ('INSERT', 'MODIFY'):
            typed_item = record['dynamodb']['NewImage']
        elif record['eventName'] == 'REMOVE':
            typed_item = record['dynamodb']['OldImage']
        else:
            continue
        # 一步完成格式转换
        plain_item = ddb_item_to_plain_dict(typed_item)
        # 序列化为JSON即可直接推送到OpenSearch
        opensearch_doc = json.dumps(plain_item, default=json_default_handler)
        documents_to_push.append(opensearch_doc)
    
    # 后续执行OpenSearch批量写入逻辑即可

额外ETL的适用场景 #

只有当你存在如下自定义需求时，才需要在反序列化完成后补充少量处理逻辑，不需要从零编写类型解析代码：

• 需要裁剪字段，只同步指定列到OpenSearch
• 需要做字段值转换，比如时间戳格式化、枚举值映射
• 需要对嵌套字段做打平、聚合等自定义加工

效率说明 #

内置的TypeDeserializer是Boto3官方维护的优化实现，性能远高于自行编写的递归类型判断逻辑，不会给Lambda同步链路带来额外延迟，也不会漏处理DynamoDB的冷门类型（比如二进制类型、自定义文档类型），是当前场景下成本最低、可靠性最高的方案。

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