Abfragen von Kafka-Themen im Warteschlangenmodus

Der Warteschlangenmodus ist ein Konsummodus, der Share-Gruppen verwendet. Dieser Modus wurde in Apache Kafka 4.0 (für den Kafka Queue Poller ist Version 4.2 oder höher erforderlich). Im Warteschlangenmodus können Sie Nachrichten zu Kafka-Themen mithilfe von Share-Gruppen ungeordnet und mit hohem Durchsatz verarbeiten. Im Gegensatz zu herkömmlichen Kafka-Benutzergruppen, die Partitionen ausschließlich Verbrauchern zuweisen, ermöglichen Freigabegruppen mehreren Verbrauchern das gleichzeitige Lesen derselben Partitionen. Dies ähnelt der Verarbeitung von Nachrichten aus einer Warteschlange durch Amazon SQS.

Wenn Sie eine Ereignisquellenzuordnung für eine selbstverwaltete Apache Kafka Cluster, Sie können zwischen zwei Nutzungsmodi wählen:

• Streamen (Standard). Verwendet Kafka-Benutzergruppen mit einer Reihenfolge pro Partition. Lambda weist jede Partition einem Verbraucher zu.

• Warteschlange. Verwendet Share-Gruppen, die jetzt verfügbar sind in Apache Kafka 4.0, für ungeordnete Verarbeitung mit hohem Durchsatz (für den Kafka Queue Poller ist Version 4.2 oder höher erforderlich). Bei Share Groups können mehrere Verbraucher gleichzeitig von denselben Partitionen lesen, ähnlich wie Amazon SQS Nachrichten aus einer Warteschlange verarbeitet.

Wenn Sie keinen Nutzungsmodus angeben, verwendet die Zuordnung der Ereignisquelle Stream. Diese Seite behandelt den Queue-Nutzungsmodus. Allgemeine Informationen zur Verwendung von Lambda mit Selbstverwaltung Apache Kafka, siehe Verwendung von Lambda mit selbstverwaltetem Apache Kafka.

Im Warteschlangenmodus verwendet Lambda den Kafka Share Consumer, um Nachrichten aus dem Thema abzufragen. Jeder Event-Poller verarbeitet Nachrichten über alle Partitionen hinweg ohne Reihenfolge auf Partitionsebene. Dadurch wird die Parallelität für jeden Poller erhöht.

Mindestanforderungen

Um den Queue-Nutzungsmodus verwenden zu können, muss Ihr Kafka-Cluster die folgenden Anforderungen erfüllen:

• Apache Kafka Version 4.2 oder höher mit aktivierten Share-Gruppen. Kafka aktiviert share.version standardmäßig, und Sie können es verwalten über kafka-features.sh.

• Self-managed Apache Kafka Nur (SMK) -Cluster. Von Amazon verwaltetes Streaming für Apache Kafka (Amazon MSK) unterstützt derzeit den Warteschlangenmodus nicht. AWS fügt MSK-Unterstützung hinzu, sobald MSK Kafka 4.2 oder höher mit aktivierten Share-Gruppen anbietet.

Vergleich von Stream und Queue

Feature	Stream (Standard)	Warteschlange
Bestellung	Per-partition Bestellung garantiert	Keine Bestellgarantie
Modell für Verbraucher	Verbrauchergruppe (exklusive Partitionszuweisung)	Gruppe teilen (gleichzeitiger Partitionszugriff)
Durchsatz pro Event-Poller	Bis zu 5 MB/s	Bis zu 5 MB/s
Gleichzeitige Aufrufe pro Event-Poller	Bis zu 5	Bis zu 10
Modi skalieren	On-demand und bereitgestellt	On-demand und bereitgestellt
Gemeinsame Nutzung von EPUs	Kann EPUs mit anderen Kafka Streams-Zuordnungen von Eventquellen teilen	Isoliert. Quellzuordnungen für Warteschlangenereignisse werden nur mit anderen Quellzuordnungen für Warteschlangenereignisse gemeinsam genutzt
Anwendungsfall	Geordnete Verarbeitung von Ereignissen	High-throughput ungeordnete Arbeitslasten

