Verarbeitung von Amazon MSK-Nachrichten mit Lambda

Hinzufügen von Amazon MSK als Ereignisquelle

Um ein Ereignisquellen-Mapping zu erstellen, fügen Sie Amazon MSK als Lambda-Funktionsauslöser mithilfe der Lambda-Konsole, eines AWS-SDKs oder AWS Command Line Interface (AWS CLI) hinzu. Beachten Sie, dass Lambda beim Hinzufügen von Amazon MSK als Auslöser die VPC-Einstellungen des Amazon-MSK-Clusters annimmt, nicht die VPC-Einstellungen der Lambda-Funktion.

In diesem Abschnitt wird beschrieben, wie Sie mit der Lambda-Konsole und AWS CLI ein Ereignisquellen-Zuweisung erstellen.

Voraussetzungen

Anpassbare Konsumentengruppen-ID

Wenn Sie Kafka als Ereignisquelle einrichten, können Sie eine Konsumentengruppen-ID angeben. Diese Konsumentengruppen-ID ist eine vorhandene Kennung für die Kafka-Konsumentengruppe, der Ihre Lambda-Funktion beitreten soll. Mit diesem Feature können Sie alle laufenden Kafka-Datensatzverarbeitungs-Setups nahtlos von anderen Verbrauchern auf Lambda migrieren.

Wenn Sie eine Konsumentengruppen-ID angeben und sie innerhalb dieser Konsumentengruppe weitere aktive Poller gibt, verteilt Kafka Nachrichten an alle Konsumenten. Mit anderen Worten, Lambda erhält nicht alle Nachrichten zum Thema Kafka. Wenn Sie möchten, dass Lambda alle Nachrichten im Thema verarbeitet, deaktivieren Sie alle anderen Poller in dieser Konsumentengruppe.

Wenn Sie außerdem eine Konsumentengruppen-ID angeben und Kafka eine gültige vorhandene Konsumentengruppe mit derselben ID findet, ignoriert Lambda die StartingPosition-Parameter für Ihre Ereignisquellen-Zuweisung. Stattdessen beginnt Lambda mit der Verarbeitung von Datensätzen gemäß dem zugesagten Versatz der Konsumentengruppe. Wenn Sie eine Konsumentengruppen-ID angeben und Kafka keine vorhandene Konsumentengruppe finden kann, konfiguriert Lambda Ihre Ereignisquelle mit der angegebenen StartingPosition.

Die Konsumentengruppen-ID, die Sie angeben, muss unter all Ihren Kafka-Ereignisquellen eindeutig sein. Nachdem Sie eine Kafka-Ereignisquellenzuordnung mit der angegebenen Konsumentengruppen-ID erstellt haben, können Sie diesen Wert nicht aktualisieren.

Hinzufügen eines Amazon MSK-Auslösers (Konsole)

Befolgen Sie diese Schritte, um Ihren Amazon-MSK-Cluster und ein Kafka-Thema als Auslöser für Ihre Lambda-Funktion hinzuzufügen.

Hinzufügen eines Amazon-MSK-Auslösers (AWS CLI)

Verwenden Sie die folgenden AWS CLI-Beispielbefehle, um einen Amazon-MSK-Auslöser für Ihre Lambda-Funktion zu erstellen und anzuzeigen.

Erstellen eines Auslösers mit dem AWS CLI

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'

Weitere Informationen finden Sie in der CreateEventSourceMapping-API-Referenzdokumentation.

Anzeigen des Zustands mithilfe des AWS CLI

Das folgende Beispiel verwendet get-event-source-mapping AWS CLI, um den Status der Zuordnung von Ereignisquellen zu beschreiben, die Sie erstellt haben.

aws lambda get-event-source-mapping \
  --uuid 6d9bce8e-836b-442c-8070-74e77903c815

Amazon-MSK-Konfigurationsparameter

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

Erstellen von kontenübergreifenden Zuordnungen von Ereignisquellen

Sie können private Multi-VPC-Konnektivität verwenden, um eine Lambda-Funktion mit einem bereitgestellten MSK-Cluster in einem anderen AWS-Konto zu verbinden. Multi-VPC-Konnektivität verwendet AWS PrivateLink, wodurch der gesamte Datenverkehr im AWS-Netzwerk bleibt.

Um eine kontenübergreifende Zuordnung von Ereignisquellen zu erstellen, müssen Sie zunächst die Multi-VPC-Konnektivität für den MSK-Cluster konfigurieren. Verwenden Sie beim Erstellen der Zuordnung von Ereignisquellen den ARN der verwalteten VPC-Verbindung anstelle des Cluster-ARNs, wie in den folgenden Beispielen gezeigt. Die Operation CreateEventSourceMapping unterscheidet sich auch je nachdem, welchen Authentifizierungstyp der MSK-Cluster verwendet.

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function \
  --source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function \
  --source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'

Verwenden eines Amazon MSK-Clusters als Ereignisquelle

Wenn Sie Ihren Apache Kafka-Cluster oder Amazon MSK als Auslöser für Ihre Lambda-Funktion hinzufügen, wird der Cluster als Ereignisquelle verwendet.

