dynamodb

AWS DynamoDB

Amazon DynamoDB 是一项完全托管的 NoSQL 数据库服务，可在任何规模下提供快速、可预测的性能。它支持键值和文档数据结构。

目录

• 核心概念
• 常见模式
• CLI 参考
• 最佳实践
• 故障排查
• 参考资料

核心概念

键

键类型	描述
分区键（PK）	必需。决定数据分布
排序键（SK）	可选。支持在分区内进行范围查询
复合键	PK + SK 组合

二级索引

索引类型	描述
GSI（全局二级索引）	PK/SK 不同，吞吐量独立，最终一致
LSI（本地二级索引）	PK 相同、SK 不同，与表共享吞吐量，可选择强一致性

容量模式

模式	适用场景
按需	流量不可预测，按请求付费
预置	流量可预测，成本较低，可使用自动扩缩容

常见模式

创建表

AWS CLI：

aws dynamodb create-table \
  --table-name Users \
  --attribute-definitions \
    AttributeName=PK,AttributeType=S \
    AttributeName=SK,AttributeType=S \
  --key-schema \
    AttributeName=PK,KeyType=HASH \
    AttributeName=SK,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST

boto3：

import boto3

dynamodb = boto3.resource('dynamodb')

table = dynamodb.create_table(
    TableName='Users',
    KeySchema=[
        {'AttributeName': 'PK', 'KeyType': 'HASH'},
        {'AttributeName': 'SK', 'KeyType': 'RANGE'}
    ],
    AttributeDefinitions=[
        {'AttributeName': 'PK', 'AttributeType': 'S'},
        {'AttributeName': 'SK', 'AttributeType': 'S'}
    ],
    BillingMode='PAY_PER_REQUEST'
)

table.wait_until_exists()

基本 CRUD 操作

import boto3
from boto3.dynamodb.conditions import Key, Attr

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

# Put item
table.put_item(
    Item={
        'PK': 'USER#123',
        'SK': 'PROFILE',
        'name': 'John Doe',
        'email': '[email protected]',
        'created_at': '2024-01-15T10:30:00Z'
    }
)

# Get item
response = table.get_item(
    Key={'PK': 'USER#123', 'SK': 'PROFILE'}
)
item = response.get('Item')

# Update item
table.update_item(
    Key={'PK': 'USER#123', 'SK': 'PROFILE'},
    UpdateExpression='SET #name = :name, updated_at = :updated',
    ExpressionAttributeNames={'#name': 'name'},
    ExpressionAttributeValues={
        ':name': 'John Smith',
        ':updated': '2024-01-16T10:30:00Z'
    }
)

# Delete item
table.delete_item(
    Key={'PK': 'USER#123', 'SK': 'PROFILE'}
)

查询操作

# Query by partition key
response = table.query(
    KeyConditionExpression=Key('PK').eq('USER#123')
)

# Query with sort key condition
response = table.query(
    KeyConditionExpression=Key('PK').eq('USER#123') & Key('SK').begins_with('ORDER#')
)

# Query with filter
response = table.query(
    KeyConditionExpression=Key('PK').eq('USER#123'),
    FilterExpression=Attr('status').eq('active')
)

# Query with projection
response = table.query(
    KeyConditionExpression=Key('PK').eq('USER#123'),
    ProjectionExpression='PK, SK, #name, email',
    ExpressionAttributeNames={'#name': 'name'}
)

# Paginated query
paginator = dynamodb.meta.client.get_paginator('query')
for page in paginator.paginate(
    TableName='Users',
    KeyConditionExpression='PK = :pk',
    ExpressionAttributeValues={':pk': {'S': 'USER#123'}}
):
    for item in page['Items']:
        print(item)

批量操作

# Batch write (up to 25 items)
with table.batch_writer() as batch:
    for i in range(100):
        batch.put_item(Item={
            'PK': f'USER#{i}',
            'SK': 'PROFILE',
            'name': f'User {i}'
        })

# Batch get (up to 100 items)
dynamodb = boto3.resource('dynamodb')
response = dynamodb.batch_get_item(
    RequestItems={
        'Users': {
            'Keys': [
                {'PK': 'USER#1', 'SK': 'PROFILE'},
                {'PK': 'USER#2', 'SK': 'PROFILE'}
            ]
        }
    }
)

创建 GSI

