DynamoDB 自动增量

Question

由于 DynamoDB 本身并不像 MySQL 那样支持“user_id”的顺序自动递增；将“user_id”按顺序增加到下一个的最佳方法是什么？

我正在查询最新的“user_id”并递增它；然而，这感觉很蹩脚。

有更好的方法吗？ :-)

def get_latest_user_id():
    try:
        # Scan the table to find the highest user_id
        response = table.scan(
            ProjectionExpression="user_id",
            FilterExpression=Attr("user_id").gt(0),
            Limit=1,
            ScanIndexForward=False  # Sort in descending order
        )
        items = response.get('Items', [])
        if not items:
            return 0  # Return 0 if table is empty
        return max(item['user_id'] for item in items)
    except ClientError as e:
        logger.error(f"Error getting latest user ID: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

Answer 1

有两种方法可以做到这一点。

1. 使用原子计数器

生成不断增加的序列号的第一种方法是使用原子计数器。这是一个两步过程。首先，发出增加计数器的请求并在响应中接收新值。其次，在后续写入中使用该新值。

table = boto3.resource('dynamodb').Table('orders')

# Add one to the counter and ask for the new value to be returned
response = table.update_item(
    Key={'pk': 'orderCounter'},
    UpdateExpression="ADD #cnt :val",
    ExpressionAttributeNames={'#cnt': 'count'},
    ExpressionAttributeValues={':val': 1},
    ReturnValues="UPDATED_NEW"
)

# Retrieve the new value
nextOrderId = response['Attributes']['count']

# Use the new value
table.put_item(
    Item={'pk': str(nextOrderId), 'deliveryMethod' : 'expedited'}
)

此设计不存在竞争条件，因为对 DynamoDB 中单个项目的所有写入都是串行应用的。这可确保每个计数器值永远不会返回多次。

此方法的成本是更新计数器项目的 1 次写入，加上存储新项目的通常写入成本。这种方法的最大吞吐量受到计数器项的限制。 DynamoDB 中单个小项目的最大吞吐量与分区的最大吞吐量相同。

1. 使用排序键：

第二种方法使用排序键内的最大值 用于跟踪该项目的最大序列值的项目集合 收藏。

存储在 DynamoDB 表中的项目可以将分区键和可选的排序键作为其主键的一部分。项目集合中的项目具有相同的分区键但不同的排序键。 DynamoDB 查询可以以项目集合为目标来检索集合中的所有项目，或者可以提供排序键条件来检索子集。

通过设计排序键来表示序列中项目的值，可以有效地使用查询来检索序列的最大值。

import boto3
from boto3.dynamodb.conditions import Key

PROJECT_ID = 'projectA'

dynamo = boto3.resource('dynamodb')
client = dynamo.Table('projects')
highestIssueId = 0
saved = False

# Query for the last sorted value in the given item collection
response = client.query(
    KeyConditionExpression=Key('pk').eq(PROJECT_ID),
    ScanIndexForward=False,
    Limit=1
)

# Retrieve the sort key value
if response['Count'] > 0:
    highestIssueId = int(response['Items'][0]['sk'])

while not saved:
    try:
        # Write using the next value in the sequence, but only if the item doesn’t exist
        response = client.put_item(
            Item={
                'pk': PROJECT_ID, 
                'sk' : highestIssueId + 1, 
                'priority' : 'low'
            },
            ConditionExpression='attribute_not_exists(pk)'
        )
        saved = True
    # An exception indicates we lost a race condition, so increment the value and loop again
    except dynamo.meta.client.exceptions.ConditionalCheckFailedException as e:
         highestIssueId = highestIssueId + 1

参考：https://aws.amazon.com/blogs/database/implement-auto-increment-with-amazon-dynamodb/