如何使用Python botot3将AWS DynamoDB表数据导入Microsoft SQL Server？

完全可以实现这个需求，核心思路是用Boto3读取DynamoDB中的数据，再通过Python的SQL Server驱动（如pyodbc）将数据写入目标库。以下是具体实现步骤和代码示例：

1. 安装依赖库 #

首先安装所需的Python包：

pip install boto3 pyodbc

2. 配置AWS访问权限 #

确保你的运行环境已配置AWS凭证（用于访问DynamoDB），可以通过以下方式之一：

• 环境变量：设置AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY
• 本地配置文件：在~/.aws/credentials中添加凭证信息
• IAM角色：如果代码运行在AWS EC2/ECS等服务上，直接关联具有DynamoDB读取权限的IAM角色

3. 读取DynamoDB数据 #

使用Boto3的scan()或query()方法读取数据，注意处理大表的分页逻辑（DynamoDB单次扫描最多返回1MB数据）：

import boto3

def read_dynamodb_table(table_name):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)
    
    items = []
    response = table.scan()
    items.extend(response['Items'])
    
    # 处理分页
    while 'LastEvaluatedKey' in response:
        response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
        items.extend(response['Items'])
    
    return items

4. 连接SQL Server并写入数据 #

使用pyodbc建立SQL Server连接，将读取到的DynamoDB数据批量写入目标表。注意根据DynamoDB的数据类型转换为SQL Server兼容的类型：

import pyodbc

def write_to_sql_server(items, db_config):
    # 构建连接字符串
    conn_str = (
        f"DRIVER={{ODBC Driver 17 for SQL Server}};"
        f"SERVER={db_config['server']};"
        f"DATABASE={db_config['database']};"
        f"UID={db_config['username']};"
        f"PWD={db_config['password']}"
    )
    
    with pyodbc.connect(conn_str) as conn:
        cursor = conn.cursor()
        
        # 示例：假设DynamoDB表有id（字符串）、name（字符串）、age（数字）字段
        # 先创建目标表（如果不存在）
        create_table_query = """
        IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='target_table' AND xtype='U')
        CREATE TABLE target_table (
            id VARCHAR(255) PRIMARY KEY,
            name VARCHAR(255),
            age INT
        )
        """
        cursor.execute(create_table_query)
        
        # 批量插入数据
        insert_query = "INSERT INTO target_table (id, name, age) VALUES (?, ?, ?)"
        # 转换DynamoDB数据类型（比如Number转int）
        values = [(item['id'], item['name'], int(item['age'])) for item in items]
        cursor.executemany(insert_query, values)
        
        conn.commit()

5. 主函数整合 #

def main():
    # 配置参数
    dynamodb_table_name = 'your-dynamodb-table'
    sql_server_config = {
        'server': 'your-sql-server-host',
        'database': 'your-db-name',
        'username': 'db-username',
        'password': 'db-password'
    }
    
    # 读取数据
    dynamodb_items = read_dynamodb_table(dynamodb_table_name)
    if not dynamodb_items:
        print("DynamoDB表中无数据")
        return
    
    # 写入SQL Server
    write_to_sql_server(dynamodb_items, sql_server_config)
    print("数据导入完成")

if __name__ == "__main__":
    main()

关键注意事项 #

• 数据类型转换：DynamoDB的Number类型存储为字符串，需要手动转换为int/float；Boolean对应SQL Server的BIT类型等。
• 性能优化：对于超大型表，可使用DynamoDB的并行扫描提升读取速度；写入时采用批量提交减少网络开销。
• 错误处理：建议添加异常捕获（如try-except），处理连接失败、数据写入失败等情况，必要时实现重试机制。

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