Warteschlangenzuordnung erstellen

Um eine Ereignisquellenzuordnung im Warteschlangenmodus zu erstellen, legen Sie den ConsumptionMode Feld von SelfManagedKafkaEventSourceConfig zu Queue. Ersetzen Sie in den folgenden Beispielen my-function, die Broker-Endpunkte und my-queue-topic mit deinen eigenen Werten.

AWS CLI

Folgendes AWS Der CLI-Befehl erstellt eine Zuordnung der Ereignisquelle im Warteschlangenmodus. Ersetzen my-function mit Ihrem Funktionsnamen, den Broker-Endpunkten bei Ihren Brokern und my-queue-topic mit deinem Thema.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 1, "MaximumPollers": 10}'

AWS CloudFormation

Folgendes AWS CloudFormation Die Vorlage erstellt eine Zuordnung der Ereignisquelle im Warteschlangenmodus. Ersetzen Sie die Broker-Endpunkte und my-queue-topic mit deinen eigenen Werten.

MyKafkaQueueESM:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    FunctionName: !Ref MyFunction
    SelfManagedEventSource:
      Endpoints:
        KafkaBootstrapServers:
          - broker1.us-west-2.example.com:9092
          - broker2.us-west-2.example.com:9092
    Topics:
      - my-queue-topic
    BatchSize: 100
    SelfManagedKafkaEventSourceConfig:
      ConsumptionMode: Queue
    ProvisionedPollerConfig:
      MinimumPollers: 1
      MaximumPollers: 10

Konfigurationsparameter

In der folgenden Tabelle sind die Parameter aufgeführt, die für die Zuordnungen von Ereignisquellen im Warteschlangenmodus gelten.

Parameter	Description	Bereich/Standard
BatchSize	Maximale Anzahl von Datensätzen pro Stapel, die an die Funktion gesendet werden	1—10.000. Standard: 100
Enabled	Ob die Zuordnung der Ereignisquelle aktiv ist	true oder false. Standard: true
FilterCriteria	Muster zur Ereignisfilterung, um nur übereinstimmende Datensätze zu verarbeiten	Bis zu 5 Filtermuster
MaximumRetryAttempts	Maximale Anzahl von Wiederholungsversuchen für fehlgeschlagene Datensätze	-1—10.000. Standard: 0 (im Warteschlangenmodus führt der Kafka-Broker Wiederholungsversuche durch, wenn kein Ziel für den Fall eines Fehlers konfiguriert ist)
BisectBatchOnFunctionError	Bei einem Funktionsfehler den Stapel in zwei Hälften teilen	true oder false. Standard: false
ProvisionedPollerConfig.MinimumPollers	Mindestanzahl an Event-Pollern (bereitgestellter Modus)	1—200. Standard: 1
ProvisionedPollerConfig.MaximumPollers	Maximale Anzahl von Event-Pollern (bereitgestellter Modus)	1—2.000. Standard: 200
ProvisionedPollerConfig.PollerGroupName	Gruppenname für die gemeinsame Nutzung von EPU (Bereitstellungsmodus, nur Quelle-Zuordnungen für Warteschlangenereignisse)	Zeichenfolge. Bis zu 100 Zuordnungen von Ereignisquellen pro Gruppe
SelfManagedKafkaEventSourceConfig.ConsumptionMode	Nutzungsmodus für die Ereignisquelle	Muss sein Queue für den Warteschlangenmodus
SchemaRegistryConfig	Schemaregistrierung für die Ereignisvalidierung	Siehe Verwenden von Schemaregistern mit Kafka-Ereignisquellen in Lambda

Wichtig

Das MaximumBatchingWindowInSeconds Parameter ist wird nicht unterstützt für Zuordnungen von Ereignisquellen im Warteschlangenmodus. Der Kafka Share Consumer verwendet eine ausdrückliche Bestätigung. Alle Datensätze in einem Stapel müssen explizit bestätigt werden, bevor der Verbraucher neue Datensätze abrufen kann. Dies ist mit einem zeitbasierten Batching-Fenster nicht kompatibel. Wenn Sie angeben MaximumBatchingWindowInSeconds Beim Erstellen einer Ereignisquellenzuordnung im Warteschlangenmodus gibt die API einen Validierungsfehler zurück.

