DynamoDB Java SDK v2：高效批量删除数据指南

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>dynamodb-enhanced</artifactId>
    <version>2.x.x</version> <!-- 使用最新稳定版本 -->
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>dynamodb</artifactId>
    <version>2.x.x</version> <!-- 使用最新稳定版本 -->
</dependency>

implementation 'software.amazon.awssdk:dynamodb-enhanced:2.x.x' // 使用最新稳定版本
implementation 'software.amazon.awssdk:dynamodb:2.x.x' // 使用最新稳定版本

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbPartitionKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSortKey;

/**
 * 示例数据模型类，表示DynamoDB表中的一个项目。
 * 至少需要包含主键属性。
 */
@DynamoDbBean
public class ItemEntity {
    private String partitionKey; // 分区键
    private String sortKey;      // 排序键 (如果表有复合主键)
    private String attribute1;   // 其他属性

    // 默认构造函数是必需的
    public ItemEntity() {}

    public ItemEntity(String partitionKey, String sortKey) {
        this.partitionKey = partitionKey;
        this.sortKey = sortKey;
    }

    @DynamoDbPartitionKey
    public String getPartitionKey() {
        return partitionKey;
    }

    public void setPartitionKey(String partitionKey) {
        this.partitionKey = partitionKey;
    }

    @DynamoDbSortKey // 如果有排序键
    public String getSortKey() {
        return sortKey;
    }

    public void setSortKey(String sortKey) {
        this.sortKey = sortKey;
    }

    public String getAttribute1() {
        return attribute1;
    }

    public void setAttribute1(String attribute1) {
        this.attribute1 = attribute1;
    }

    // 重写equals和hashCode方法，对于基于对象的删除非常重要
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        ItemEntity that = (ItemEntity) o;
        return partitionKey.equals(that.partitionKey) &&
               (sortKey != null ? sortKey.equals(that.sortKey) : that.sortKey == null);
    }

    @Override
    public int hashCode() {
        int result = partitionKey.hashCode();
        result = 31 * result + (sortKey != null ? sortKey.hashCode() : 0);
        return result;
    }
}

