Bedingte Massenaktualisierung

DynamoDB unterstützt Massenoperationen wie BatchWriteItem, mit denen Sie bis zu 25 PutItem- und DeleteItem-Anforderungen in einem einzigen Batch ausführen können. BatchWriteItem unterstützt jedoch keine UpdateItem-Operationen und Bedingungsausdrücke. Um das Problem zu umgehen, können Sie andere DynamoDB verwenden, APIs z. B. TransactWriteItems für Batchgrößen von bis zu 100.

Wenn mehr Elemente betroffen sind und ein großer Teil der Daten geändert werden muss, können Sie Dienste wie Amazon EMR AWS Step Functions oder benutzerdefinierte Skripts und Tools wie DynamoDB-Shell für effiziente Massenaktualisierungen verwenden. AWS Glue

Wann dieses Muster verwendet werden sollte

DynamoDB-Shell ist kein Anwendungsfall, der für die Produktion unterstützt wird.
TransactWriteItems – bis zu 100 einzelne Updates mit oder ohne Bedingungen, die als ACID nach dem „alles oder nichts“-Prinzip gebündelt ausgeführt werden. TransactWriteItems-Aufrufe können auch mit einem ClientRequestToken versehen werden, falls Ihre Anwendung Idempotenz benötigt, was bedeutet, dass mehrere identische Aufrufe dieselbe Wirkung haben wie ein einziger Aufruf. Dadurch wird sichergestellt, dass Sie dieselbe Transaktion nicht mehrmals ausführen und am Ende einen falschen Datenstatus erhalten.

Kompromiss — Es wird zusätzlicher Durchsatz verbraucht. 2 WCUs pro 1 KB Schreibvorgang anstelle der standardmäßigen 1 WGU pro 1 KB Schreibvorgang.
PartiQL BatchExecuteStatement – bis zu 25 Updates mit oder ohne Bedingungen. BatchExecuteStatement gibt immer eine erfolgreiche Antwort auf die gesamte Anforderung sowie eine Liste der Antworten auf einzelne Operationen zurück, wobei die Reihenfolge gewahrt bleibt.

Kompromiss – Bei größeren Batches ist zusätzliche Client-seitige Logik erforderlich, um Anforderungen in Batches von 25 zu verteilen. Bei der Festlegung der Wiederholungsstrategie müssen individuelle Fehlerantworten berücksichtigt werden.

Codebeispiele

Diese Codebeispiele verwenden die boto3-Bibliothek, das AWS SDK für Python. In diesen Beispielen wird vorausgesetzt, dass Sie boto3 installiert und mit den entsprechenden AWS -Anmeldeinformationen konfiguriert haben.

Gehen wir von einer Inventardatenbank für einen Elektrogerätehändler aus, der über mehrere Lager in europäischen Städten verfügt. Da es Ende des Sommers ist, möchte der Verkäufer die Tischventilatoren ausräumen, um Platz für andere Lagerbestände zu schaffen. Der Verkäufer möchte einen Preisnachlass für alle Tischventilatoren gewähren, die aus Lagern in Italien geliefert werden, aber nur, wenn ein Vorratsbestand von 20 Tischventilatoren vorhanden ist. Die DynamoDB-Tabelle heißt inventory. Sie hat ein Schlüsselschema mit Partitionsschlüssel sku, einer eindeutigen Kennung für jedes Produkt, und einen Sortierschlüssel warehouse, der eine Kennung für ein Lager ist.

Der folgende Python-Code zeigt, wie diese bedingte Massenaktualisierung mithilfe eines BatchExecuteStatement-API-Aufrufs durchgeführt wird.


import boto3

client=boto3.client("dynamodb")

before_image=client.query(TableName='inventory', KeyConditionExpression='sku=:pk_val AND begins_with(warehouse, :sk_val)', ExpressionAttributeValues={':pk_val':{'S':'F123'},':sk_val':{'S':'WIT'}}, ProjectionExpression='sku,warehouse,quantity,price')
print("Before update: ", before_image['Items'])