Unterstützte Funktionen zur Zuordnung von Ereignisquellen

In der folgenden Tabelle sind die Funktionen zur Zuordnung von Lambda-Ereignisquellen aufgeführt, die im Warteschlangenmodus verfügbar sind.

Funktionsmerkmal	Unterstützt	Hinweise
Modus bereitgestellter Kapazität	Ja	Konfigurieren MinimumPollers und MaximumPollers
On-demand Modus	Ja	Standard-Skalierungsmodus. Lambda skaliert Poller automatisch auf bis zu 5 MB/s oder 50 gleichzeitige Aufrufe, je nachdem, welcher Wert zuerst erreicht wird. Keine zusätzliche Konfiguration erforderlich.
Ereignisfilterung	Ja	Bis zu 5 Filtermuster pro Zuordnung der Ereignisquelle
Schema Registry	Ja	Avro und Protobuf mit Glue Schema Registry oder Confluent Schema Registry. Die Filterung der Schemaregistrierung ist nur im bereitgestellten Modus verfügbar.
Low-latency Verarbeitung	Ja	Im bereitgestellten Modus verfügbar
Versuchen Sie es erneut mit den Konfigurationen	Ja	MaximumRetryAttempts, BisectBatchOnFunctionError
On-failure Ziel (Kafka-Thema)	Ja	Fehlgeschlagene Datensätze an ein Kafka-Thema senden
On-failure Ziel (SNS/SQS)	Ja	Fehlgeschlagene Datensätze an SNS oder SQS senden
Protokollierung der Zuordnung der Ereignisquelle	Ja	CloudWatch Protokolle für die Diagnose der Ereignisquellenzuordnung
Gruppierung der Zuordnung von Ereignisquellen (PollerGroupName)	Ja	Zuordnungen von Ereignisquellen im Warteschlangenmodus können EPUs nur mit anderen Ereignisquellenzuordnungen im Warteschlangenmodus gemeinsam nutzen (bereitgestellter Modus)
Fenster zum Stapeln	Nein	Wird aufgrund des Verhaltens im Share Consumer-Blockiermodus nicht unterstützt
Startposition	Nein	StartingPosition und StartingPositionTimestamp werden nicht unterstützt. Der Warteschlangenmodus liest immer aus der Commit-Position der Share-Gruppe.

Skalierungsmodi

Die Zuordnungen der Ereignisquellen im Warteschlangenmodus unterstützen sowohl den bedarfsgesteuerten (Standard) als auch den bereitgestellten Skalierungsmodus.

On-demand Modus (Standard)

Im On-Demand-Modus verwaltet Lambda automatisch Event-Poller für die Zuordnung Ihrer Ereignisquellen im Warteschlangenmodus.

On-demand Der Modus skaliert auf bis zu 5 Event-Poller pro Eventquellen-Mapping. Er erreicht seine Obergrenze entweder bei 5 MB/s Gesamtdurchsatz oder 50 gleichzeitige Aufrufe (5 Poller mal 10). Dies geschieht unabhängig davon, welches Limit zuerst erreicht wird. Verwenden Sie den bereitgestellten Modus, um über diese Grenzen hinaus zu skalieren. On-demand Für den Modus ist neben den standardmäßigen Zuordnungsparametern für die Ereignisquelle keine zusätzliche Konfiguration erforderlich.

In der folgenden Tabelle sind die Grenzwerte für den On-Demand-Modus aufgeführt.

Limit	Wert
Durchsatz pro Event-Poller	Bis zu 1 MB/s
Gleichzeitige Aufrufe pro Event-Poller	Bis zu 10
Maximale Anzahl von Event-Pollern	5
Maximaler Gesamtdurchsatz	Bis zu 5 MB/s (5 Poller mal 1 MB/s)
Maximale aggregierte Parallelität	Bis zu 50 gleichzeitige Aufrufe (5 Poller mal 10)

Beispiel — Zuordnung der Ereignisquelle im Warteschlangenmodus im On-Demand-Modus (AWS CLI)

