如何将DynamoDB批量写入的未处理WriteRequest转换为Item？

明确结论：可以直接从WriteRequest中提取数据并转换为Item。

核心转换逻辑 #

DynamoDB的WriteRequest若是PUT操作，内部会包含PutRequest对象，其中的item字段存储着Map<String, AttributeValue>结构——这正是构建Item的核心数据。AWS SDK提供了Item.fromAttributeMap()方法，可直接将该结构转换为Item实例。

结合你的代码实现 #

针对你现有的后续处理逻辑，可按如下方式修改以提取失败的Item：

ParallelPromises.yieldAll(promises).map(results -> {
    // 遍历所有结果，提取未处理的WriteRequest并转换为Item
    List<Item> failedItems = results.stream()
            // 取出每个ExecResult的实际返回值（未处理项集合）
            .flatMap(result -> result.getValue().stream())
            // 遍历每个表的未处理WriteRequest列表
            .flatMap(writeRequests -> writeRequests.stream())
            // 过滤出PUT类型的WriteRequest（你的场景是批量插入，所以都是PUT）
            .filter(writeRequest -> writeRequest.getPutRequest() != null)
            // 从PutRequest中取出AttributeValue映射，转换为Item
            .map(writeRequest -> Item.fromAttributeMap(writeRequest.getPutRequest().getItem()))
            .collect(Collectors.toList());

    // 这里将failedItems记录到Kibana，比如转成JSON字符串后输出至日志系统
    failedItems.forEach(item -> {
        // 示例：输出Item的JSON格式，方便Kibana采集解析
        System.out.println(item.toJSON());
    });

    return failedItems;
});

额外说明 #

• 若你的批量操作包含DELETE请求，需额外判断writeRequest.getDeleteRequest()，但你的场景是插入，可忽略此逻辑；
• 转换后的Item可调用toJSON()、toMap()等方法，快速序列化为Kibana可识别的格式。

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