将AWS Amplify应用从DynamoDB迁移至PostgreSQL Aurora Serverless V1

一、数据迁移工具与步骤 #

可用工具/AWS服务 #

• AWS Glue：适合复杂数据结构转换的ETL场景，自动爬取DynamoDB schema，生成转换脚本将数据导入Aurora，支持批量处理和格式转换。
• AWS Data Pipeline：构建可视化迁移管道，支持DynamoDB直接导出到Aurora，或先导出到S3再导入，适合结构化数据迁移。
• 自定义脚本：用AWS SDK（如boto3）读取DynamoDB数据，通过psycopg2写入Aurora，适合小数据集或需要自定义转换逻辑的场景。

迁移步骤示例 #

1. 导出DynamoDB数据：使用DynamoDB原生导出功能将数据导出到S3（可选，用于批量处理）。
2. 数据转换：用AWS Glue爬虫扫描DynamoDB或S3中的数据，生成表定义，创建Glue作业将DynamoDB的Map/List类型转换为PostgreSQL的JSONB/ARRAY类型。
3. 导入Aurora：通过Glue作业或Data Pipeline将转换后的数据写入Aurora Serverless集群，注意控制并发量避免触发限流。

二、兼容性注意事项 #

• 数据类型映射：
  • DynamoDB Number → PostgreSQL INT/FLOAT/NUMERIC
  • DynamoDB String → PostgreSQL VARCHAR/TEXT
  • DynamoDB Map → PostgreSQL JSONB
  • DynamoDB List → PostgreSQL ARRAY
  • 处理DynamoDB的空值：PostgreSQL部分字段不允许空，需提前填充默认值或修改表结构。
• Schema重构：DynamoDB单表设计需拆分为PostgreSQL的多表关系，建立外键、索引以优化查询性能。
• 主键适配：DynamoDB的分区键+排序键可对应PostgreSQL的复合主键，确保唯一性约束。
• 时间戳转换：将DynamoDB的ISO格式时间字符串转换为PostgreSQL的TIMESTAMP类型。

三、更新AWS Amplify配置指向Aurora #

1. 添加RDS数据源：进入Amplify控制台，在Data模块点击「Add data source」，选择「Amazon RDS」，关联你的Aurora Serverless集群。
2. 配置网络权限：
   • 确保Aurora安全组开放5432端口，允许Amplify所在VPC或IP访问。
   • 若Aurora在私有VPC，需在Amplify的「Settings > VPC」中配置VPC peering。
3. 权限配置：给Amplify服务角色添加rds:Connect、rds:Query等权限，或启用Aurora的IAM数据库认证，用IAM角色生成临时凭证连接。
4. 代码更新：
   • 若使用Amplify DataStore，需重新定义GraphQL schema适配PostgreSQL结构。
   • 自定义API调用需替换原DynamoDB逻辑，改用API Gateway + Lambda连接Aurora，或直接通过psycopg2调用。

四、迁移失败排查与修复 #

数据传输失败/应用崩溃的可能原因 #

• 网络不通：Aurora安全组未开放Amplify的访问端口，或未配置VPC peering导致无法连接。
• 权限不足：Amplify角色缺少Aurora访问权限，或IAM数据库认证配置错误。
• 数据类型不兼容：未转换DynamoDB数据格式，如将List写入PostgreSQL非ARRAY字段，导致插入失败。
• 连接配置错误：Amplify的数据库连接字符串（主机地址、端口、用户名/密码）有误。
• Aurora限流：迁移数据量过大触发Aurora Serverless自动扩容延迟，导致连接超时或写入失败。

无缝迁移的Amplify特定配置 #

• 测试连接：在Amplify控制台测试数据源连接，确保能成功访问Aurora。
• 蓝绿部署：先在测试环境验证迁移后的功能，再逐步切换用户流量到Aurora后端，避免全量切换风险。
• 日志排查：开启Amplify API日志和Aurora查询日志，定位具体错误（如SQL语法错误、权限问题）。
• IAM认证：启用Aurora IAM数据库认证，避免硬编码密码，提升安全性。

自定义迁移脚本示例 #

import boto3
import psycopg2
from psycopg2.extras import execute_values

# 读取DynamoDB数据（分页处理）
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your-dynamodb-table')
items = []
response = table.scan()
items.extend(response['Items'])
while 'LastEvaluatedKey' in response:
    response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
    items.extend(response['Items'])

# 连接Aurora PostgreSQL
conn = psycopg2.connect(
    host='your-aurora-cluster-endpoint',
    port=5432,
    user='your-db-user',
    password='your-db-password',
    dbname='your-db-name'
)
cur = conn.cursor()

# 转换数据并批量写入
insert_query = """
INSERT INTO target_table (id, username, profile_data, created_at)
VALUES %s
"""
records = []
for item in items:
    records.append(
        (
            item['id'],
            item['username'],
            item.get('profile', {}),  # 自动转为PostgreSQL JSONB类型
            item.get('created_at')
        )
    )

execute_values(cur, insert_query, records)
conn.commit()

cur.close()
conn.close()

内容的提问来源于stack exchange，提问作者Furkan Yılmaz