Folgendes AWS Der CLI-Befehl erstellt eine Ereignisquellenzuordnung im Warteschlangenmodus im On-Demand-Modus. Ersetzen my-function, die Broker-Endpunkte, my-queue-topic, die Subnetz-IDs und die Sicherheitsgruppen-ID mit Ihren eigenen Werten.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

Modus bereitgestellter Kapazität

Verwenden Sie für Workloads mit strengen Leistungsanforderungen oder starken Datenverkehrsmustern den Bereitstellungsmodus. Im Bereitstellungsmodus haben Sie die direkte Kontrolle über die Mindest- und Höchstanzahl von Event-Pollern, die für die Zuordnung der Ereignisquellen im Warteschlangenmodus vorgesehen sind.

Jeder Event-Poller im Warteschlangenmodus im bereitgestellten Modus unterstützt bis zu 5 MB/s des Durchsatzes und bis zu 10 gleichzeitige Lambda-Aufrufe.

In der folgenden Tabelle sind die Grenzwerte für den bereitgestellten Modus aufgeführt.

Limit	Wert
Durchsatz pro Event-Poller	Bis zu 5 MB/s
Gleichzeitige Aufrufe pro Event-Poller	Bis zu 10
Mindestanzahl von Event-Pollern (MinimumPollers)	1—200. Standard: 1
Maximale Anzahl von Event-Pollern (MaximumPollers)	1—2.000. Standard: 200
Zuordnungen der Ereignisquellen pro Poller-Gruppe	Bis zu 100
Aggregieren Sie die maximale Anzahl von Pollern pro Poller-Gruppe	2.000
Durchsatz pro EPU (aggregiert)	Bis zu 5 MB/s pro Poller; das Aggregat hängt von der Packungsdichte des Pollers ab

Der bereitgestellte Modus wird auf der Grundlage der Nutzung der Event Poller Unit (EPU) berechnet. Details zu den Preisen finden Sie unter AWS Lambda -Preise.

Beispiel — Zuordnung der Ereignisquelle im Warteschlangenmodus mit dem bereitgestellten Modus (AWS CLI)

Folgendes AWS Der CLI-Befehl erstellt eine Ereignisquellenzuordnung im Warteschlangenmodus im bereitgestellten Modus.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

Ersetzen Sie die folgenden Platzhalter durch Ihre eigenen Werte:

• subnet-0123456789abcdef0 und subnet-0123456789abcdef1: Ihre Subnetz-IDs.

• sg-0123456789abcdef0: Ihre Sicherheitsgruppen-ID.

• MinimumPollers und MaximumPollers: entsprechend Ihren Durchsatzanforderungen einstellen.

Anmerkung

Um vom Bereitstellungsmodus zurück zum On-Demand-Modus zu wechseln, geben Sie ein leeres Feld ein ProvisionedPollerConfig:

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

Auswahl des Skalierungsmodus

Überlegungen	On-demand	Bereitgestellt
Einrichtung	Keine zusätzliche Konfiguration	Konfiguriere MinimumPollers und MaximumPollers
Cost (Kosten)	Keine zusätzlichen EPU-Gebühren	Die EPU-Gebühren basieren auf der Nutzung pro Stunde
Am besten geeignet für	Low-concurrency Arbeitsbelastungen	Strenge Latenzanforderungen, starker Datenverkehr
Gruppierung der Zuordnung der Ereignisquelle (PollerGroupName)	Nicht unterstützt	Unterstützt. Teilen Sie EPUs über Quelle-Quellzuordnungen von Queue-Ereignissen hinweg
Low-latency Verarbeitung	Nicht verfügbar	Available (Verfügbar)
Filterung der Schemaregistrierung (Avro/Protobuf)	Nicht verfügbar	Available (Verfügbar)

EPU teilt sich die Isolierung

