Verwenden von AWS Lambda mit Amazon DynamoDB - AWS Lambda

Verwenden von AWS Lambda mit Amazon DynamoDB

Sie können eine AWS Lambda-Funktion verwenden, um Datensätze in einem Amazon-DynamoDB-Stream zu verarbeiten. Mit DynamoDB Streams können Sie eine Lambda-Funktion auslösen, damit jedes Mal, wenn eine DynamoDB-Tabelle aktualisiert wird, weitere Aufgaben ausgeführt werden.

Lambda liest Datensätze aus dem Stream und ruft Ihre Funktion synchron mit einem Ereignis auf, das Stream-Datensätze enthält. Lambda liest Datensätze in Batches und ruft Ihre Funktion auf, um Datensätze aus dem Batch zu verarbeiten.

Beispiel Datensatzereignis von DynamoDB Streams
{ "Records": [ { "eventID": "1", "eventVersion": "1.0", "dynamodb": { "Keys": { "Id": { "N": "101" } }, "NewImage": { "Message": { "S": "New item!" }, "Id": { "N": "101" } }, "StreamViewType": "NEW_AND_OLD_IMAGES", "SequenceNumber": "111", "SizeBytes": 26 }, "awsRegion": "us-west-2", "eventName": "INSERT", "eventSourceARN": "arn:aws:dynamodb:us-west-2:111122223333:table/TestTable/stream/2015-05-11T21:21:33.291", "eventSource": "aws:dynamodb" }, { "eventID": "2", "eventVersion": "1.0", "dynamodb": { "OldImage": { "Message": { "S": "New item!" }, "Id": { "N": "101" } }, "SequenceNumber": "222", "Keys": { "Id": { "N": "101" } }, "SizeBytes": 59, "NewImage": { "Message": { "S": "This item has changed" }, "Id": { "N": "101" } }, "StreamViewType": "NEW_AND_OLD_IMAGES" }, "awsRegion": "us-west-2", "eventName": "MODIFY", "eventSourceARN": "arn:aws:dynamodb:us-west-2:111122223333:table/TestTable/stream/2015-05-11T21:21:33.291", "eventSource": "aws:dynamodb" } ]}

Lambda fragt Shards in Ihrem DynamoDB-Stream nach Datensätzen bei einer Basisrate von viermal pro Sekunde ab. Sind Datensätze verfügbar, ruft Lambda Ihre Funktion auf und wartet auf das Ergebnis. Ist die Verarbeitung erfolgreich, setzt Lambda die Abrufe fort, bis es weitere Datensätze erhält.

Standardmäßig ruft Lambda Ihre Funktion auf, sobald Datensätze verfügbar sind. Wenn der Batch, den Lambda aus der Ereignisquelle liest, nur einen Datensatz enthält, sendet Lambda nur einen Datensatz an die Funktion. Damit die Funktion nicht mit einer kleinen Anzahl von Datensätzen aufgerufen wird, können Sie die Ereignisquelle anweisen, Datensätze bis zu 5 Minuten lang zu puffern, indem Sie ein Batch-Fenster konfigurieren. Bevor die Funktion aufgerufen wird, liest Lambda so lange Datensätze aus der Ereignisquelle, bis es einen vollständigen Batch erfasst hat, das Batch-Verarbeitungsfenster abläuft oder der Batch die Nutzlastgrenze von 6 MB erreicht. Weitere Informationen finden Sie unter Batching-Verhalten.

Wenn Ihre Funktion einen Fehler zurückgibt, wiederholt Lambda den Vorgang mit dem Batch, bis die Verarbeitung erfolgreich ist oder die Daten ablaufen. Um unterbrochenen Shards zu vermeiden, können Sie die Ereignisquellenzuordnung so konfigurieren, dass es mit einer kleineren Batchgröße wiederholt wird, die Anzahl der Wiederholungen beschränkt wird oder Datensätze verworfen werden, die zu alt sind. Um verworfene Ereignisse beizubehalten, können Sie die Ereignisquellenzuordnung so konfigurieren, dass Details zu fehlgeschlagenen Batches an eine SQS-Warteschlange oder ein SNS-Thema gesendet werden.

