So verarbeitet Lambda Datensätze aus Amazon Kinesis Data Streams

Sie können eine Lambda-Funktion verwenden, um Datensätze in einem Amazon Kinesis Kinesis-Datenstream zu verarbeiten. Sie können eine Lambda-Funktion einem Kinesis Data Streams Streams-Verbraucher mit gemeinsamem Durchsatz (Standard-Iterator) oder einem dedizierten Durchsatzverbraucher mit erweitertem Fan-Out zuordnen. Bei Standard-Iteratoren fragt Lambda jeden Shard in Ihrem Kinesis-Stream mithilfe des Protokolls nach Datensätzen ab. HTTP Die Ereignisquellenzuordnung teilt den Lesedurchsatz mit anderen Konsumenten des Shards zusammen.

Weitere Informationen zu Kinesis-Datenströmen finden Sie unter Daten aus Amazon Kinesis Data Streams.

Anmerkung

Kinesis berechnet Gebühren für jeden Shard, sowie bei verbessertem Rundsenden für Daten, die aus dem Stream gelesen werden. Details zu den Preisen finden Sie unter Amazon-Kinesis- Preise.

Themen

• Abfragen und Stapeln von Streams
• Beispielereignis
• Amazon Kinesis Data Streams Streams-Datensätze mit Lambda verarbeiten
• Konfiguration einer partiellen Batch-Antwort mit Kinesis Data Streams und Lambda
• Verworfene Batch-Datensätze für eine Kinesis Data Streams Streams-Ereignisquelle in Lambda aufbewahren
• Implementierung der statusbehafteten Kinesis Data Streams Streams-Verarbeitung in Lambda
• Lambda-Parameter für Amazon Kinesis Data Streams Streams-Ereignisquellenzuordnungen
• Verwenden der Ereignisfilterung mit einer Kinesis-Ereignisquelle
• Tutorial: Lambda mit Kinesis Data Streams verwenden

Abfragen und Stapeln von Streams

Lambda liest Datensätze aus dem Datenstrom 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. Jeder Batch enthält Datensätze aus einem einzelnen Shard/Datenstrom.

Bei standardmäßigen Kinesis-Datenströmen fragt Lambda Shards in Ihrem Stream mit einer Geschwindigkeit von einmal pro Sekunde für jeden Shard nach Datensätzen ab. Für Kinesis Enhanced Fan-Out verwendet Lambda eine HTTP /2-Verbindung, um auf Datensätze zu warten, die von Kinesis übertragen werden. Sind Datensätze verfügbar, ruft Lambda Ihre Funktion auf und wartet auf das Ergebnis.

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.

Warnung

Lambda-Ereignisquellenzuordnungen verarbeiten jedes Ereignis mindestens einmal, und es kann zu einer doppelten Verarbeitung von Datensätzen kommen. Um mögliche Probleme im Zusammenhang mit doppelten Ereignissen zu vermeiden, empfehlen wir Ihnen dringend, Ihren Funktionscode idempotent zu machen. Weitere Informationen finden Sie im Knowledge Center unter Wie mache ich meine Lambda-Funktion idempotent? AWS

Lambda wartet nicht, bis konfigurierte Erweiterungen abgeschlossen sind, bevor der nächste Stapel zur Verarbeitung gesendet wird. Anders ausgedrückt: Ihre Erweiterungen werden möglicherweise weiter ausgeführt, während Lambda den nächsten Stapel von Datensätzen verarbeitet. Dies kann zu Drosselungsproblemen führen, wenn Sie gegen eine Einstellung oder gegen einen Grenzwert im Zusammenhang mit der Parallelität Ihres Kontos verstoßen. Um zu erkennen, ob möglicherweise ein Problem vorliegt, müssen Sie Ihre Funktionen überwachen sowie überprüfen, ob für Ihre Zuordnung von Ereignisquellen unerwartet hohe Parallelitätsmetriken vorliegen. Aufgrund der kurzen Zeit zwischen den Aufrufen kann Lambda kurzzeitig eine höhere Gleichzeitigkeitsnutzung als die Anzahl der Shards melden. Dies kann sogar für Lambda-Funktionen ohne Erweiterungen gelten.

Konfigurieren Sie die ParallelizationFactorEinstellung so, dass ein Shard eines Kinesis-Datenstroms mit mehr als einem Lambda-Aufruf gleichzeitig verarbeitet wird. Sie können die Anzahl der gleichzeitigen Batches angeben, die Lambda von einem Shard über einen Parallelisierungsfaktor von 1 (Standard) bis 10 abfragt. Wenn Sie beispielsweise ParallelizationFactor auf 2 setzen, können Sie maximal 200 gleichzeitige Lambda-Aufrufe haben, um 100 Kinesis-Datensplitter zu verarbeiten (obwohl Sie in der Praxis möglicherweise unterschiedliche Werte für die Metrik sehen). ConcurrentExecutions Dies hilft, den Verarbeitungsdurchsatz hochzuskalieren, wenn das Datenvolumen flüchtig ist und IteratorAge hoch ist. Wenn Sie die Anzahl der gleichzeitigen Batches pro Shard erhöhen, sorgt Lambda trotzdem für die Verarbeitung in der Reihenfolge auf Partitionsschlüsselebene.

Sie können es auch ParallelizationFactor mit Kinesis-Aggregation verwenden. Das Verhalten der Ereignisquellenzuordnung hängt davon ab, ob Sie das erweiterte Fan-Out verwenden:

• Ohne erweitertes Fan-Out: Alle Ereignisse innerhalb eines aggregierten Ereignisses müssen denselben Partitionsschlüssel haben. Der Partitionsschlüssel muss außerdem mit dem des aggregierten Ereignisses übereinstimmen. Wenn die Ereignisse innerhalb des aggregierten Ereignisses unterschiedliche Partitionsschlüssel haben, kann Lambda nicht garantieren, dass die Ereignisse in der richtigen Reihenfolge nach Partitionsschlüsseln verarbeitet werden.

• Mit verbessertem Fan-Out: Zunächst dekodiert Lambda das aggregierte Ereignis in seine einzelnen Ereignisse. Das aggregierte Ereignis kann einen anderen Partitionsschlüssel haben als die darin enthaltenen Ereignisse. Ereignisse, die nicht dem Partitionsschlüssel entsprechen, werden jedoch gelöscht und gehen verloren. Lambda verarbeitet diese Ereignisse nicht und sendet sie nicht an ein konfiguriertes Fehlerziel.

Beispielereignis

{
    "Records": [
        {
            "kinesis": {
                "kinesisSchemaVersion": "1.0",
                "partitionKey": "1",
                "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
                "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
                "approximateArrivalTimestamp": 1545084650.987
            },
            "eventSource": "aws:kinesis",
            "eventVersion": "1.0",
            "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
            "eventName": "aws:kinesis:record",
            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
            "awsRegion": "us-east-2",
            "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
        },
        {
            "kinesis": {
                "kinesisSchemaVersion": "1.0",
                "partitionKey": "1",
                "sequenceNumber": "49590338271490256608559692540925702759324208523137515618",
                "data": "VGhpcyBpcyBvbmx5IGEgdGVzdC4=",
                "approximateArrivalTimestamp": 1545084711.166
            },
            "eventSource": "aws:kinesis",
            "eventVersion": "1.0",
            "eventID": "shardId-000000000006:49590338271490256608559692540925702759324208523137515618",
            "eventName": "aws:kinesis:record",
            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
            "awsRegion": "us-east-2",
            "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
        }
    ]
}