如何将DynamoDB数据迁移至Aurora PostgreSQL？求可行方案

一、有效从DynamoDB获取数据的工具/方法 #

• DynamoDB导出到S3：AWS官方推荐的批量导出方式，支持按需一次性导出或连续流式导出到S3，导出数据为JSON格式，适合大规模数据迁移。操作直接在DynamoDB控制台完成，无需编写复杂脚本。
• DynamoDB Streams：用于增量迁移或实时同步场景，可捕获DynamoDB的所有变更操作（插入、更新、删除），配合Lambda或Kinesis Data Firehose就能把变更数据传递到后续处理环节。
• AWS Glue爬虫：通过爬虫自动扫描DynamoDB表并生成数据目录，后续可直接用Glue ETL作业读取这些数据，比手动写脚本高效得多。

二、两类数据库迁移的最佳实践 #

• 先全量再增量：先用DynamoDB导出到S3完成全量数据迁移，再通过DynamoDB Streams配合Lambda/Glue处理增量数据，避免长时间业务中断或锁表问题。
• 数据验证前置：迁移前抽样检查DynamoDB数据的结构、格式，提前发现嵌套属性、数据类型差异等问题，避免迁移后出现数据丢失或错误。
• 优先用AWS原生服务链：采用DynamoDB → S3 → Aurora PostgreSQL的集成链路，Aurora支持直接从S3导入CSV/JSON数据，无需额外中转工具，降低复杂度。
• 大表分批处理：如果DynamoDB表数据量超过100GB，建议分批次导出，或用Glue的分区功能拆分处理，避免单次迁移失败导致重复工作。

三、解决DynamoDB与Aurora PostgreSQL的结构不匹配问题 #

• 扁平化嵌套属性：DynamoDB的嵌套JSON结构，可拆分为PostgreSQL的单独列，或直接用PostgreSQL的jsonb类型存储。比如DynamoDB中{"user": {"name": "foo", "age": 20}}，可拆成user_name和user_age列，或存为user jsonb字段。
• 对齐数据类型：DynamoDB的Number对应PostgreSQL的numeric/int，String对应varchar/text，Boolean对应boolean；注意DynamoDB的NULL值要映射为PostgreSQL的空值，避免插入失败。
• 主键映射：将DynamoDB的分区键+排序键映射到PostgreSQL的主键或唯一约束，比如把分区键设为PostgreSQL主键，排序键设为辅助索引。
• ETL工具做转换：用AWS Glue的PySpark脚本处理数据转换，示例代码：

from pyspark.sql.functions import col

# 扁平化嵌套字段
df = df.withColumn("user_name", col("user.name")) \
       .withColumn("user_age", col("user.age")) \
       .drop("user")

# 转换数据类型
df = df.withColumn("user_age", col("user_age").cast("integer"))

四、实操示例与经验分享 #

全量迁移步骤示例 #

1. 在DynamoDB控制台选择目标表，点击「导出到S3」，选择JSON格式，指定加密方式和目标S3桶；
2. 导出完成后，在Aurora PostgreSQL执行COPY命令导入数据：

COPY target_table FROM 's3://your-bucket/export-path/'
WITH (FORMAT 'json', REGION 'us-east-1', CREDENTIALS 'aws_iam_role=arn:aws:iam::account-id:role/your-role');

3. 若JSON格式导入有兼容性问题，可先用Glue将JSON转成CSV，再用COPY命令导入CSV。

增量同步示例 #

1. 开启DynamoDB表的Streams，选择「新图像+旧图像」捕获模式；
2. 创建Lambda函数，触发源设为DynamoDB Streams，函数内解析记录并操作Aurora：

import psycopg2

def lambda_handler(event, context):
    conn = psycopg2.connect(
        host="your-aurora-endpoint",
        database="db-name",
        user="user",
        password="password"
    )
    cur = conn.cursor()

    for record in event['Records']:
        if record['eventName'] == 'INSERT':
            item = record['dynamodb']['NewImage']
            # 解析DynamoDB格式属性（S=字符串，N=数字）
            user_id = item['user_id']['S']
            user_name = item['user_name']['S']
            cur.execute("INSERT INTO users (id, name) VALUES (%s, %s)", (user_id, user_name))
        # 同理处理UPDATE和DELETE事件
    
    conn.commit()
    cur.close()
    conn.close()

踩坑经验 #

• Lambda连接Aurora失败大概率是VPC/安全组配置问题：要确保Lambda在Aurora所在的VPC内，且Aurora安全组允许Lambda的IP访问；
• Glue作业处理大量DynamoDB数据时，要开启分页读取和并行执行，避免超时；
• 迁移后必须做数据校验：对比DynamoDB和PostgreSQL的行数，抽样检查字段值，确保数据一致性。

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