Sie können die Parallelität auch erhöhen, indem Sie mehrere Batches aus jedem Shard parallel verarbeiten. Lambda kann bis zu 10 Batches in jedem Shard gleichzeitig verarbeiten. Wenn Sie die Anzahl gleichzeitiger Batches pro Shard erhöhen, stellt Lambda weiterhin die Auftragsverarbeitung auf Shard-Ebene sicher.

Konfigurieren Sie die ParallelizationFactor-Einstellung, um einen Shard eines Kinesis- oder DynamoDB-Datenstroms mit mehr als einem Lambda-Aufruf gleichzeitig zu verarbeiten. Sie können die Anzahl der gleichzeitigen Batches angeben, die Lambda von einem Shard über einen Parallelisierungsfaktor von 1 (Standard) bis 10 abfragt. Wenn ParallelizationFactor beispielsweise auf 2 gesetzt ist, können Sie maximal 200 gleichzeitige Lambda-Aufrufe haben, um 100 Kinesis-Daten-Shards zu verarbeiten. Dies hilft, den Verarbeitungsdurchsatz hochzuskalieren, wenn das Datenvolumen flüchtig ist und IteratorAge hoch ist. Beachten Sie, dass der Parallelisierungsfaktor nicht funktioniert, wenn Sie die Kinesis-Aggregation verwenden. Weitere Informationen finden Sie unter Neue AWS Lambda-Skalierungssteuerelemente für Kinesis- und DynamoDB-Ereignisquellen. Vollständige Tutorials finden Sie auch im Workshop Serverless-Datenverarbeitung auf AWS.

Berechtigungen für die Ausführungsrolle

Lambda benötigt die folgenden Berechtigungen zum Verwalten von Ressourcen, die zu Ihrem DynamoDB-Stream gehören. Fügen Sie sie der Ausführungsrolle Ihrer Funktion hinzu.

Die verwaltete AWSLambdaDynamoDBExecutionRole-Richtlinie enthält die folgenden Berechtigungen. Weitere Informationen finden Sie unter Lambda-Ausführungsrolle.

Um Datensätze fehlgeschlagener Batches an eine SQS-Warteschlange oder ein SNS-Thema zu senden, benötigt Ihre Funktion zusätzliche Berechtigungen. Jeder Zielservice benötigt eine andere Berechtigung wie folgt:

Konfigurieren eines Streams als Ereignisquelle

Erstellen Sie eine Ereignisquellenzuordnung, um Lambda anzuweisen, Datensätze aus Ihrem Stream an eine Lambda-Funktion zu senden. Sie können mehrere Ereignisquellen-Zuweisung erstellen, um gleiche Daten mit mehreren Lambda-Funktionen oder Elemente aus mehreren Streams mit nur einer Funktion zu verarbeiten.

Um Ihre Funktion für das Lesen von DynamoDB Streams in der Lambda-Konsole zu konfigurieren, erstellen Sie einen DynamoDB-Auslöser.

So erstellen Sie einen Auslöser
  1. Öffnen Sie die Seite Funktionen der Lambda-Konsole.

  2. Wählen Sie den Namen einer Funktion aus.

  3. Wählen Sie unter Function overview (Funktionsübersicht) die Option Add trigger (Trigger hinzufügen).

  4. Wählen Sie einen Auslösertyp aus.

  5. Konfigurieren Sie die erforderlichen Optionen und wählen Sie dann Add (Hinzufügen) aus.

Lambda unterstützt die folgenden Optionen für DynamoDB-Ereignisquellen.