response=client.batch_execute_statement(
        Statements=[
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITTUR1'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITROM1'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITROM2'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITROM5'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITVEN1'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITVEN2'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
            {'Statement': 'UPDATE inventory SET price=price-5 WHERE sku=? AND warehouse=? AND quantity > 20', 'Parameters': [{'S':'F123'}, {'S':'WITVEN3'}], 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'},
        ],
        ReturnConsumedCapacity='TOTAL'
    )

after_image=client.query(TableName='inventory', KeyConditionExpression='sku=:pk_val AND begins_with(warehouse, :sk_val)', ExpressionAttributeValues={':pk_val':{'S':'F123'},':sk_val':{'S':'WIT'}}, ProjectionExpression='sku,warehouse,quantity,price')
print("After update: ", after_image['Items'])

Die Ausführung erzeugt die folgende Ausgabe für Beispieldaten:


Before update:  [{'quantity': {'N': '20'}, 'warehouse': {'S': 'WITROM1'}, 'price': {'N': '40'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '25'}, 'warehouse': {'S': 'WITROM2'}, 'price': {'N': '40'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '28'}, 'warehouse': {'S': 'WITROM5'}, 'price': {'N': '38'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '26'}, 'warehouse': {'S': 'WITTUR1'}, 'price': {'N': '40'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '10'}, 'warehouse': {'S': 'WITVEN1'}, 'price': {'N': '38'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '20'}, 'warehouse': {'S': 'WITVEN2'}, 'price': {'N': '38'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '50'}, 'warehouse': {'S': 'WITVEN3'}, 'price': {'N': '35'}, 'sku': {'S': 'F123'}}]
After update:  [{'quantity': {'N': '20'}, 'warehouse': {'S': 'WITROM1'}, 'price': {'N': '40'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '25'}, 'warehouse': {'S': 'WITROM2'}, 'price': {'N': '35'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '28'}, 'warehouse': {'S': 'WITROM5'}, 'price': {'N': '33'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '26'}, 'warehouse': {'S': 'WITTUR1'}, 'price': {'N': '35'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '10'}, 'warehouse': {'S': 'WITVEN1'}, 'price': {'N': '38'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '20'}, 'warehouse': {'S': 'WITVEN2'}, 'price': {'N': '38'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '50'}, 'warehouse': {'S': 'WITVEN3'}, 'price': {'N': '30'}, 'sku': {'S': 'F123'}}]

Da es sich um eine begrenzte Operation für ein internes System handelt, wurden die Anforderungen an die Idempotenz nicht berücksichtigt. Es ist möglich, einen zusätzlichen Integritätsschutz festzulegen, z. B. dass Preisaktualisierungen nur dann durchgeführt werden sollten, wenn der Preis höher als 35 und weniger als 40 ist, damit die Aktualisierungen robuster sind.

Alternativ können wir dieselbe Massenaktualisierung auch mit TransactWriteItems durchführen, falls strengere Anforderungen an Idempotenz und ACID gelten. Es ist jedoch wichtig, daran zu denken, dass entweder alle Operationen im Transaktionspaket ausgeführt werden oder das gesamte Paket fehlschlägt.

Nehmen wir an, dann aufgrund einer Hitzewelle in Italien die Nachfrage nach Tischventilatoren stark gestiegen ist. Der Verkäufer möchte seine Kosten für Schreibtischventilatoren beim Versand aus jedem Lager in Italien um 20 Euro erhöhen, aber die Aufsichtsbehörde erlaubt diese Kostenerhöhung nur, wenn die aktuellen Kosten für den gesamten Lagerbestand unter 70 Euro liegen. Es ist wichtig, dass der Preis im gesamten Inventar nur ein einziges Mal aktualisiert wird und nur dann, wenn die Kosten in jedem Lager weniger als 70 Euro betragen.

Der folgende Python-Code zeigt, wie diese Massenaktualisierung mithilfe eines TransactWriteItems-API-Aufrufs durchgeführt wird.



import boto3

client=boto3.client("dynamodb")

before_image=client.query(TableName='inventory', KeyConditionExpression='sku=:pk_val AND begins_with(warehouse, :sk_val)', ExpressionAttributeValues={':pk_val':{'S':'F123'},':sk_val':{'S':'WIT'}}, ProjectionExpression='sku,warehouse,quantity,price')
print("Before update: ", before_image['Items'])

response=client.transact_write_items(
        ClientRequestToken='UUIDAWS124',
        TransactItems=[
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITTUR1'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITROM1'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITROM2'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITROM5'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITVEN1'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITVEN2'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
            {'Update': { 'Key': {'sku': {'S':'F123'}, 'warehouse': {'S':'WITVEN3'}}, 'UpdateExpression': 'SET price = price + :inc', 'ConditionExpression': 'price < :cap', 'ExpressionAttributeValues': { ':inc': {'N': '20'}, ':cap': {'N': '70'}}, 'TableName': 'inventory', 'ReturnValuesOnConditionCheckFailure': 'ALL_OLD'}},
        ],
        ReturnConsumedCapacity='TOTAL'
    )

after_image=client.query(TableName='inventory', KeyConditionExpression='sku=:pk_val AND begins_with(warehouse, :sk_val)', ExpressionAttributeValues={':pk_val':{'S':'F123'},':sk_val':{'S':'WIT'}}, ProjectionExpression='sku,warehouse,quantity,price')
print("After update: ", after_image['Items'])

Die Ausführung erzeugt die folgende Ausgabe für Beispieldaten:



Before update:  [{'quantity': {'N': '20'}, 'warehouse': {'S': 'WITROM1'}, 'price': {'N': '60'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '25'}, 'warehouse': {'S': 'WITROM2'}, 'price': {'N': '55'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '28'}, 'warehouse': {'S': 'WITROM5'}, 'price': {'N': '53'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '26'}, 'warehouse': {'S': 'WITTUR1'}, 'price': {'N': '55'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '10'}, 'warehouse': {'S': 'WITVEN1'}, 'price': {'N': '58'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '20'}, 'warehouse': {'S': 'WITVEN2'}, 'price': {'N': '58'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '50'}, 'warehouse': {'S': 'WITVEN3'}, 'price': {'N': '50'}, 'sku': {'S': 'F123'}}]
After update:  [{'quantity': {'N': '20'}, 'warehouse': {'S': 'WITROM1'}, 'price': {'N': '80'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '25'}, 'warehouse': {'S': 'WITROM2'}, 'price': {'N': '75'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '28'}, 'warehouse': {'S': 'WITROM5'}, 'price': {'N': '73'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '26'}, 'warehouse': {'S': 'WITTUR1'}, 'price': {'N': '75'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '10'}, 'warehouse': {'S': 'WITVEN1'}, 'price': {'N': '78'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '20'}, 'warehouse': {'S': 'WITVEN2'}, 'price': {'N': '78'}, 'sku': {'S': 'F123'}}, {'quantity': {'N': '50'}, 'warehouse': {'S': 'WITVEN3'}, 'price': {'N': '70'}, 'sku': {'S': 'F123'}}]

Es gibt mehrere Ansätze für die Durchführung von Massenaktualisierungen in DynamoDB. Der geeignete Ansatz hängt von Faktoren wie den Anforderungen an die and/or ACID-Idempotenz, der Anzahl der zu aktualisierenden Elemente und der Vertrautheit mit ihnen ab. APIs

Warnung JavaScript ist in Ihrem Browser nicht verfügbar oder deaktiviert.

Zur Nutzung der AWS-Dokumentation muss JavaScript aktiviert sein. Weitere Informationen finden auf den Hilfe-Seiten Ihres Browsers.

Dokumentkonventionen

Massendatenoperationen

Effiziente Massenoperationen