使用 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中文網其他相關文章！