Optionen für die Ereignisquelle
  • DynamoDB-Tabelle – Die DynamoDB-Tabelle, aus der Datensätze gelesen werden sollen.

  • Batchgröße – Die Anzahl der Datensätze, die in jedem Batch bis zu 10.000 an die Funktion gesendet werden sollen. Lambda übergibt alle Datensätze im Batch in einem einzigen Aufruf an die Funktion, solange die Gesamtgröße der Ereignisse nicht das Nutzlast-Limit für synchrone Aufrufe überschreitet (6 MB).

  • Batchfenster – Geben Sie die maximale Zeitspanne zur Erfassung von Datensätzen vor dem Aufruf der Funktion in Sekunden an.

  • Startposition – Verarbeiten Sie nur neue Datensätze oder alle vorhandenen Datensätze.

    • Neueste – Verarbeiten Sie Datensätze, die neu zum Stream hinzugefügt wurden.

    • Horizont trimmen – Verarbeiten Sie alle Datensätze im Stream.

    Nach der Verarbeitung aller vorhandenen Datensätze hat die Funktion aufgeholt und setzt die Verarbeitung neuer Datensätze fort.

  • Ziel bei Ausfall – Eine SQS-Warteschlange oder ein SNS-Thema für Datensätze, die nicht verarbeitet werden können. Wenn Lambda einen Datensatz-Batch verwirft, weil er zu alt ist oder alle Wiederholungen erschöpft hat, sendet es Details zum Batch an die Warteschlange oder das Thema.

  • Wiederholungsversuche – Die maximale Anzahl von Wiederholungen von Lambda, wenn die Funktion einen Fehler zurückgibt. Dies gilt nicht für Servicefehler oder Drosselungen, bei denen der Batch die Funktion nicht erreicht hat.

  • Höchstalter des Datensatzes – Das maximale Alter eines Datensatzes, den Lambda an Ihre Funktion sendet.

  • Batch bei Fehler aufteilen – Wenn die Funktion einen Fehler zurückgibt, teilen Sie den Batch vor dem erneuten Versuch in zwei Teile. Ihre ursprüngliche Einstellung für die Batch-Größe bleibt unverändert.

  • Gleichzeitige Batches pro Shard – Verarbeitet gleichzeitig mehrere Batches aus demselben Shard.

  • Aktiviert – Auf „true“ festlegen, um die Ereignisquellenzuordnung zu aktivieren. Auf "false" festlegen, um die Verarbeitung von Datensätzen zu beenden. Lambda merkt sich den zuletzt verarbeiteten Datensatz und setzt die Verarbeitung nach erneuter Aktivierung der Zuordnung an dieser Stelle fort.

Anmerkung

Ihnen werden keine GetRecords-API-Aufrufe berechnet, die von Lambda als Teil von DynamoDB-Auslösern aufgerufen werden.

Um die Konfiguration der Ereignisquelle zu einem späteren Zeitpunkt zu verwalten, wählen Sie den Auslöser im Designer aus.

APIs für die Ereignisquellenzuordnung

Um eine Ereignisquelle mit der AWS Command Line Interface (AWS CLI) oder einem AWS-SDK zu verwalten, können Sie die folgenden API-Operationen verwenden:

Im folgenden Beispiel wird mithilfe der AWS CLI eine Funktion namens my-function einem DynamoDB-Stream zugeordnet, der durch seinen Amazon-Ressourcennamen (ARN) angegeben wird – mit einer Batchgröße von 500.

aws lambda create-event-source-mapping --function-name my-function --batch-size 500 --maximum-batching-window-in-seconds 5 --starting-position LATEST \ --event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2019-06-10T19:26:16.525

Die Ausgabe sollte folgendermaßen aussehen:

