基于DocumentDB+Lambda+OpenSearch架构的数据流丢失问题排查

• 架构：基于AWS云服务搭建全文检索系统，包含：开启changeStreams的双实例DocumentDB集群、运行Python脚本的Lambda、接收变更数据的OpenSearch
• 测试异常：通过mongoshell逐个插入10条文档，多次测试均出现第5条未同步至OpenSearch的情况
• 尝试操作：使用stream.try_next()始终返回None
• Lambda配置：内存1024MB，超时15分钟，预置并发数10；触发器批量大小1000，全文档配置为updateLookup
• 用户提供的Lambda代码：

client = pymongo.MongoClient(host, port, username, password, ssl)

db = client['my_db']
coll = db.get_collection('my_collection', read_preference=pymongo.ReadPreference.PRIMARY)
stream = coll.watch()

try:
    change = stream.next() 
    if change is not None:
        document = change['fullDocument']
        myObject = document['myObject']

        headers = { "Content-Type": "application/json" }
        query = {
            'id': document['docId'],
            'name': myObject['docName']
        }
        url = 'url-to-openSearch-domain/my-index/_doc/' + document['docId']
 
        basic = HTTPBasicAuth(username, password)
        r = requests.put(url, auth=basic, headers, data=json.dumps(query))

根本原因 #

1. 触发器与代码逻辑完全不匹配：你配置了DocumentDB changeStream触发器（批量推送事件），但代码却手动创建coll.watch()流读取事件，触发器推送的批量事件被直接忽略，代码仅处理单个事件就结束Lambda执行，剩余事件全部丢失。
2. stream.try_next()返回None的原因：触发器触发Lambda时，已将变更事件推送到Lambda的event参数中，手动创建的watch()流是从当前时间点或起始位置重新监听，无法获取触发器已推送过的事件，因此返回None。
3. 单事件处理逻辑缺陷：stream.next()仅处理一个事件就结束执行，而触发器每次会打包多个变更事件（哪怕是逐个插入，DocumentDB也会批量聚合），未被处理的事件直接被丢弃，这就是部分数据丢失的核心原因。

修正方案 #

1. 改用触发器传递的event参数处理事件 #

Lambda的DocumentDB changeStream触发器会把批量变更事件存入event['Records']，直接遍历该列表处理即可，无需手动创建watch()流：

import json
from requests import put, HTTPBasicAuth

def lambda_handler(event, context):
    # 遍历触发器推送的批量变更事件
    for record in event['Records']:
        # 获取全文档（对应触发器的updateLookup配置）
        document = record['d']['fullDocument']
        if not document:
            continue
            
        myObject = document['myObject']
        headers = {"Content-Type": "application/json"}
        query = {
            'id': document['docId'],
            'name': myObject['docName']
        }
        url = f'url-to-openSearch-domain/my-index/_doc/{document["docId"]}'
        
        basic = HTTPBasicAuth('your-os-username', 'your-os-password')
        # 增加异常处理，避免单个事件失败导致批量处理中断
        try:
            response = put(url, auth=basic, headers=headers, data=json.dumps(query))
            response.raise_for_status()
        except Exception as e:
            # 可记录日志或将失败事件放入SQS队列重试
            print(f"索引文档失败 {document['docId']}: {str(e)}")

2. 优化触发器配置（可选） #

• 若插入频率较低，可将批量大小调小（如10），降低单次Lambda处理的事件数量
• 确认触发器起始位置配置符合业务需求（可选Trim Horizon或Latest）

3. 增加错误重试机制 #

为避免网络波动或OpenSearch临时不可用导致的数据丢失，建议将处理失败的事件发送至SQS队列，配置死信队列，再通过另一个Lambda消费SQS进行重试。

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