如何在Java中使用batchWriteItem实现DynamoDB数据批量删除？

核心说明 #

batchWriteItem 是DynamoDB批量操作的核心API，一次请求最多支持25个删除操作，且总请求大小不能超过16MB。如果GSI查询返回大量数据，需要将主键列表分批处理，同时必须处理API返回的UnprocessedItems（部分操作失败时会返回，需重试）。

代码示例（AWS SDK for Java v2） #

1. 依赖配置（Maven）

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>dynamodb</artifactId>
    <version>2.25.0</version> <!-- 替换为最新稳定版 -->
</dependency>

2. 完整批量删除代码

import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.services.dynamodb.model.*;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class DynamoDBBatchDelete {
    private static final String TABLE_NAME = "your-table-name";
    private static final int BATCH_SIZE = 25; // 符合DynamoDB限制

    public static void main(String[] args) {
        // 初始化DynamoDB客户端
        try (DynamoDbClient ddbClient = DynamoDbClient.builder()
                .region(Region.US_EAST_1) // 替换为你的区域
                .credentialsProvider(DefaultCredentialsProvider.create())
                .build()) {

            // 步骤1：从GSI查询过期数据的主键（替换为你的实际查询逻辑）
            List<Map<String, AttributeValue>> expiredItemKeys = getExpiredItemKeysFromGSI(ddbClient);

            // 步骤2：分批处理删除请求
            for (int i = 0; i < expiredItemKeys.size(); i += BATCH_SIZE) {
                int endIndex = Math.min(i + BATCH_SIZE, expiredItemKeys.size());
                List<Map<String, AttributeValue>> batchKeys = expiredItemKeys.subList(i, endIndex);

                // 构造批量删除请求
                Map<String, List<WriteRequest>> requestItems = new HashMap<>();
                List<WriteRequest> writeRequests = new ArrayList<>();

                for (Map<String, AttributeValue> key : batchKeys) {
                    DeleteRequest deleteRequest = DeleteRequest.builder()
                            .key(key)
                            .build();
                    writeRequests.add(WriteRequest.builder().deleteRequest(deleteRequest).build());
                }

                requestItems.put(TABLE_NAME, writeRequests);
                BatchWriteItemRequest batchWriteRequest = BatchWriteItemRequest.builder()
                        .requestItems(requestItems)
                        .build();

                // 发送请求并处理未完成的项
                BatchWriteItemResponse response = ddbClient.batchWriteItem(batchWriteRequest);
                handleUnprocessedItems(ddbClient, response.unprocessedItems());
            }

            System.out.println("过期数据批量删除完成");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 替换为你的GSI查询逻辑：返回所有过期数据的主键集合
    private static List<Map<String, AttributeValue>> getExpiredItemKeysFromGSI(DynamoDbClient ddbClient) {
        List<Map<String, AttributeValue>> keys = new ArrayList<>();

        // 示例：假设表的分区键是"id"，排序键是"timestamp"（根据你的表结构调整）
        // 这里编写QueryRequest查询GSI的代码，遍历结果提取主键
        Map<String, AttributeValue> key1 = new HashMap<>();
        key1.put("id", AttributeValue.builder().s("item-id-1").build());
        key1.put("timestamp", AttributeValue.builder().n("1609459200").build());
        keys.add(key1);

        // 更多主键...
        return keys;
    }

    // 循环重试未完成的批量操作，直到全部完成
    private static void handleUnprocessedItems(DynamoDbClient ddbClient, Map<String, List<WriteRequest>> unprocessedItems) {
        while (!unprocessedItems.isEmpty()) {
            System.out.println("存在未处理的删除请求，进行重试...");
            BatchWriteItemRequest retryRequest = BatchWriteItemRequest.builder()
                    .requestItems(unprocessedItems)
                    .build();
            BatchWriteItemResponse retryResponse = ddbClient.batchWriteItem(retryRequest);
            unprocessedItems = retryResponse.unprocessedItems();
        }
    }
}

关键注意事项 #

• 主键要求：DeleteRequest中的key必须是表的分区键+排序键（如果有），不能用GSI的键，DynamoDB只能通过主键删除项。
• 重试逻辑：batchWriteItem不保证原子性，部分失败会返回UnprocessedItems，必须循环重试这些项直到为空。
• 性能优化：如果数据量极大，建议添加重试间隔（比如Thread.sleep），避免触发DynamoDB限流。
• SDK版本：示例采用AWS SDK v2，这是官方推荐的新版本，v1 API结构类似但不再优先维护。

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