import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient;
import software.amazon.awssdk.enhanced.dynamodb.model.BatchWriteItemEnhancedRequest;
import software.amazon.awssdk.enhanced.dynamodb.model.WriteBatch;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.model.BatchWriteItemResponse;
import software.amazon.awssdk.services.dynamodb.model.WriteRequest;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class DynamoDBBatchDeleteService {

    private final DynamoDbEnhancedClient enhancedClient;
    private final String tableName;

    /**
     * 构造函数，初始化DynamoDB增强型客户端和表名。
     *
     * @param enhancedClient DynamoDB增强型客户端实例。
     * @param tableName 要操作的DynamoDB表名。
     */
    public DynamoDBBatchDeleteService(DynamoDbEnhancedClient enhancedClient, String tableName) {
        this.enhancedClient = enhancedClient;
        this.tableName = tableName;
    }

    /**
     * 执行批量删除操作。
     * DynamoDB的batchWriteItem请求限制为25个项目。
     * 此方法会自动将传入的项目列表分割成25个项目一批进行处理。
     *
     * @param itemsToDelete 包含要删除项目主键的ItemEntity对象列表。
     *                      每个ItemEntity对象必须至少填充其主键属性。
     * @return 成功删除的项目总数。
     */
    public int batchDeleteItems(List<ItemEntity> itemsToDelete) {
        if (itemsToDelete == null || itemsToDelete.isEmpty()) {
            System.out.println("没有提供要删除的项目。");
            return 0;
        }

        final int BATCH_SIZE = 25; // DynamoDB批处理限制
        int deletedCount = 0;

        // 将项目列表分割成25个一组的批次
        for (int i = 0; i < itemsToDelete.size(); i += BATCH_SIZE) {
            List<ItemEntity> currentBatch = itemsToDelete.subList(i, Math.min(i + BATCH_SIZE, itemsToDelete.size()));

            // 构建BatchWriteItemEnhancedRequest
            BatchWriteItemEnhancedRequest.Builder builder = BatchWriteItemEnhancedRequest.builder();

            currentBatch.forEach(item -> {
                // 为每个项目添加删除请求
                builder.addWriteBatch(WriteBatch.builder(ItemEntity.class)
                        .tableName(tableName) // 指定表名
                        .addDeleteItem(item) // 传入ItemEntity对象，增强客户端会自动提取主键
                        .build());
            });

            try {
                System.out.println(String.format("尝试删除批次中的 %d 个项目 (总进度: %d/%d)...",
                                                 currentBatch.size(), i + currentBatch.size(), itemsToDelete.size()));

                // 执行批量写入操作
                BatchWriteItemResponse response = enhancedClient.batchWriteItem(builder.build());

                // 处理未处理的项目 (UnprocessedItems)
                // DynamoDB可能会因为吞吐量限制或其他原因返回未处理的项目
                if (response.hasUnprocessedItems() && !response.unprocessedItems().isEmpty()) {
                    System.out.println("警告: 批处理中存在未处理的项目。需要重试。");
                    handleUnprocessedItems(response.unprocessedItems());
                } else {
                    System.out.println("批次删除请求发送成功。");
                    deletedCount += currentBatch.size(); // 假设成功发送的都将最终被删除
                }
            } catch (Exception e) {
                System.err.println("批量删除过程中发生错误: " + e.getMessage());
                // 根据业务需求处理异常，例如记录日志、重试等
            }
        }
        return deletedCount;
    }

    /**
     * 处理BatchWriteItem操作返回的未处理项目。
     * 实际应用中可能需要实现指数退避重试逻辑。
     *
     * @param unprocessedItems 未处理的项目映射。
     */
    private void handleUnprocessedItems(Map<String, List<WriteRequest>> unprocessedItems) {
        // 这是一个简化的处理，实际应用中应实现更健壮的重试机制，如指数退避。
        unprocessedItems.forEach((table, requests) -> {
            System.out.println(String.format("表 '%s' 中有 %d 个未处理的请求。", table, requests.size()));
            // 在这里可以提取未处理的请求，并尝试重新提交它们
            // 例如：将它们添加到一个新的批处理请求中，并稍后重试
            // 为了演示，我们只打印信息
            requests.forEach(request -> {
                if (request.deleteRequest() != null) {
                    System.out.println("未处理的删除请求: " + request.deleteRequest().key());
                }
            });
            // 实际应用中：
            // List<ItemEntity> retryItems = convertWriteRequestsToItemEntities(requests);
            // batchDeleteItems(retryItems); // 递归或循环重试，注意避免无限循环
        });

        // 简化的重试等待
        try {
            TimeUnit.SECONDS.sleep(1); // 等待一小段时间再重试
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("重试等待被中断。");
        }
    }

    public static void main(String[] args) {
        // 1. 初始化DynamoDbClient和DynamoDbEnhancedClient
        // 在生产环境中，这些客户端通常通过依赖注入或单例模式进行管理
        DynamoDbClient ddbClient = DynamoDbClient.builder()
                .region(Region.AP_SOUTHEAST_2) // 请替换为您的AWS区域
                // .credentialsProvider(...) // 如果需要，配置AWS凭证
                .build();

        DynamoDbEnhancedClient enhancedClient = DynamoDbEnhancedClient.builder()
                .dynamoDbClient(ddbClient)
                .build();

        String myTableName = "YourActualTableName"; // 替换为您的实际表名
        DynamoDBBatchDeleteService service = new DynamoDBBatchDeleteService(enhancedClient, myTableName);

        // 2. 准备要删除的项目列表
        List<ItemEntity> itemsToDelete = new ArrayList<>();
        // 假设您的表主键是 partitionKey 和 sortKey
        for (int i = 1; i <= 30; i++) { // 示例：创建30个项目，将触发两次批处理请求
            itemsToDelete.add(new ItemEntity("pk" + i, "sk" + i));
        }

        // 3. 执行批量删除
        System.out.println("开始批量删除...");
        int totalDeleted = service.batchDeleteItems(itemsToDelete);
        System.out.println(String.format("批量删除完成。总共处理了 %d 个项目。", totalDeleted));

        // 4. 关闭客户端 (重要)
        ddbClient.close();
        System.out.println("DynamoDB客户端已关闭。");
    }
}