如何通过NodeJS Lambda将DynamoDB Streams数据写入OpenSearch

原代码存在的核心问题 #

• 引入的@aws-sdk/client-opensearch是OpenSearch服务控制平面客户端，仅支持集群创建、配置修改等管控操作，不能用来读写索引数据
• 索引配置错误：index是OpenSearch中存储数据的逻辑容器（类似关系型数据库的表），不是字段名，需要针对name字段检索只需要在写入时给name字段配置文本类型索引即可
• 缺少请求签名逻辑：AWS托管的OpenSearch服务要求所有HTTP请求必须经过SigV4签名才能通过鉴权，直接发起未签名请求会返回403权限错误
• 数据格式不匹配：DynamoDB Streams返回的NewImage是DynamoDB专属的AttributeValue格式（例如{S: "xxx"}标记字符串类型），需要反序列化为普通JS对象才能正常写入
• 事件覆盖不全：仅考虑了新数据写入，没有覆盖修改、删除事件的同步逻辑，无法保证数据最终一致

前置准备 #

1. 给Lambda执行角色附加对应权限：
   • 缩权配置可授予目标OpenSearch域名的es:ESHttp*权限，测试阶段可临时使用托管策略AmazonOpenSearchServiceFullAccess
   • 保留原有的DynamoDB Streams读取权限
2. 如果OpenSearch集群部署在VPC内，将Lambda接入同一VPC，配置安全组允许Lambda所属安全组访问OpenSearch域名的443端口
3. 安装Lambda运行依赖：

npm install @opensearch-project/opensearch @aws-sdk/credential-provider-node @aws-sdk/util-dynamodb

完整可运行代码 #

const { Client } = require('@opensearch-project/opensearch');
const { defaultProvider } = require('@aws-sdk/credential-provider-node');
const { unmarshall } = require('@aws-sdk/util-dynamodb');
const { AwsSigv4Signer } = require('@opensearch-project/opensearch/aws');

// 基础配置 - 按实际环境修改
const OPENSEARCH_DOMAIN = 'search-ddbsearch-yxxxxxxgu.us-east-2.es.amazonaws.com';
const OPENSEARCH_INDEX = 'cars'; // 索引名，对应存储汽车类数据，不要用字段名作为索引名
const REGION = process.env.AWS_REGION; // 复用Lambda自带的区域环境变量，无需硬编码

// 初始化带SigV4自动签名的OpenSearch客户端
const client = new Client({
  ...AwsSigv4Signer({
    region: REGION,
    service: 'es', // 托管OpenSearch服务标识固定为es，Serverless版本需改为aoss
    getCredentials: () => defaultProvider()(),
  }),
  node: `https://${OPENSEARCH_DOMAIN}`,
});

// 首次运行自动创建索引，提前配置字段映射
async function initIndex() {
  const indexExists = await client.indices.exists({ index: OPENSEARCH_INDEX });
  if (!indexExists.body) {
    await client.indices.create({
      index: OPENSEARCH_INDEX,
      body: {
        mappings: {
          properties: {
            id: { type: 'keyword' }, // 主键用keyword类型，支持精确匹配
            name: { type: 'text', analyzer: 'standard' }, // name字段设为text类型，支持全文检索
            color: { type: 'keyword' },
            price: { type: 'keyword' } // 示例中price带逗号为字符串格式，若后续改为纯数值可替换为float类型
          }
        }
      }
    });
    console.log(`Index ${OPENSEARCH_INDEX} initialized`);
  }
}

// 冷启动阶段只初始化一次索引
let indexReady = false;

module.exports.handler = async (event) => {
  if (!indexReady) {
    await initIndex();
    indexReady = true;
  }

  console.log("DDB Stream event === ", JSON.stringify(event, null, 2));
  const bulkOperations = [];

  for (const record of event.Records) {
    const eventType = record.eventName;
    const primaryKey = unmarshall(record.dynamodb.Keys);
    const docId = primaryKey.id; // 用DynamoDB主键作为OpenSearch文档ID，保证写入幂等

    if (eventType === 'REMOVE') {
      // 删除事件：同步移除OpenSearch中对应文档
      bulkOperations.push({ delete: { _index: OPENSEARCH_INDEX, _id: docId } });
    } else {
      // INSERT/Modify事件：全量覆盖写入文档
      const formattedData = unmarshall(record.dynamodb.NewImage);
      bulkOperations.push({ index: { _index: OPENSEARCH_INDEX, _id: docId } });
      bulkOperations.push(formattedData);
    }
  }

  // 批量写入提升性能，适配DynamoDB Streams批量推送特性
  if (bulkOperations.length > 0) {
    const { body: bulkRes } = await client.bulk({ 
      body: bulkOperations, 
      refresh: event.Records.length < 10 // 小批量实时刷新，大批量可关闭刷新提升写入速度
    });
    if (bulkRes.errors) {
      const errorItems = bulkRes.items.filter(item => item.index?.error || item.delete?.error);
      console.error('Bulk sync failed items:', JSON.stringify(errorItems, null, 2));
      throw new Error(`Failed to sync ${errorItems.length} records`);
    }
  }

  return `Successfully processed ${event.Records.length} records.`;
}

关键说明 #

• 你之前疑问的type参数：OpenSearch 7.0及以上版本已废弃自定义type概念，所有文档默认归属_doc类型，写入时无需单独指定type参数即可。
• 用DynamoDB主键id作为OpenSearch文档ID，重复写入同ID数据会自动覆盖，天然支持修改事件同步，即使Lambda重试也不会产生重复数据，适配DynamoDB Streams至少一次交付的特性。
• 采用批量写入（bulk）接口，相比单条写入QPS提升5~10倍，更适配生产环境高并发数据同步场景。
• 提前配置name字段为text类型，默认使用standard分词器，开箱支持模糊搜索、关键词匹配、分词检索等需求。
• 凭证自动从Lambda运行环境获取，无需硬编码AK/SK，符合安全最佳实践。
• 若OpenSearch开启了细粒度访问控制，需要在OpenSearch访问配置中给Lambda执行角色的ARN授予对应索引的读写权限。

内容的提问来源于stack exchange，提问作者Ninja-aman