使用 Python 高效批量写入 DynamoDB：分步指南-Python教程-PHP中文网

使用 Python 高效批量写入 DynamoDB：分步指南

花韻仙語

发布： 2025-01-08 14:55:51

原创

415人浏览过

高效批量写入dynamodb的python指南

使用 Python 高效批量写入 DynamoDB：分步指南

对于处理大量数据的应用程序而言，高效地将数据插入AWS DynamoDB至关重要。本指南将逐步演示一个Python脚本，实现以下功能：

检查DynamoDB表是否存在: 如果不存在则创建。
生成随机测试数据: 用于模拟大规模数据插入。
批量写入数据: 利用batch_writer()提高性能和降低成本。

你需要安装boto3库：

<code class="bash">pip install boto3</code>

登录后复制

1. 设置DynamoDB表

首先，使用boto3初始化AWS会话并指定DynamoDB区域：

立即学习“Python免费学习笔记（深入）”；

<code class="python">import boto3
from botocore.exceptions import ClientError

# 初始化AWS会话
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')  # 指定区域

# DynamoDB表名
table_name = 'my_dynamodb_table'</code>

登录后复制

接下来，create_table_if_not_exists()函数检查表是否存在。如果不存在，则创建该表。本例中，表使用简单的分区键id创建。

笔目鱼英文论文写作器

写高质量英文论文，就用笔目鱼

查看详情

<code class="python">def create_table_if_not_exists():
    try:
        table = dynamodb.Table(table_name)
        table.load()  # 加载表元数据
        print(f"表 '{table_name}' 已存在。")
        return table
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            print(f"表 '{table_name}' 未找到。正在创建新表...")
            table = dynamodb.create_table(
                TableName=table_name,
                KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],  # 分区键
                AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],  # 字符串类型
                ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
            )
            table.meta.client.get_waiter('table_exists').wait(TableName=table_name)
            print(f"表 '{table_name}' 创建成功。")
            return table
        else:
            print(f"检查或创建表时出错: {e}")
            raise</code>

登录后复制

2. 生成随机数据

本例生成包含id、name、timestamp和value的随机记录。id是16个字符的随机字符串，value是1到1000之间的随机整数。

<code class="python">import random
import string
from datetime import datetime

# 生成随机字符串
def generate_random_string(length=10):
    return ''.join(random.choices(string.ascii_letters + string.digits, k=length))

# 生成随机记录
def generate_record():
    return {
        'id': generate_random_string(16),  # 唯一ID
        'name': generate_random_string(8),  # 随机名称
        'timestamp': str(datetime.utcnow()),  # 时间戳
        'value': random.randint(1, 1000),  # 随机值
    }</code>

登录后复制

3. 批量写入数据

使用DynamoDB的batch_writer()批量写入记录，而不是逐条写入，从而提高效率。此方法允许一次最多写入25条记录。

<code class="python"># 批量写入记录
def batch_write(table, records):
    with table.batch_writer() as batch:
        for record in records:
            batch.put_item(Item=record)</code>

登录后复制

4. 主要流程

创建表和生成记录的功能完成后，定义主流程：

创建表（如果不存在）。
生成1000条随机记录。
将它们分成25条一组写入DynamoDB。

<code class="python">def main():
    table = create_table_if_not_exists()

    records_batch = []
    for i in range(1, 1001):  # 生成1000条记录
        record = generate_record()
        records_batch.append(record)
        if len(records_batch) == 25:
            batch_write(table, records_batch)
            records_batch = []
            print(f"已写入 {i} 条记录")

    if records_batch:
        batch_write(table, records_batch)
        print(f"已写入剩余 {len(records_batch)} 条记录")

if __name__ == '__main__':
    main()</code>

登录后复制