Lambda liest Ereignisdaten aus den Kafka-Themen, die Sie als Topics in einer CreateEventSourceMapping-Anforderung angeben, basierend auf der von Ihnen angegebenen StartingPosition. Nach erfolgreicher Verarbeitung wird Ihr Kafka-Thema Ihrem Kafka-Cluster zugeordnet.

Wenn Sie die StartingPosition als LATEST angeben, beginnt Lambda mit dem Lesen der neuesten Nachricht in jeder zum Thema gehörenden Partition. Da es nach der Auslöserkonfiguration eine gewisse Verzögerung geben kann, bevor Lambda mit dem Lesen der Nachrichten beginnt, liest Lambda keine Nachrichten, die innerhalb dieses Zeitraums erzeugt wurden.

Lambda liest Nachrichten sequentiell für jede Kafka-Themenpartition. Eine einzelne Lambda-Nutzlast kann Nachrichten von mehreren Partitionen enthalten. Wenn mehr Datensätze verfügbar sind, verarbeitet Lambda Datensätze weiter in Batches, basierend auf dem BatchSize-Wert, den Sie in einer CreateEventSourceMapping-Anforderung angeben, bis Ihre Funktion auf dem gleichen Stand wie das Thema ist.

Nachdem Lambda jeden Batch verarbeitet hat, werden die Offsets der Nachrichten in diesem Batch festgeschrieben. Wenn Ihre Funktion einen Fehler für eine der Nachrichten in einem Batch zurückgibt, wiederholt Lambda den gesamten Nachrichtenbatch, bis die Verarbeitung erfolgreich ist oder die Nachrichten ablaufen. Sie können Datensätze, bei denen alle Wiederholungsversuche fehlschlagen, zur späteren Verarbeitung an ein Ausfallziel senden.

Startpositionen für Abfragen und Streams

Beachten Sie, dass die Stream-Abfrage bei der Erstellung und Aktualisierung der Zuordnung von Ereignisquellen letztendlich konsistent ist.

Dieses Verhalten bedeutet, dass, wenn Sie LATEST als Startposition für den Stream angeben, die Zuordnung von Ereignisquellen bei der Erstellung oder Aktualisierung möglicherweise Ereignisse übersieht. Um sicherzustellen, dass keine Ereignisse übersehen werden, geben Sie die Startposition des Streams als TRIM_HORIZON oder AT_TIMESTAMP an.

Amazon-CloudWatch-Metriken

Lambda gibt die Metrik OffsetLag aus, während Ihre Funktion Datensätze verarbeitet. Der Wert dieser Metrik ist die Differenz (der Versatz) zwischen dem letzten Datensatz, der ins Kafka-Ereignisquellen-Thema geschrieben wurde, und dem letzten Datensatz, den die Konsumentengruppe Ihrer Funktion verarbeitet hat. Sie können mit OffsetLag die Latenz zwischen dem Hinzufügen eines Datensatzes und der Verarbeitung des Datensatzes durch Ihre Konsumentengruppe abschätzen.

Ein zunehmender Trend in OffsetLag kann auf Probleme mit Pollern in der Konsumentengruppe Ihrer Funktion hinweisen. Weitere Informationen finden Sie unter Verwendung von CloudWatch-Metriken mit Lambda.

Skalierungsverhalten des Nachrichtendurchsatzes für Amazon MSK-Zuordnungen von Ereignisquellen

Sie können zwischen zwei Modi der Skalierung des Nachrichtendurchsatzes für Ihre Amazon MSK-Zuordnung von Ereignisquellen wählen:

Standardmodus (auf Abruf)

Wenn Sie anfänglich eine Amazon MSK-Ereignisquelle erstellen, weist Lambda eine Standardanzahl von Ereignis-Pollern zu, um alle Partitionen im Kafka-Thema zu verarbeiten. Lambda erhöht oder verringert automatisch die Anzahl der Ereignis-Poller je nach Nachrichtenlast.

In Intervallen von einer Minute wertet Lambda den Offset-Lag aller Partitionen des Themas aus. Wenn die Offset-Verzögerung zu hoch ist, empfängt die Partition Nachrichten schneller als Lambda sie verarbeiten kann. Falls erforderlich, fügt Lambda dem Thema Ereignis-Poller hinzu oder entfernt sie. Dieser Prozess der automatischen Skalierung durch Hinzufügen oder Entfernen von Ereignis-Pollern erfolgt innerhalb von drei Minuten nach der Auswertung.

Wenn Ihre Ziel-Lambda-Funktion gedrosselt ist, reduziert Lambda die Anzahl der Ereignis-Poller. Diese Aktion verringert die Arbeitslast der Funktion, indem sie die Anzahl der Nachrichten reduziert, die Ereignis-Poller abrufen und an die Funktion senden können.

Konfigurieren des bereitgestellten Modus