{ "UUID": "14e0db71-5d35-4eb5-b481-8945cf9d10c2", "BatchSize": 500, "MaximumBatchingWindowInSeconds": 5, "ParallelizationFactor": 1, "EventSourceArn": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2019-06-10T19:26:16.525", "FunctionArn": "arn:aws:lambda:us-east-2:123456789012:function:my-function", "LastModified": 1560209851.963, "LastProcessingResult": "No records processed", "State": "Creating", "StateTransitionReason": "User action", "DestinationConfig": {}, "MaximumRecordAgeInSeconds": 604800, "BisectBatchOnFunctionError": false, "MaximumRetryAttempts": 10000 }

Konfigurieren Sie zusätzliche Optionen, um die Verarbeitung von Batches anzupassen und um anzugeben, wann nicht verarbeitete Datensätze verworfen werden sollen. Im folgenden Beispiel wird eine Ereignisquellenzuordnung aktualisiert, um nach zwei Wiederholungsversuchen einen Fehlerdatensatz an eine SQS-Warteschlange zu senden oder wenn die Datensätze älter als eine Stunde sind.

aws lambda update-event-source-mapping --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \ --maximum-retry-attempts 2 --maximum-record-age-in-seconds 3600 --destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:us-east-2:123456789012:dlq"}}'

Sie sollten diese Ausgabe sehen:

{ "UUID": "f89f8514-cdd9-4602-9e1f-01a5b77d449b", "BatchSize": 100, "MaximumBatchingWindowInSeconds": 0, "ParallelizationFactor": 1, "EventSourceArn": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2019-06-10T19:26:16.525", "FunctionArn": "arn:aws:lambda:us-east-2:123456789012:function:my-function", "LastModified": 1573243620.0, "LastProcessingResult": "PROBLEM: Function call failed", "State": "Updating", "StateTransitionReason": "User action", "DestinationConfig": {}, "MaximumRecordAgeInSeconds": 604800, "BisectBatchOnFunctionError": false, "MaximumRetryAttempts": 10000 }

Aktualisierte Einstellungen werden asynchron angewendet und werden erst nach Abschluss des Vorgangs in der Ausgabe berücksichtigt. Verwenden Sie den get-event-source-mapping-Befehl, um den aktuellen Status anzuzeigen.

aws lambda get-event-source-mapping --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b

Sie sollten diese Ausgabe sehen:

{ "UUID": "f89f8514-cdd9-4602-9e1f-01a5b77d449b", "BatchSize": 100, "MaximumBatchingWindowInSeconds": 0, "ParallelizationFactor": 1, "EventSourceArn": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2019-06-10T19:26:16.525", "FunctionArn": "arn:aws:lambda:us-east-2:123456789012:function:my-function", "LastModified": 1573244760.0, "LastProcessingResult": "PROBLEM: Function call failed", "State": "Enabled", "StateTransitionReason": "User action", "DestinationConfig": { "OnFailure": { "Destination": "arn:aws:sqs:us-east-2:123456789012:dlq" } }, "MaximumRecordAgeInSeconds": 3600, "BisectBatchOnFunctionError": false, "MaximumRetryAttempts": 2 }

Um mehrere Batches gleichzeitig zu verarbeiten, verwenden Sie die --parallelization-factor-Option.

aws lambda update-event-source-mapping --uuid 2b733gdc-8ac3-cdf5-af3a-1827b3b11284 \ --parallelization-factor 5

Fehlerbehandlung

Das Ereignisquellen-Zuweisung, das Datensätze aus Ihrem DynamoDB Stream liest, ruft Ihre Funktion synchron auf und versucht bei Fehlern erneut. Wenn Lambda die Funktion drosselt oder einen Fehler zurückgibt, ohne die Funktion aufzurufen, versucht es Lambda erneut, bis die Datensätze ablaufen oder das maximale Alter überschreiten, das Sie in der Ereignisquellenzuordnung konfigurieren.

Wenn die Funktion die Datensätze empfängt, aber einen Fehler zurückgibt, versucht Lambda so lange, bis die Datensätze im Batch ablaufen, das maximale Alter überschreiten oder das konfigurierte Wiederholungskontingent erreicht haben. Bei Funktionsfehlern können Sie auch die Ereignisquellenzuordnung so konfigurieren, dass ein fehlgeschlagener Batch in zwei Batches aufgeteilt wird. Bei einem erneuten Versuch mit kleineren Batches werden fehlerhafte Datensätze isoliert und Zeitüberschreitungsprobleme umgangen. Das Teilen eines Batches wird nicht auf das Wiederholungskontingent angerechnet.

Wenn die Fehlerbehandlungsmaßnahmen fehlschlagen, verwirft Lambda die Datensätze und setzt die Verarbeitung von Batches aus dem Stream fort. Bei den Standardeinstellungen bedeutet dies, dass ein fehlerhafter Datensatz die Verarbeitung auf dem betroffenen Shard für bis zu einen Tag blockieren kann. Um dies zu vermeiden, konfigurieren Sie die Ereignisquellenzuordnung Ihrer Funktion mit einer angemessenen Anzahl von Wiederholungen und einem maximalen Datensatzalter, das zu Ihrem Anwendungsfall passt.

Um eine Aufzeichnung verworfener Batches beizubehalten, konfigurieren Sie ein Ziel für fehlgeschlagene Ereignisse. Lambda sendet ein Dokument mit Details zum Batch an die Zielwarteschlange oder das Zielthema.

So konfigurieren Sie ein Ziel für Datensätze mit fehlerhaften Ereignissen
  1. Öffnen Sie die Seite Funktionen der Lambda-Konsole.

  2. Wählen Sie eine Funktion aus.

  3. Wählen Sie unter Function overview (Funktionsübersicht) die Option Add destination (Ziel hinzufügen).

  4. Wählen Sie in Source (Quelle) die Option Stream invocation (Stream-Aufruf) aus.

  5. Wählen Sie in Stream (Stream) einen der Funktion zugeordneten Stream aus.

  6. Wählen Sie unter Destination type (Zieltyp) den Ressourcentyp aus, der den Aufrufdatensatz empfängt.

  7. Wählen Sie unter Destination (Ziel) eine Ressource aus.

  8. Wählen Sie Save aus.

Das folgende Beispiel zeigt einen Aufrufdatensatz für einen DynamoDB-Stream.

Beispiel Aufrufdatensatz
{ "requestContext": { "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81", "functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction", "condition": "RetryAttemptsExhausted", "approximateInvokeCount": 1 }, "responseContext": { "statusCode": 200, "executedVersion": "$LATEST", "functionError": "Unhandled" }, "version": "1.0", "timestamp": "2019-11-14T00:13:49.717Z", "DDBStreamBatchInfo": { "shardId": "shardId-00000001573689847184-864758bb", "startSequenceNumber": "800000000003126276362", "endSequenceNumber": "800000000003126276362", "approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z", "approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z", "batchSize": 1, "streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388" } }

Sie können diese Informationen verwenden, um die betroffenen Datensätze aus dem Stream für die Fehlersuche abzurufen. Die tatsächlichen Datensätze sind nicht enthalten, daher müssen Sie diesen Datensatz verarbeiten und aus dem Stream abrufen, bevor sie ablaufen und verloren gehen.

Amazon CloudWatch-Metriken

Lambda gibt die IteratorAge-Metrik aus, wenn Ihre Funktion die Verarbeitung eines Batches von Datensätzen fertigstellt. Die Metrik gibt an, wie alt der letzte Datensatz im Batch bei Fertigstellung der Verarbeitung war. Wenn Ihre Funktion neue Ereignisse verarbeitet, können Sie mit dem Iterator-Alter die Latenz zwischen dem Zeitpunkt, zu dem ein Datensatz hinzugefügt wird und dem Zeitpunkt, zu dem die Funktion verarbeitet wird, schätzen.

Eine steigende Tendenz beim Iterator-Alter kann auf Probleme mit Ihrer Funktion hindeuten. Weitere Informationen finden Sie unter Arbeiten mit Lambda-Funktionsmetriken.

Zeitfenster

Lambda-Funktionen können kontinuierliche Stream-Verarbeitungsanwendungen ausführen. Ein Stream entspricht einer unbegrenzten Menge von Daten, die kontinuierlich durch Ihre Anwendung fließen. Um Informationen aus dieser sich ständig aktualisierenden Eingabe zu analysieren, können Sie die enthaltenen Datensätze mithilfe eines zeitlich definierten Fensters binden.

Rollierende Fenster sind unterschiedliche Zeitfenster, die sich in regelmäßigen Abständen öffnen und schließen. Standardmäßig sind Lambda-Aufrufe zustandslos – Sie können sie nicht für die Verarbeitung von Daten über mehrere kontinuierliche Aufrufe hinweg ohne eine externe Datenbank verwenden. Mit rollierenden Fenstern können Sie jedoch Ihren Status über Aufrufe hinweg beibehalten. Dieser Zustand enthält das Gesamtergebnis der Nachrichten, die zuvor für das aktuelle Fenster verarbeitet wurden. Ihr Zustand kann maximal 1 MB pro Shard betragen. Wenn er diese Größe überschreitet, wird Lambda das Fenster vorzeitig beenden.

Jeder Datensatz in einem Stream gehört zu einem bestimmten Fenster. Lambda verarbeitet jeden Datensatz mindestens einmal, garantiert jedoch nicht, dass jeder Datensatz nur einmal verarbeitet wird. In seltenen Fällen, etwa bei der Fehlerbehandlung, werden einige Datensätze möglicherweise mehrmals verarbeitet. Datensätze werden beim ersten Mal immer in der richtigen Reihenfolge verarbeitet. Wenn Datensätze mehr als einmal verarbeitet werden, werden sie nicht in der richtigen Reihenfolge verarbeitet.

Aggregation und Verarbeitung

Ihre benutzerverwaltete Funktion wird sowohl zur Aggregation als auch zur Verarbeitung der Endergebnisse dieser Aggregation aufgerufen. Lambda aggregiert alle im Fenster empfangenen Datensätze. Sie können diese Datensätze in mehreren Stapeln erhalten, jeweils als ein separater Aufruf. Jeder Aufruf erhält einen Zustand. Wenn Sie also rollierende Fenster verwenden, muss Ihre Lambda-Funktionsantwort eine state-Eigenschaft enthalten. Wenn die Antwort keine state-Eigenschaft enthält, betrachtet Lambda dies als fehlgeschlagenen Aufruf. Um diese Bedingung zu erfüllen, kann Ihre Funktion ein TimeWindowEventResponse-Objekt zurückgeben, das die folgende JSON-Form aufweist:

Beispiel TimeWindowEventResponse-Werte
{ "state": { "1": 282, "2": 715 }, "batchItemFailures": [] }
Anmerkung

Für Java-Funktionen empfehlen wir, eine Map<String, String> zu verwenden, um den Status darzustellen.

Am Ende des Fensters wird das Flag isFinalInvokeForWindow auf true gesetzt, um anzugeben, dass es sich um den Endzustand handelt und dass es für die Verarbeitung bereit ist. Nach der Verarbeitung werden das Fenster und Ihr endgültiger Aufruf wird abgeschlossen, und dann wird der Zustand gelöscht.

Am Ende Ihres Fensters verwendet Lambda die endgültige Verarbeitung für Aktionen an den Aggregationsergebnissen. Ihre endgültige Verarbeitung wird synchron aufgerufen. Nach erfolgreichem Aufruf zeigt Ihre Funktion auf die Sequenznummer und die Stream-Verarbeitung wird fortgesetzt. Wenn der Aufruf nicht erfolgreich ist, unterbricht Ihre Lambda-Funktion die weitere Verarbeitung bis zu einem erfolgreichen Aufruf.

Beispiel DynamodbTimeWindowEvent
{ "Records":[ { "eventID":"1", "eventName":"INSERT", "eventVersion":"1.0", "eventSource":"aws:dynamodb", "awsRegion":"us-east-1", "dynamodb":{ "Keys":{ "Id":{ "N":"101" } }, "NewImage":{ "Message":{ "S":"New item!" }, "Id":{ "N":"101" } }, "SequenceNumber":"111", "SizeBytes":26, "StreamViewType":"NEW_AND_OLD_IMAGES" }, "eventSourceARN":"stream-ARN" }, { "eventID":"2", "eventName":"MODIFY", "eventVersion":"1.0", "eventSource":"aws:dynamodb", "awsRegion":"us-east-1", "dynamodb":{ "Keys":{ "Id":{ "N":"101" } }, "NewImage":{ "Message":{ "S":"This item has changed" }, "Id":{ "N":"101" } }, "OldImage":{ "Message":{ "S":"New item!" }, "Id":{ "N":"101" } }, "SequenceNumber":"222", "SizeBytes":59, "StreamViewType":"NEW_AND_OLD_IMAGES" }, "eventSourceARN":"stream-ARN" }, { "eventID":"3", "eventName":"REMOVE", "eventVersion":"1.0", "eventSource":"aws:dynamodb", "awsRegion":"us-east-1", "dynamodb":{ "Keys":{ "Id":{ "N":"101" } }, "OldImage":{ "Message":{ "S":"This item has changed" }, "Id":{ "N":"101" } }, "SequenceNumber":"333", "SizeBytes":38, "StreamViewType":"NEW_AND_OLD_IMAGES" }, "eventSourceARN":"stream-ARN" } ], "window": { "start": "2020-07-30T17:00:00Z", "end": "2020-07-30T17:05:00Z" }, "state": { "1": "state1" }, "shardId": "shard123456789", "eventSourceARN": "stream-ARN", "isFinalInvokeForWindow": false, "isWindowTerminatedEarly": false }

Konfiguration

Sie können rollierende Fenster konfigurieren, wenn Sie eine Ereignisquellenzuordnung erstellen oder aktualisieren. Um ein rollierendes Fenster zu konfigurieren, geben Sie das Fenster in Sekunden an. Der folgende AWS Command Line Interface (AWS CLI) -Beispielbefehl erstellt eine Streaming-Ereignisquellenzuordnung, die ein rollierendes Fenster von 120 Sekunden hat. Die für Aggregation und Verarbeitung definierte Lambda-Funktion wird tumbling-window-example-function genannt.

aws lambda create-event-source-mapping --event-source-arn arn:aws:dynamodb:us-east-1:123456789012:stream/lambda-stream --function-name "arn:aws:lambda:us-east-1:123456789018:function:tumbling-window-example-function" --region us-east-1 --starting-position TRIM_HORIZON --tumbling-window-in-seconds 120

Lambda bestimmt die rollierenden Fenstergrenzen basierend auf dem Zeitpunkt, zu dem Datensätze in den Stream eingefügt wurden. Für alle Datensätze steht ein ungefährer Zeitstempel zur Verfügung, den Lambda in Grenzbestimmungen verwendet.

Rollierende Fensteraggregationen unterstützen kein Resharding. Wenn der Shard endet, betrachtet Lambda das Fenster als geschlossen und die untergeordneten Shards beginnen ihr eigenes Fenster in einem neuen Zustand.

Rollierende Fenster unterstützen vollständig die bestehenden Wiederholungsrichtlinien maxRetryAttempts und maxRecordAge.

Beispiel Handler.py – Aggregation und Verarbeitung

Die folgende Python-Funktion veranschaulicht, wie Sie Ihren Endzustand aggregieren und dann verarbeiten:

def lambda_handler(event, context): print('Incoming event: ', event) print('Incoming state: ', event['state']) #Check if this is the end of the window to either aggregate or process. if event['isFinalInvokeForWindow']: # logic to handle final state of the window print('Destination invoke') else: print('Aggregate invoke') #Check for early terminations if event['isWindowTerminatedEarly']: print('Window terminated early') #Aggregation logic state = event['state'] for record in event['Records']: state[record['dynamodb']['NewImage']['Id']] = state.get(record['dynamodb']['NewImage']['Id'], 0) + 1 print('Returning state: ', state) return {'state': state}

Melden von Batch-Elementen

Beim Konsumieren und Verarbeiten von Streaming-Daten aus einer Ereignisquelle werden standardmäßig Lambda-Checkpoints auf die höchste Sequenznummer eines Batches nur dann überprüft, wenn der Batch ein voller Erfolg ist. Lambda behandelt alle anderen Ergebnisse als einen vollständigen Fehler und versucht, den Batch bis zum Wiederholungslimit zu verarbeiten. Um beim Verarbeiten von Stapeln aus einem Stream Teilerfolge zu ermöglichen, aktivieren Sie ReportBatchItemFailures. Das Zulassen von Teilerfolgen kann dazu beitragen, die Anzahl der Wiederholungen in einer Aufzeichnung zu reduzieren, obwohl die Möglichkeit von Wiederholungen in einer erfolgreichen Aufzeichnung nicht vollständig verhindert wird.

Zum Aktivieren von ReportBatchItemFailures fügen Sie den Enum-Wert ReportBatchItemFailures der FunctionResponseTypes-Liste hinzu. Diese Liste zeigt an, welche Antworttypen für Ihre Funktion aktiviert sind. Sie können diese Liste konfigurieren, wenn Sie eine Ereignisquellenzuordnung erstellen oder aktualisieren.

Berichtsyntax

Beim Konfigurieren von Berichten zu Batch-Elementfehlern wird die StreamsEventResponse-Klasse mit einer Liste von Batch-Elementfehlern zurückgegeben. Sie können ein StreamsEventResponse-Objekt verwenden, um die Sequenznummer des ersten fehlgeschlagenen Datensatzes im Batch zurückzugeben. Sie können auch Ihre eigene benutzerdefinierte Klasse mit der richtigen Antwortsyntax erstellen. Die folgende JSON-Struktur zeigt die erforderliche Antwortsyntax:

{ "batchItemFailures": [ { "itemIdentifier": "<id>" } ] }
Anmerkung

Wenn das batchItemFailures-Array mehrere Elemente enthält, verwendet Lambda den Datensatz mit der niedrigsten Sequenznummer als Kontrollpunkt. Lambda wiederholt dann alle Datensätze ab diesem Kontrollpunkt.

Erfolgs- und Misserfolgsbedingungen

Lambda behandelt einen Batch als vollständigen Erfolg, wenn Sie eines der folgenden Elemente zurückgeben:

  • Eine leere batchItemFailure-Liste

  • Eine ungültige batchItemFailure-Liste

  • Ein leeres EventResponse

  • Ein ungültiges EventResponse

Lambda behandelt einen Batch als vollständigen Misserfolg, wenn Sie eines der folgenden Elemente zurückgeben:

  • Eine leere Zeichenfolge itemIdentifier

  • Ein ungültiges itemIdentifier

  • Ein itemIdentifier mit einem falschen Schlüsselnamen

Lambda wiederholt Fehler basierend auf Ihrer Wiederholungsstrategie.

Einen Batch halbieren

Wenn Ihr Aufruf fehlschlägt und BisectBatchOnFunctionError eingeschaltet ist, wird der Stapel unabhängig von Ihrer ReportBatchItemFailures-Einstellung halbiert.

Wenn eine partielle Batch-Erfolgsantwort empfangen wird und sowohl BisectBatchOnFunctionError als auch ReportBatchItemFailures aktiviert sind, wird der Batch mit der zurückgegebenen Sequenznummer halbiert und Lambda versucht nur die verbleibenden Datensätze erneut.

Java
Beispiel Handler.java – gibt neue StreamsEventResponse() zurück
import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import com.amazonaws.services.lambda.runtime.events.DynamodbEvent; import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse; import com.amazonaws.services.lambda.runtime.events.models.dynamodb.StreamRecord; import java.io.Serializable; import java.util.ArrayList; import java.util.List; public class ProcessDynamodbRecords implements RequestHandler<DynamodbEvent, Serializable> { @Override public StreamsEventResponse handleRequest(DynamodbEvent input, Context context) { List<StreamsEventResponse.BatchItemFailure> batchItemFailures = new ArrayList<>(); String curRecordSequenceNumber = ""; for (DynamodbEvent.DynamodbStreamRecord dynamodbStreamRecord : input.getRecords()) { try { //Process your record StreamRecord dynamodbRecord = dynamodbStreamRecord.getDynamodb(); curRecordSequenceNumber = dynamodbRecord.getSequenceNumber(); } catch (Exception e) { /* Since we are working with streams, we can return the failed item immediately. Lambda will immediately begin to retry processing from this failed item onwards. */ batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber)); return new StreamsEventResponse(batchItemFailures); } } return new StreamsEventResponse(); } }
Python
Beispiel Handler.py – gibt batchItemFailures[] zurück
def handler(event, context): records = event.get("Records") curRecordSequenceNumber = ""; for record in records: try: # Process your record curRecordSequenceNumber = record["dynamodb"]["SequenceNumber"] except Exception as e: # Return failed record's sequence number return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]} return {"batchItemFailures":[]}

Konfigurationsparameter zu Amazon DynamoDB Streams

Alle Lambda-Ereignisquellentypen verwenden dieselben API-Operationen CreateEventSourceMapping und UpdateEventSourceMapping. Allerdings gelten nur einige der Parameter für DynamoDB Streams.

Ereignisquellparameter, die für DynamoDB Streams gelten
Parameter Erforderlich Standard Hinweise

BatchSize

N

100

Höchstwert: 10 000.

BisectBatchOnFunctionError

N

false

DestinationConfig

N

Ein Ziel der Amazon-SQS-Warteschlange oder des Amazon-SNS-Themas für verworfene Datensätze

Enabled

N

true

EventSourceArn

Y

Der ARN des Datenstroms oder eines Stream-Konsumenten

FunctionName

Y

MaximumBatchingWindowInSeconds

N

0

MaximumRecordAgeInSeconds

N

-1

-1 bedeutet unendlich: Fehlgeschlagene Datensätze werden wiederholt, bis der Datensatz abläuft

Minimum: -1

Maximum: 604 800

MaximumRetryAttempts

N

-1

-1 bedeutet unendlich: Fehlgeschlagene Datensätze werden wiederholt, bis der Datensatz abläuft

Minimum: -1

Maximum: 604 800

ParallelizationFactor

N

1

Maximum: 10

StartingPosition

Y

TRIM_HORIZON oder LATEST

TumblingWindowInSeconds

N

Minimum: 0

Maximum: 900