Im bereitgestellten Modus können Sie mehrere Ereignisquellenzuordnungen innerhalb derselben VPC gruppieren, um die Kapazität der Event Poller Unit (EPU) gemeinsam zu nutzen, indem Sie ihnen dieselbe zuweisen PollerGroupName. Durch die Gruppierung der Ereignisquellenzuordnung können Ereignisquellenzuordnungen mit geringem Durchsatz auf gemeinsam genutzte EPUs gepackt werden, sodass nicht für jede einzelne Einheit dedizierte Kapazität erforderlich ist, wodurch die Kosten für den Bereitstellungsmodus gesenkt werden. Details hierzu finden Sie unter Kostenoptimierung für den Bereitstellungsmodus.

Zuordnungen von Ereignisquellen im Warteschlangenmodus verwenden EPUs nur gemeinsam mit anderen Zuordnungen von Ereignisquellen im Warteschlangenmodus. EIN PollerGroupName Die von einer Ereignisquellenzuordnung im Warteschlangenmodus verwendete Ereignisquellenzuordnung kann nicht mit einer Kafka Streams-Ereignisquellenzuordnung gemeinsam genutzt werden und umgekehrt. Wenn Sie eine konfigurieren PollerGroupName wird bereits vom anderen Verbrauchsmodus verwendet, CreateEventSourceMapping gibt einen Validierungsfehler zurück. Um die Gruppierung der Ereignisquellenzuordnung im Warteschlangenmodus zu verwenden, erstellen Sie eine PollerGroupName speziell für Ihre Quelle-Zuordnungen für Queue-Ereignisse reserviert.

Beispielereignis

Wenn Lambda Ihre Funktion mit einer Ereignisquelle im Warteschlangenmodus aufruft, ist die Nutzlast des Ereignisses identisch mit einem Standard-Kafka-Ereignis. Jeder Datensatz enthält das Thema, die Partition, den Offset, den Zeitstempel sowie den Base64-kodierten Schlüssel und Wert:

{
  "eventSource": "SelfManagedKafka",
  "bootstrapServers": "broker1.example.com:9092,broker2.example.com:9092",
  "records": {
    "my-queue-topic-0": [
      {
        "topic": "my-queue-topic",
        "partition": 0,
        "offset": 15,
        "timestamp": 1545084650987,
        "timestampType": "CREATE_TIME",
        "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==",
        "value": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
        "headers": [
          {
            "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
          }
        ]
      }
    ],
    "my-queue-topic-3": [
      {
        "topic": "my-queue-topic",
        "partition": 3,
        "offset": 42,
        "timestamp": 1545084651000,
        "timestampType": "CREATE_TIME",
        "key": "YW5vdGhlcktleQ==",
        "value": "VGhpcyBpcyBhbm90aGVyIG1lc3NhZ2U=",
        "headers": []
      }
    ]
  }
}

Idempotenz erforderlich

Lambda-Ereignisquellenzuordnungen verarbeiten jedes Ereignis mindestens einmal, sodass es zu einer doppelten Verarbeitung von Datensätzen kommen kann. Da der Warteschlangenmodus die Reihenfolge nicht garantiert, kann eine doppelte oder falsche Verarbeitung zu falschen Ergebnissen führen, z. B. zu Doppelzählungen oder einem inkonsistenten Zustand, wenn Ihr Funktionscode nicht idempotent ist. Machen Sie Ihren Funktionscode idempotent.

Einschränkungen

• Self-managed Nur Kafka-Cluster. Amazon MSK wird beim Start nicht unterstützt.

• Apache Kafka 4.2 oder höher erforderlich. Der Cluster muss das Share Consumer-Protokoll mit aktivierten Share-Gruppen unterstützen.

• Das Batching-Fenster wird nicht unterstützt. Das MaximumBatchingWindowInSeconds Der Parameter kann nicht mit Kafka Queue-Ereignisquellenzuordnungen verwendet werden.

• Die Startposition wird nicht unterstützt. StartingPosition und StartingPositionTimestamp kann nicht mit Ereignisquellenzuordnungen im Warteschlangenmodus verwendet werden. Die Share-Gruppe verbraucht von ihrer zugesagten Offset-Position aus.

• EPU-Isolierung. Zuordnungen von Ereignisquellen im Warteschlangenmodus können EPUs nicht gemeinsam mit Kafka Streams-Ereignisquellenzuordnungen verwenden. Verwenden Sie separate PollerGroupName Werte.