Ефективне записування даних пакетами в DynamoDB за допомогою Python: покрокова інструкція - javascript.org.ua

Ефективне пакетне записування в DynamoDB за допомогою Python: Покроковий посібник

При роботі з AWS DynamoDB, особливо для додатків, які обробляють великі обсяги даних, важливо забезпечити ефективне вставлення записів. У цьому пості ми розглянемо Python-скрипт, який демонструє, як:

Перевірити, чи існує таблиця DynamoDB, і створити її, якщо вона відсутня.
Згенерувати випадкові дані для таблиці.
Записувати дані пакетами в DynamoDB для підвищення продуктивності та зниження витрат.

Для взаємодії з DynamoDB ми будемо використовувати бібліотеку boto3, тому переконайтеся, що вона встановлена перед початком.

pip install boto3

1. Налаштування таблиці DynamoDB

Спочатку ми ініціалізуємо сесію з AWS, використовуючи boto3, та вказуємо регіон для DynamoDB:

import boto3  
from botocore.exceptions import ClientError  

# Ініціалізація сесії з AWS  
dynamodb = boto3.resource('dynamodb', region_name='us-east-1') # Вказуємо регіон  

# Вказуємо назву таблиці DynamoDB  
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_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

2. Генерація випадкових даних

Для цього прикладу ми згенеруємо випадкові записи з полями id, name, timestamp та value. id буде випадковим рядком з 16 символів, а value — випадковим цілим числом між 1 і 1000.

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), # Випадкове значення  
 }

3. Пакетне записування даних

Тепер, замість того щоб записувати записи один за одним, що може бути повільно та неефективно, ми скористаємося методом batch_writer() DynamoDB, щоб записувати записи пакетами. Цей метод дозволяє вставляти до 25 записів в одному пакеті.

# Функція для пакетного запису записів  
def batch_write(table, records):  
 with table.batch_writer() as batch:  
 for record in records:  
 batch.put_item(Item=record)

4. Основний робочий процес

Тепер, коли ми маємо функції для створення таблиці та генерації записів, ми можемо визначити основний робочий процес. Це буде:

Створити таблицю, якщо її ще немає.
Згенерувати 1000 випадкових записів.
3.
Записуйте їх у DynamoDB пакетами по 25.

def main():  
 # Створити таблицю, якщо вона не існує  
 table = create_table_if_not_exists()  

 records_batch = []  
 for i in range(1, 1001): # Цикл для створення 1000 записів  
 record = generate_record()  
 records_batch.append(record)  

 # Якщо розмір пакету досягне 25 елементів, записуємо в DynamoDB і скидаємо  
 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()

5. Підсумок

Використовуючи batch_writer(), ми значно покращуємо ефективність запису великих обсягів даних у DynamoDB. Ось короткий огляд основних кроків:

Створити таблицю DynamoDB, якщо її ще немає.
Згенерувати випадкові дані для тестування.
Записувати пакетами до 25 записів за раз.

Цей скрипт допоможе вам автоматизувати процес запису великих наборів даних у DynamoDB і зробить ваш додаток більш ефективним.

import boto3  
import random  
import string  
from datetime import datetime  
from botocore.exceptions import ClientError  

# Ініціалізація сесії з AWS  
dynamodb = boto3.resource('dynamodb', region_name='us-east-1') # Вказуємо регіон  

# Вказуємо назву таблиці DynamoDB  
table_name = 'My_DynamoDB_Table_Name'  

# Перевіряємо, чи існує таблиця, і якщо її немає, створюємо  
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' # Тип String  
 },  
 ],  
 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  

# Функція для генерації випадкових рядків  
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), # Випадкове значення  
 }  

# Функція для запису даних пакетами  
def batch_write(table, records):  
 with table.batch_writer() as batch:  
 for record in records:  
 batch.put_item(Item=record)  

def main():  
 # Створення таблиці, якщо вона не існує  
 table = create_table_if_not_exists()  

 records_batch = []  
 for i in range(1, 1001): # Цикл для створення 1000 записів  
 record = generate_record()  
 records_batch.append(record)  

 # Якщо розмір пакету досягне 25 елементів, записуємо в DynamoDB і скидаємо  
 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()

Підсумок

Обробка великих обсягів даних у DynamoDB може бути складною, але використання правильних методів — таких як перевірка існування таблиці, динамічна генерація даних і запис пакетами — дозволяє зробити цей процес безшовним і ефективним. Ви можете змінити скрипт відповідно до вашого конкретного випадку та дослідити інші можливості DynamoDB, такі як глобальні вторинні індекси або автоскалування для ще більш оптимізованої роботи.

Перекладено з: Efficient Batch Writing to DynamoDB with Python: A Step-by-Step Guide