如何结合SAM Deploy实现DynamoDB数据的迁移与回滚？

核心思路 #

DynamoDB是非关系型数据库，无法直接复用SQL脚本模式，但可以借鉴SQL版本化迁移的思路，结合SAM的部署生命周期能力，实现自动化的数据升级与回滚。

具体实现方案 #

1. 版本化迁移脚本管理 #

将每个版本的升级/回滚逻辑封装为独立Lambda函数，按版本号命名（例如migrate-v1.1.0-upgrade、migrate-v1.1.0-downgrade），逻辑与你之前的SQL脚本对应：

• 升级逻辑：扫描/查询所有column1 = "x"的条目，更新为column1 = "y"并设置modified字段
• 回滚逻辑：查询column1 = "y"且modified = "whoReallyCaresIfThisWorks,YouGetTheIdea"的条目，恢复column1 = "x"并清空modified

示例Lambda代码片段（Python）：

# 升级脚本 v1.1.0
import boto3

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

def handler(event, context):
    # 分页处理大量数据，避免触发DynamoDB限流
    paginator = table.get_paginator('scan')
    for page in paginator.paginate(FilterExpression=boto3.dynamodb.conditions.Attr('column1').eq('x')):
        with table.batch_writer() as batch:
            for item in page['Items']:
                batch.put_item(
                    Item={
                        **item,
                        'column1': 'y',
                        'modified': 'whoReallyCaresIfThisWorks,YouGetTheIdea'
                    }
                )
    return {'status': 'success'}

回滚脚本只需调整过滤条件和更新值即可，保持幂等性（重复执行不会破坏数据）。

2. SAM模板集成迁移逻辑 #

方式一：部署钩子（推荐）

利用SAM的DeploymentPreference配置，在版本发布的流量切换前后自动触发迁移脚本：

Resources:
  MyServiceFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.11
      AutoPublishAlias: live
      DeploymentPreference:
        Type: Linear10PercentEvery1Minute
        PreTraffic: !Ref V110UpgradeHook
        RollbackConfiguration:
          RollbackLambda: !Ref V110DowngradeHook

  V110UpgradeHook:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: migrations/v1.1.0-upgrade/
      Handler: migrate.handler
      Runtime: python3.11
      Policies:
        - DynamoDBCrudPolicy:
            TableName: table_name

  V110DowngradeHook:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: migrations/v1.1.0-downgrade/
      Handler: rollback.handler
      Runtime: python3.11
      Policies:
        - DynamoDBCrudPolicy:
            TableName: table_name

该配置会在新版本流量切换前执行升级脚本，若部署失败触发回滚时，自动执行降级脚本。

方式二：自定义资源

通过CloudFormation自定义资源，在表创建/更新时触发迁移：

Resources:
  TargetDynamoDBTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: table_name
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5

  V110UpgradeFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: migrations/v1.1.0-upgrade/
      Handler: migrate.handler
      Runtime: python3.11
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref TargetDynamoDBTable

  MigrationTrigger:
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt V110UpgradeFunction.Arn
      TableName: !Ref TargetDynamoDBTable

自定义资源会在栈创建/更新时执行升级函数，回滚时可通过Lambda处理Delete事件触发降级逻辑。

3. 版本控制与幂等性保障 #

• 在DynamoDB中新增独立配置表（如schema_version），记录当前数据版本，迁移脚本执行前先校验版本，避免重复执行。
• 所有迁移/回滚操作必须保持幂等性，例如通过modified字段判断条目是否已被处理，防止重复更新。

4. 关键注意事项 #

• 分批处理数据：使用分页查询+批量写入，避免触发DynamoDB读写容量限制。
• 监控告警：为迁移函数配置CloudWatch告警，监控执行失败、超时等情况，及时触发回滚。
• 测试验证：在预发布环境充分测试迁移与回滚逻辑，确保数据一致性。

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