aws dynamodb update-table \
  --table-name Users \
  --attribute-definitions AttributeName=email,AttributeType=S \
  --global-secondary-index-updates '[
    {
      "Create": {
        "IndexName": "email-index",
        "KeySchema": [{"AttributeName": "email", "KeyType": "HASH"}],
        "Projection": {"ProjectionType": "ALL"}
      }
    }
  ]'

条件写入

from botocore.exceptions import ClientError

# Only put if item doesn't exist
try:
    table.put_item(
        Item={'PK': 'USER#123', 'SK': 'PROFILE', 'name': 'John'},
        ConditionExpression='attribute_not_exists(PK)'
    )
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        print("Item already exists")

# Optimistic locking with version
table.update_item(
    Key={'PK': 'USER#123', 'SK': 'PROFILE'},
    UpdateExpression='SET #name = :name, version = version + :inc',
    ConditionExpression='version = :current_version',
    ExpressionAttributeNames={'#name': 'name'},
    ExpressionAttributeValues={
        ':name': 'New Name',
        ':inc': 1,
        ':current_version': 5
    }
)

CLI 参考

表操作

命令	描述
aws dynamodb create-table	创建表
aws dynamodb describe-table	获取表信息
aws dynamodb update-table	修改表/索引
aws dynamodb delete-table	删除表
aws dynamodb list-tables	列出所有表

项目操作

命令	描述
aws dynamodb put-item	创建/替换项目
aws dynamodb get-item	读取单个项目
aws dynamodb update-item	更新项目属性
aws dynamodb delete-item	删除项目
aws dynamodb query	按键查询
aws dynamodb scan	扫描整个表

批量操作

命令	描述
aws dynamodb batch-write-item	批量写入（最多 25 个）
aws dynamodb batch-get-item	批量读取（最多 100 个）
aws dynamodb transact-write-items	事务写入
aws dynamodb transact-get-items	事务读取

最佳实践

数据建模

• 围绕访问模式设计 — 设计前先了解查询需求
• 使用复合键 — PK 用于分组，SK 用于排序/筛选
• 优先使用 query，而非 scan — scan 开销较大
• 使用稀疏索引 — 只有包含索引属性的项目才会被索引
• 对相关实体考虑采用单表设计

性能

• 均匀分布分区键 — 避免热点分区
• 使用批量操作以减少 API 调用
• 为读密集型工作负载启用 DAX
• 使用投影以减少数据传输

成本优化

• 可变工作负载使用按需模式
• 可预测的工作负载使用预置容量 + 自动扩缩容
• 为过期数据设置 TTL
• 将冷数据归档到 S3

故障排查

限流

症状： ProvisionedThroughputExceededException

原因：

• 热点分区（键分布不均）
• 突发流量超出容量
• GSI 限流影响基表

解决方案：

# Use exponential backoff
import time
from botocore.config import Config

config = Config(
    retries={
        'max_attempts': 10,
        'mode': 'adaptive'
    }
)
dynamodb = boto3.resource('dynamodb', config=config)

热点分区

调试：

# Check consumed capacity by partition
aws cloudwatch get-metric-statistics \
  --namespace AWS/DynamoDB \
  --metric-name ConsumedReadCapacityUnits \
  --dimensions Name=TableName,Value=Users \
  --start-time $(date -d '1 hour ago' -u +%Y-%m-%dT%H:%M:%SZ) \
  --end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
  --period 60 \
  --statistics Sum

解决方案：

• 为分区键添加随机性
• 使用写分片
• 将访问分散到多个分区

查询未返回任何项目

调试检查清单：

1. 确认键值完全匹配（区分大小写）
2. 检查键类型（S、N、B）
3. 确认表/索引名称
4. 检查筛选表达式（它们会在读取之后应用）

扫描性能

问题： 扫描速度慢且成本高

解决方案：

• 对大型表使用并行扫描
• 为访问模式创建 GSI
• 使用筛选表达式减少返回的数据

# Parallel scan
import concurrent.futures

def scan_segment(segment, total_segments):
    return table.scan(
        Segment=segment,
        TotalSegments=total_segments
    )

with concurrent.futures.ThreadPoolExecutor() as executor:
    results = list(executor.map(
        lambda s: scan_segment(s, 4),
        range(4)
    ))

参考资料

• DynamoDB 开发人员指南
• DynamoDB API 参考
• DynamoDB CLI 参考
• boto3 DynamoDB
• DynamoDB 最佳实践