如何使用Java通过DynamoDB触发器将表数据流式索引同步至Elasticsearch

前置准备 #

• 开启目标DynamoDB表的DynamoDB Streams功能，流的视图类型选择NEW_AND_OLD_IMAGES（后续如需兼容更新、删除场景的同步不需要额外改造）
• 准备JDK 17+、Maven/Gradle的Java开发环境
• 部署可正常访问的Elasticsearch集群，提前规划好待同步的索引结构，或通过代码自动创建映射
• 配置权限：给同步逻辑的运行主体开放DynamoDB Streams的读取权限、Elasticsearch的文档写入权限

核心实现选型 #

优先推荐AWS Lambda无服务器方案，适合新手快速落地：

• 原生支持将DynamoDB Streams作为事件源，无需自行编写流轮询、位点维护的逻辑
• 按调用量计费，无需长期维护常驻服务器
• 事件触发延迟低，表数据插入后几秒即可完成同步
  如果有自托管Java服务的需求，也可以在自有服务中直接消费DynamoDB Streams，核心逻辑和Lambda实现完全一致，仅需要额外维护流消费的检查位点。

具体Java实现 #

1. Maven依赖配置 #

<dependencies>
    <!-- AWS Lambda Java 核心运行依赖 -->
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-lambda-java-core</artifactId>
        <version>1.2.2</version>
    </dependency>
    <!-- DynamoDB Streams 事件对象解析依赖 -->
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-lambda-java-events</artifactId>
        <version>3.11.0</version>
    </dependency>
    <!-- Elasticsearch 官方Java客户端（适配7.x/8.x集群） -->
    <dependency>
        <groupId>co.elastic.clients</groupId>
        <artifactId>elasticsearch-java</artifactId>
        <version>8.11.0</version>
    </dependency>
</dependencies>

2. 核心同步逻辑代码 #

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.DynamodbEvent;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest5.lowlevel.Rest5Client;
import org.apache.http.HttpHost;
import java.util.HashMap;
import java.util.Map;

public class DynamoDbToEsSyncHandler implements RequestHandler<DynamodbEvent, String> {
    // 从环境变量读取ES配置，避免硬编码敏感信息
    private final ElasticsearchClient esClient;
    private static final String ES_INDEX_NAME = "dynamodb_synced_index";

    public DynamoDbToEsSyncHandler() {
        String esEndpoint = System.getenv("ES_CLUSTER_ENDPOINT");
        Rest5Client restClient = Rest5Client.builder()
                .endpoint(new HttpHost(
                        esEndpoint.split(":")[0],
                        Integer.parseInt(esEndpoint.split(":")[1]),
                        "http"
                ))
                .build();
        this.esClient = new ElasticsearchClient(restClient, new JacksonJsonpMapper());
    }

    @Override
    public String handleRequest(DynamodbEvent input, Context context) {
        int processedCount = 0;
        for (DynamodbEvent.DynamodbStreamRecord record : input.getRecords()) {
            // 仅处理新插入的记录，可根据需要扩展MODIFY、REMOVE事件的处理逻辑
            if (!"INSERT".equals(record.getEventName())) {
                continue;
            }
            // 解析DynamoDB流中的记录，转换为ES可识别的文档结构
            Map<String, Object> esDocument = new HashMap<>();
            record.getDynamodb().getNewImage().forEach((attrName, attrValue) -> {
                // 覆盖常用DynamoDB属性类型转换，按需扩展其他类型
                if (attrValue.getS() != null) esDocument.put(attrName, attrValue.getS());
                if (attrValue.getN() != null) esDocument.put(attrName, Double.parseDouble(attrValue.getN()));
                if (attrValue.getBOOL() != null) esDocument.put(attrName, attrValue.getBOOL());
                if (attrValue.getM() != null) {
                    // 处理嵌套Map类型的属性转换
                    Map<String, Object> nestedAttr = new HashMap<>();
                    attrValue.getM().forEach((k, v) -> {
                        if (v.getS() != null) nestedAttr.put(k, v.getS());
                    });
                    esDocument.put(attrName, nestedAttr);
                }
            });
            // 用DynamoDB记录的主键作为ES文档ID，保证重复消费时的幂等性
            String docId = record.getDynamodb().getKeys().get("id").getS();
            try {
                IndexRequest<Map<String, Object>> indexRequest = IndexRequest.of(i -> i
                        .index(ES_INDEX_NAME)
                        .id(docId)
                        .document(esDocument)
                );
                esClient.index(indexRequest);
                processedCount++;
            } catch (Exception e) {
                context.getLogger().log("Document sync failed, ID: " + docId + ", Error: " + e.getMessage());
                // 可将失败记录投递到死信队列，避免数据丢失
            }
        }
        return "Sync completed, processed " + processedCount + " new records";
    }
}

落地注意事项 #

• 绝对不要把ES集群的认证信息、访问地址硬编码在代码中，通过Lambda环境变量、Secrets Manager等方式存储敏感配置
• DynamoDB Streams存在重复投递的特性，必须用DynamoDB记录的主键作为ES文档ID，保证重复处理时不会产生重复数据
• 若单次触发的流记录数量较多，需将单条写入ES的逻辑替换为批量写入，避免触发ES的写入限流
• 自托管ES集群时，要保证Lambda所在VPC与ES集群网络连通，安全组配置需放通Lambda的出站访问权限
• 不需要自行维护流消费位点，Lambda服务会自动记录流的处理进度，处理失败的记录会自动按策略重试

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