本指南演示了使用 Python 将数据高效插入 AWS DynamoDB，重点关注大型数据集。 我们将介绍：表创建（如果需要）、随机数据生成和批量写入，以实现最佳性能和节省成本。 需要 boto3 库；使用pip install boto3安装它。
1。 DynamoDB 表设置：
首先，我们建立一个 AWS 会话并定义 DynamoDB 表的区域：
<code class="language-python">import boto3
from botocore.exceptions import ClientError
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table_name = 'My_DynamoDB_Table_Name'</code>create_table_if_not_exists() 函数检查表是否存在，如果不存在，则使用主键 (id) 创建它：
<code class="language-python">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</code>2。随机数据生成：
我们将使用 id、name、timestamp 和 value 生成示例记录：
<code class="language-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),
'name': generate_random_string(8),
'timestamp': str(datetime.utcnow()),
'value': random.randint(1, 1000)
}</code>3。批量数据写入：
batch_write() 函数利用 DynamoDB 的 batch_writer() 进行高效的批量插入（每批最多 25 个项目）：
<code class="language-python">def batch_write(table, records):
with table.batch_writer() as batch:
for record in records:
batch.put_item(Item=record)</code>4。主要工作流程：
main函数统筹建表、数据生成、批量写入：
<code class="language-python">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()</code>5。结论：
此脚本利用批量写入来优化大量数据的 DynamoDB 交互。 请记住调整参数（批量大小、记录计数等）以满足您的特定需求。 考虑探索高级 DynamoDB 功能以进一步增强性能。
以上是使用 Python 高效批量写入 DynamoDB：分步指南的详细内容。更多信息请关注PHP中文网其他相关文章！
