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

本指南演示了使用 Python 将数据高效插入 AWS DynamoDB，重点关注大型数据集。 我们将介绍：表创建（如果需要）、随机数据生成和批量写入，以实现最佳性能和节省成本。 需要 boto3 库；使用pip install boto3安装它。

1。 DynamoDB 表设置：

首先，我们建立一个 AWS 会话并定义 DynamoDB 表的区域：

import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table_name = 'My_DynamoDB_Table_Name'

create_table_if_not_exists() 函数检查表是否存在，如果不存在，则使用主键 (id) 创建它：

def create_table_if_not_exists():
    try:
        table = dynamodb.Table(table_name)
        table.load()
        print(f"Table '{table_name}' exists.")
        return table
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            print(f"Creating table '{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 '{table_name}' created.")
            return table
        else:
            print(f"Error: {e}")
            raise

2。随机数据生成：

我们将使用 id、name、timestamp 和 value 生成示例记录：

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),
        'name': generate_random_string(8),
        'timestamp': str(datetime.utcnow()),
        'value': random.randint(1, 1000)
    }

3。批量数据写入：

batch_write() 函数利用 DynamoDB 的 batch_writer() 进行高效的批量插入（每批最多 25 个项目）：

def batch_write(table, records):
    with table.batch_writer() as batch:
        for record in records:
            batch.put_item(Item=record)

4。主要工作流程：

main函数统筹建表、数据生成、批量写入：

def main():
    table = create_table_if_not_exists()
    records_batch = []
    for i in range(1, 1001):
        record = generate_record()
        records_batch.append(record)
        if len(records_batch) == 25:
            batch_write(table, records_batch)
            records_batch = []
            print(f"Wrote {i} records")
    if records_batch:
        batch_write(table, records_batch)
        print(f"Wrote remaining {len(records_batch)} records")

if __name__ == '__main__':
    main()

5。结论：

此脚本利用批量写入来优化大量数据的 DynamoDB 交互。 请记住调整参数（批量大小、记录计数等）以满足您的特定需求。 考虑探索高级 DynamoDB 功能以进一步增强性能。

以上是使用 Python 高效批量写入 DynamoDB：分步指南的详细内容。更多信息请关注PHP中文网其他相关文章！