Für Workloads, bei denen Sie den Durchsatz Ihrer Zuordnung von Ereignisquellen optimieren müssen, können Sie den Bereitstellungsmodus verwenden. Im Bereitstellungsmodus definieren Sie Mindest- und Höchstgrenzen für die Anzahl der bereitgestellten Ereignis-Poller. Diese bereitgestellten Event-Poller sind auf Ihre Zuordnung von Ereignisquellen ausgerichtet und können unerwartete Nachrichtenspitzen durch reaktionsschnelle Autoskalierung bewältigen. Wir empfehlen die Verwendung des Bereitstellungsmodus für Kafka-Workloads, die strenge Leistungsanforderungen haben.

In Lambda ist ein Event-Poller eine Recheneinheit, die einen Durchsatz von bis zu 5 MBps verarbeiten kann. Nehmen wir als Referenz an, dass Ihre Ereignisquelle eine durchschnittliche Nutzlast von 1 MB erzeugt und die durchschnittliche Funktionsdauer 1 Sekunde beträgt. Wenn die Nutzdaten keiner Transformation (wie Filterung) unterzogen wird, kann ein einzelner Poller einen Durchsatz von 5 MB/s und 5 gleichzeitige Lambda-Aufrufe unterstützen. Die Verwendung des Bereitstellungsmodus verursacht zusätzliche Kosten. Kostenvoranschläge finden Sie unter AWS Lambda-Preise.

Im Bereitstellungsmodus liegt der Bereich der akzeptierten Werte für die Mindestanzahl der Ereignis-Poller (MinimumPollers) zwischen 1 und 200 (einschließlich). Der Bereich der akzeptierten Werte für die maximale Anzahl von Event-Pollern (MaximumPollers) liegt zwischen 1 und einschließlich 2.000. MaximumPollers muss größer als oder gleich MinimumPollers sein. Um eine geordnete Verarbeitung innerhalb der Partitionen zu gewährleisten, begrenzt Lambda das MaximumPollers auf die Anzahl der Partitionen im Thema.

Weitere Einzelheiten zur Auswahl geeigneter Werte für minimale und maximale Ereignis-Poller finden Sie unter Bewährte Verfahren und Überlegungen bei der Verwendung des Bereitstellungsmodus.

Sie können den Bereitstellungsmodus für Ihre Amazon MSK-Zuordnung von Ereignisquellen über die Konsole oder die Lambda-API konfigurieren.

Sie können den Bereitstellungsmodus programmatisch konfigurieren, indem Sie das ProvisionedPollerConfig-Objekt in Ihrer EventSourceMappingConfiguration verwenden. Der folgende CLI-Befehl UpdateEventSourceMapping konfiguriert beispielsweise einen MinimumPollers-Wert von 5 und einen MaximumPollers-Wert von 100.

aws lambda update-event-source-mapping \
    --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
    --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}'

Nachdem Sie den Bereitstellungsmodus konfiguriert haben, können Sie die Verwendung von Event-Pollern für Ihren Workload beobachten, indem Sie die ProvisionedPollers-Metrik überwachen. Weitere Informationen finden Sie unter Metriken zur Zuordnung von Ereignisquellen.

Um den Bereitstellungsmodus zu deaktivieren und zum Standardmodus (on-demand) zurückzukehren, können Sie den folgenden CLI-Befehl UpdateEventSourceMapping verwenden:

aws lambda update-event-source-mapping \
    --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
    --provisioned-poller-config '{}'

Bewährte Verfahren und Überlegungen bei der Verwendung des Bereitstellungsmodus

Die optimale Konfiguration der minimalen und maximalen Ereignispoller für Ihre Zuordnung von Ereignisquellen hängt von den Leistungsanforderungen Ihrer Anwendung ab. Wir empfehlen Ihnen, mit den standardmäßigen Minimum Event Pollers zu beginnen, um das Leistungsprofil zu erstellen. Passen Sie Ihre Konfiguration auf der Grundlage der beobachteten Nachrichtenverarbeitungsmuster und Ihres gewünschten Leistungsprofils an.

Erhöhen Sie bei Workloads mit starkem Datenverkehr und strengen Leistungsanforderungen die Mindestanzahl der Event-Poller, um plötzliche Nachrichtenfluten zu bewältigen. Um den Mindestbedarf an Event-Pollern zu ermitteln, berücksichtigen Sie die Nachrichten pro Sekunde und die durchschnittliche Größe der Nutzdaten Ihres Workloads und verwenden Sie die Durchsatzkapazität eines einzelnen Event-Pollers (bis zu 5 MBps) als Referenz.

Um eine geordnete Verarbeitung innerhalb einer Partition aufrechtzuerhalten, begrenzt Lambda die maximale Anzahl der Event-Poller auf die Anzahl der Partitionen im Thema. Darüber hinaus hängt die maximale Anzahl von Event-Pollern, auf die Ihr Zuordnung von Ereignisquellen skaliert werden kann, von den Gleichzeitigkeitseinstellungen der Funktion ab.

Wenn Sie den Bereitstellungsmodus aktivieren, aktualisieren Sie Ihre Netzwerkeinstellungen, um AWS PrivateLink VPC-Endpunkte und zugehörige Berechtigungen zu entfernen.