如何通过AppSync resolver获取DynamoDB特定区域最新用户版本？

要实现需求，有两种可行方案，可根据业务场景选择：

方案一：查询后在解析器中过滤结果 #

这种方案无需修改现有写入逻辑，通过AppSync的VTL模板对Query返回的所有结果进行分组处理，保留每个name下版本号最大的条目。

步骤1：调整DynamoDB查询请求 #

修改Query参数，确保结果按版本号降序排列，让同name的最新版本条目优先返回：

{
    "operation": "Query",
    "index": "RegionIndex",
    "keyConditionExpression": "region = :region",
    "expressionValues": util.dynamodb.toMapValues({ ":region": ${region} }),
    "select": "ALL_ATTRIBUTES",
    "scanIndexForward": false,
    "projectionExpression": "name, version, region"
}

说明：如果你的GSI排序键不是version，建议调整GSI结构为region（PK） + name#version（SK），确保同name的条目能按版本号排序。

步骤2：在响应模板中处理结果 #

在AppSync解析器的响应模板中添加以下VTL代码，完成分组去重：

#set($latestEntries = {})
#foreach($item in $ctx.result.items)
    #set($currentName = $item.name.S)
    #set($currentVersion = $util.toInteger($item.version.N))
    
    #if(!$latestEntries.containsKey($currentName) || $currentVersion > $util.toInteger($latestEntries.get($currentName).version.N))
        $latestEntries.put($currentName, $item)
    #end
#end

#set($finalResult = [])
#foreach($entry in $latestEntries.entrySet())
    $finalResult.add($entry.getValue())
#end

$util.toJson($finalResult)

方案二：写入时维护最新版本（性能更优） #

若数据量较大，方案一的全量查询+过滤会影响性能，建议在写入数据时就维护每个实体的最新版本标记，后续查询可直接精准获取。

步骤1：调整表结构 #

在原表中新增布尔类型属性isLatest，用于标记该条目是否为当前name的最新版本。

步骤2：写入时的事务更新逻辑 #

在写入新版本的解析器请求中，通过事务确保只有当新版本号更大时才写入，同时将旧的最新版本标记为非最新：

{
    "operation": "TransactWriteItems",
    "transactItems": [
        // 写入新版本并标记为最新
        {
            "Put": {
                "TableName": "Users",
                "Item": {
                    "name": $util.dynamodb.toDynamoDBJson($ctx.args.name),
                    "version": $util.dynamodb.toDynamoDBJson($ctx.args.version),
                    "region": $util.dynamodb.toDynamoDBJson($ctx.args.region),
                    "isLatest": $util.dynamodb.toDynamoDBJson(true)
                },
                "ConditionExpression": "attribute_not_exists(name) OR version < :newVersion",
                "ExpressionAttributeValues": {
                    ":newVersion": $util.dynamodb.toDynamoDBJson($ctx.args.version)
                }
            }
        },
        // 将旧的最新版本取消标记
        {
            "Update": {
                "TableName": "Users",
                "Key": {
                    "name": $util.dynamodb.toDynamoDBJson($ctx.args.name)
                },
                "UpdateExpression": "SET isLatest = :falseVal",
                "ConditionExpression": "isLatest = :trueVal AND version < :newVersion",
                "ExpressionAttributeValues": {
                    ":trueVal": $util.dynamodb.toDynamoDBJson(true),
                    ":falseVal": $util.dynamodb.toDynamoDBJson(false),
                    ":newVersion": $util.dynamodb.toDynamoDBJson($ctx.args.version)
                }
            }
        }
    ]
}

步骤3：查询最新版本 #

此时只需精准查询标记为最新的条目即可：

{
    "operation": "Query",
    "index": "RegionIndex",
    "keyConditionExpression": "region = :region",
    "filterExpression": "isLatest = :true",
    "expressionValues": util.dynamodb.toMapValues({ 
        ":region": ${region},
        ":true": true
    }),
    "select": "ALL_ATTRIBUTES"
}

方案对比 #

• 方案一：实现简单，无需改动写入流程，但数据量大时会返回冗余数据，解析器处理耗时增加。
• 方案二：查询性能最优，适合大规模数据场景，但需要额外的写入事务逻辑。

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