能否仅用Sort Key查询DynamoDB表？Enhanced SDK查询方案咨询

问题背景 #

你创建的DynamoDB表结构如下：

aws dynamodb --endpoint-url=http://localhost:4566 create-table \
    --table-name localTable \
    --attribute-definitions \
        AttributeName=id,AttributeType=S \
        AttributeName=date,AttributeType=S \
    --key-schema \
        AttributeName=id,KeyType=HASH \
        AttributeName=date,KeyType=RANGE \
--provisioned-throughput \
        ReadCapacityUnits=10,WriteCapacityUnits=5

核心需求：

• 按date范围批量拉取1000条数据做处理
• 按id（如correlation-id、traceId）单条查询用于调试
• 已创建索引但不知道如何通过Enhanced SDK调用

现有代码的问题 #

1. 按traceId查询的逻辑完全错误：
   • 用UUID.randomUUID().toString()作为分区键值，完全不匹配实际数据，不可能查到结果
   • Scan操作中exclusiveStartKey用法错误：该参数是分页起始键（必须是表的主键id+date），不是查询条件；且HashMap.put返回旧值，直接强转会导致start为null
2. 未利用索引实现高效查询：要按date查询，必须通过全局二级索引（GSI），但代码中未初始化索引对应的表对象
3. 方法名与逻辑不匹配：read(String primaryKey, String secondaryKey)注释标注为按日期范围查询，但实际是通过主键id+date单条获取，不符合需求
4. 异步写入的潜在问题：putItemWithResponse返回的响应被忽略，若需要处理旧值或写入结果，当前逻辑无法实现

正确实现方案 #

第一步：创建全局二级索引（GSI） #

如果还未创建，执行以下命令创建以date为分区键的GSI：

aws dynamodb --endpoint-url=http://localhost:4566 update-table \
    --table-name localTable \
    --attribute-definitions AttributeName=date,AttributeType=S \
    --global-secondary-index-updates '[
        {
            "Create": {
                "IndexName": "date-index",
                "KeySchema": [{"AttributeName": "date", "KeyType": "HASH"}],
                "Projection": {"ProjectionType": "ALL"},
                "ProvisionedThroughput": {"ReadCapacityUnits": 10, "WriteCapacityUnits": 5}
            }
        }
    ]'

第二步：实体类映射配置 #

确保Payload类通过注解完成主表与GSI的字段映射：

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;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondaryPartitionKey;

@DynamoDbBean
public class Payload {
    private String id;
    private String date;
    // 其他字段：jsonString、processDetails等

    @DynamoDbPartitionKey
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }

    @DynamoDbSortKey
    public String getDate() { return date; }
    public void setDate(String date) { this.date = date; }

    // 绑定GSI的分区键
    @DynamoDbSecondaryPartitionKey(indexNames = "date-index")
    public String getDate() { return date; }

    // 其他字段的getter/setter
}

第三步：初始化索引表对象 #

在初始化主表的同时，初始化GSI对应的表对象：

import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient;
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable;
import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import java.net.URI;

// 初始化DynamoDB客户端
DynamoDbClient dynamoDbClient = DynamoDbClient.builder()
        .endpointOverride(URI.create("http://localhost:4566"))
        .region(Region.US_EAST_1)
        .build();

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

// 主表对象
DynamoDbTable<Payload> productTable = enhancedClient.table("localTable", TableSchema.fromBean(Payload.class));
// GSI表对象
DynamoDbTable<Payload> dateIndexTable = productTable.index("date-index");

第四步：按date范围批量查询实现 #

import java.util.Collection;
import java.util.stream.Collectors;
import software.amazon.awssdk.enhanced.dynamodb.model.QueryConditional;
import software.amazon.awssdk.enhanced.dynamodb.model.QueryEnhancedRequest;

/**
 * 按日期范围批量拉取数据（最多1000条）
 * @param startDate 起始日期（需与表中存储格式一致，如ISO8601格式：2024-01-01T00:00:00Z）
 * @param endDate 结束日期
 * @return 符合条件的Payload集合
 */
@Override
public Collection<Payload> readByDateRange(String startDate, String endDate) {
    QueryConditional queryConditional = QueryConditional.sortBetween(
            Key.builder().partitionValue(startDate).build(),
            Key.builder().partitionValue(endDate).build()
    );

    QueryEnhancedRequest request = QueryEnhancedRequest.builder()
            .queryConditional(queryConditional)
            .scanIndexForward(false) // 倒序查询，最新数据优先
            .limit(1000) // 限制最多返回1000条
            .consistentRead(false)
            .build();

    return dateIndexTable.query(request)
            .stream()
            .flatMap(page -> page.items().stream())
            .collect(Collectors.toList());
}

第五步：按id单条查询实现 #

假设id是表的HASH键，若每个id对应多条记录，会返回第一条匹配结果：

import java.util.Optional;
import software.amazon.awssdk.enhanced.dynamodb.model.QueryConditional;

/**
 * 按id（traceId/correlation-id）单条查询
 * @param id 主键id
 * @return 对应的Payload，不存在则返回null
 */
@Override
public Payload readById(String id) {
    QueryConditional queryConditional = QueryConditional.keyEqualTo(
            Key.builder().partitionValue(id).build()
    );

    return productTable.query(queryConditional)
            .stream()
            .flatMap(page -> page.items().stream())
            .findFirst()
            .orElse(null);
}

如果traceId是普通字段而非主键，需为其单独创建GSI，再用类似上述的查询逻辑。

第六步：修复异步写入方法 #

import java.util.concurrent.CompletableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

private static final Logger log = LoggerFactory.getLogger(YourClassName.class);

private void addItem(Payload payload) {
    CompletableFuture.runAsync(() -> {
        try {
            // 简单写入（无需返回旧值）
            productTable.putItem(payload);
            
            // 若需要返回旧值，使用以下逻辑：
            // PutItemEnhancedRequest<Payload> request = PutItemEnhancedRequest.builder(Payload.class)
            //         .item(payload)
            //         .returnValues(ReturnValue.ALL_OLD)
            //         .build();
            // PutItemResponse response = productTable.putItemWithResponse(request).response();
            // 此处可处理response
        } catch (Exception ex) {
            log.error("Unable to save item", ex);
        }
    });
}

关键说明 #

• GSI是高效查询非主键字段的唯一方案：Scan操作性能极低，不适合生产环境，尤其是数据量较大时
• Query操作规则：主表Query必须指定HASH键，GSI的Query必须指定GSI的HASH键，Range条件可选
• 分页处理：若查询结果超过1000条，需通过lastEvaluatedKey设置exclusiveStartKey实现分页

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