Sondaggio degli argomenti di Kafka in modalità Coda

La modalità coda è una modalità di consumo che utilizza gruppi di condivisione, che è diventata disponibile in Apache Kafka 4.0 (Kafka Queue Poller richiede la versione 4.2 o successiva). Con la modalità Queue, puoi elaborare i messaggi relativi agli argomenti di Kafka in modo non ordinato e ad alta velocità utilizzando i gruppi di condivisione. A differenza dei tradizionali gruppi di consumatori di Kafka che assegnano le partizioni esclusivamente ai consumatori, i gruppi di condivisione consentono a più consumatori di leggere dalle stesse partizioni contemporaneamente. È simile al modo in cui Amazon SQS elabora i messaggi provenienti da una coda.

Quando si crea una mappatura della sorgente di un evento per un evento autogestito Apache Kafka cluster, puoi scegliere tra due modalità di consumo:

• Stream (impostazione predefinita). Utilizza i gruppi di consumatori Kafka con ordinamento per partizione. Lambda assegna ogni partizione a un consumatore.

• Coda. Utilizza i gruppi di condivisione, che sono diventati disponibili in Apache Kafka 4.0, per un'elaborazione non ordinata e ad alto rendimento (Kafka Queue Poller richiede la versione 4.2 o successiva). Con i gruppi di condivisione, più consumatori possono leggere contemporaneamente dalle stesse partizioni, in modo simile a come Amazon SQS elabora i messaggi da una coda.

Se non si specifica una modalità di consumo, la mappatura della sorgente degli eventi utilizza Stream. Questa pagina descrive la modalità di consumo della coda. Per informazioni generali sull'uso di Lambda con gestione automatica Apache Kafka, vedi Utilizzo di Lambda con Apache Kafka gestito dal cliente.

In modalità Queue, Lambda utilizza Kafka Share Consumer per interrogare i messaggi relativi all'argomento. Ogni event poller elabora i messaggi su tutte le partizioni senza ordinarli a livello di partizione. Ciò aumenta la concorrenza per ogni poller.

Requisiti minimi

Per utilizzare la modalità di consumo in coda, il cluster Kafka deve soddisfare i seguenti requisiti:

• Apache Kafka versione 4.2 o successiva con i gruppi di condivisione abilitati. Kafka abilita share.version per impostazione predefinita, e puoi gestirlo tramite kafka-features.sh.

• Self-managed Apache Kafka (SMK) solo cluster. Amazon Managed Streaming per Apache Kafka (Amazon MSK) attualmente non supporta la modalità Queue. AWS aggiunge il supporto MSK una volta che MSK offre Kafka 4.2 o versioni successive con i gruppi di condivisione abilitati.

Confronto tra stream e code

Funzionalità	Stream (impostazione predefinita)	Queue
Ordinazione	Per-partition ordinazione garantita	Nessuna garanzia di ordinazione
Modello di consumo	Gruppo di consumatori (assegnazione esclusiva delle partizioni)	Gruppo di condivisione (accesso simultaneo alle partizioni)
Produttività per poller di eventi	Fino a 5 MB/s	Fino a 5 MB/s
Invocazioni simultanee per poller di eventi	Fino a 5	Fino a 10
Modalità di ridimensionamento	On-demand e predisposto	On-demand e accantonato
condivisione dell'EPU	Può condividere le EPU con altre mappature delle sorgenti di eventi di Kafka Streams	Isolato. Le mappature delle origini degli eventi in coda sono condivise solo con altre mappature delle origini degli eventi in coda
Caso d’uso	Elaborazione ordinata degli eventi	High-throughput carichi di lavoro non ordinati

Crea una mappatura delle code

Per creare una mappatura della sorgente degli eventi in modalità Queue, imposta ConsumptionMode campo di SelfManagedKafkaEventSourceConfig a Queue. Negli esempi seguenti, sostituisci my-function, gli endpoint del broker e my-queue-topic con i tuoi valori.

AWS CLI

Quanto segue AWS Il comando CLI crea una mappatura della sorgente degli eventi in modalità Queue. Sostituisci my-function con il nome della tua funzione, i terminali del broker con i tuoi broker e my-queue-topic con il tuo argomento.

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

Quanto segue AWS CloudFormation il modello crea una mappatura della sorgente degli eventi in modalità Queue. Sostituisci gli endpoint del broker e my-queue-topic con i tuoi valori.

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

Parametri di configurazione

La tabella seguente elenca i parametri applicabili alle mappature delle sorgenti degli eventi in modalità Queue.

Parametro	Description	Intervallo/Predefinito
BatchSize	Numero massimo di record per batch inviati alla funzione	1—10.000. Impostazione predefinita: 100
Enabled	Se la mappatura della sorgente dell'evento è attiva	true o false. Impostazione predefinita: true
FilterCriteria	Schemi di filtraggio degli eventi per elaborare solo i record corrispondenti	Fino a 5 modelli di filtro
MaximumRetryAttempts	Numero massimo di tentativi per i record non riusciti	-1—10.000. Predefinito: 0 (in modalità Coda, il broker Kafka esegue nuovi tentativi quando non è configurata alcuna destinazione in caso di errore)
BisectBatchOnFunctionError	Dividi il batch a metà in caso di errore di funzione	true o false. Impostazione predefinita: false
ProvisionedPollerConfig.MinimumPollers	Poller di eventi minimi (modalità predisposta)	1—200. Impostazione predefinita: 1
ProvisionedPollerConfig.MaximumPollers	Numero massimo di poller di eventi (modalità predisposta)	1—2.000. Impostazione predefinita: 200
ProvisionedPollerConfig.PollerGroupName	Nome del gruppo per la condivisione EPU (modalità con provisioning, solo mappature delle sorgenti degli eventi in coda)	Stringa. Fino a 100 mappature delle sorgenti di eventi per gruppo
SelfManagedKafkaEventSourceConfig.ConsumptionMode	Modalità di consumo per la fonte dell'evento	Deve essere Queue per la modalità Queue
SchemaRegistryConfig	Registro degli schemi per la convalida degli eventi	Per informazioni, consultare Utilizzo dei registri degli schemi con le sorgenti di eventi Kafka in Lambda

Importante

Il MaximumBatchingWindowInSeconds il parametro è non supportato per le mappature delle sorgenti degli eventi in modalità Queue. Kafka Share Consumer utilizza un riconoscimento esplicito. Tutti i record di un batch devono essere riconosciuti esplicitamente prima che il consumatore possa recuperarne di nuovi. Ciò non è compatibile con una finestra di batch basata sul tempo. Se si specifica MaximumBatchingWindowInSeconds quando si crea una mappatura della sorgente degli eventi in modalità Queue, l'API restituisce un errore di convalida.

Funzionalità di mappatura delle sorgenti di eventi supportate

La tabella seguente elenca le funzionalità di mappatura delle sorgenti di eventi Lambda disponibili con la modalità Queue.

Funzionalità	Supportato	Note
Modalità provisioning	Sì	Configura MinimumPollers e MaximumPollers
On-demand modalità	Sì	Modalità di ridimensionamento predefinita. Lambda ridimensiona automaticamente i poller fino a 5 MB/s o 50 invocazioni simultanee, a seconda di quale valore viene raggiunto per primo. Non è richiesta alcuna configurazione aggiuntiva.
Filtro eventi	Sì	Fino a 5 modelli di filtro per mappatura della sorgente dell'evento
Registro degli schemi	Sì	Avro e Protobuf con Glue Schema Registry o Confluent Schema Registry. Il filtro del registro degli schemi è disponibile solo in modalità provisioning.
Low-latency in elaborazione	Sì	Disponibile in modalità predisposta
Riprova le configurazioni	Sì	MaximumRetryAttempts, BisectBatchOnFunctionError
On-failure destinazione (argomento Kafka)	Sì	Inviare record non riusciti a un argomento di Kafka
On-failure destinazione (SNS/SQS)	Sì	Inviare i record non riusciti a SNS o SQS
Registrazione della mappatura dell'origine degli eventi	Sì	CloudWatch Log per la diagnostica della mappatura delle sorgenti di eventi
Raggruppamento della mappatura delle sorgenti degli eventi (PollerGroupName)	Sì	Le mappature delle sorgenti di eventi in modalità coda possono condividere le EPU con altre mappature delle sorgenti di eventi in modalità coda solo (modalità provisioned)
Finestra di raggruppamento	No	Non supportata a causa del comportamento della modalità di blocco di Share Consumer
Posizione di inizio	No	StartingPosition e StartingPositionTimestamp non sono supportate. La modalità coda legge sempre dalla posizione impegnata del gruppo di condivisione.

Modalità di ridimensionamento

Le mappature delle sorgenti degli eventi in modalità coda supportano sia le modalità di ridimensionamento su richiesta (impostazione predefinita) che con provisioning.

On-demand modalità (impostazione predefinita)

In modalità on-demand, Lambda gestisce automaticamente i poller di eventi per la mappatura della sorgente degli eventi in modalità Queue.

On-demand la modalità scala fino a 5 poller di eventi per mappatura della sorgente di eventi. Raggiunge il suo limite massimo a 5 MB/s throughput aggregato o 50 chiamate simultanee (5 poller per 10). Ciò accade a seconda del limite raggiunto per primo. Per superare questi limiti, utilizza la modalità provvisoria. On-demand la modalità non richiede alcuna configurazione aggiuntiva oltre ai parametri standard di mappatura della sorgente degli eventi.

La tabella seguente elenca i limiti della modalità su richiesta.

Limite	Valore
Throughput per poller di eventi	Fino a 1 MB/s
Invocazioni simultanee per poller di eventi	Fino a 10
Numero massimo di poller di eventi	5
Produttività aggregata massima	Fino a 5 MB/s (5 sondaggi moltiplicati per 1) MB/s)
Concorrenza aggregata massima	Fino a 50 chiamate simultanee (5 poller per 10)

Esempio: mappatura della sorgente degli eventi in modalità coda in modalità su richiesta (AWS CLI)

Quanto segue AWS Il comando CLI crea una mappatura della sorgente degli eventi in modalità Queue in modalità on-demand. Sostituisci my-function, gli endpoint del broker, my-queue-topic, gli ID della sottorete e l'ID del gruppo di sicurezza con valori personalizzati.

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"}]'

Modalità provisioning

Per carichi di lavoro con requisiti prestazionali rigorosi o modelli di traffico intensi, utilizza la modalità provisioned. La modalità provisioned ti offre il controllo diretto sul numero minimo e massimo di poller di eventi dedicati alla mappatura delle sorgenti degli eventi in modalità Queue.

Ogni poller di eventi in modalità Queue in modalità provisioning supporta fino a 5 MB/s di velocità effettiva e fino a 10 chiamate Lambda simultanee.

La tabella seguente elenca i limiti della modalità predisposta.

Limite	Valore
Throughput per poller di eventi	Fino a 5 MB/s
Invocazioni simultanee per poller di eventi	Fino a 10
Poller minimi di eventi (MinimumPollers)	1—200. Impostazione predefinita: 1
Numero massimo di poller di eventi (MaximumPollers)	1—2.000. Impostazione predefinita: 200
Mappature delle sorgenti degli eventi per gruppo di poller	Fino a 100
Aggrega il numero massimo di poller per gruppo di poller	2.000
Rendimento per EPU (aggregato)	Fino a 5 MB/s per poller; l'aggregato dipende dalla densità di imballaggio del poller

La modalità provisioned viene addebitata in base all'utilizzo dell'Event Poller Unit (EPU). Per i dettagli sui prezzi, vedere Prezzi di AWS Lambda.

Esempio: mappatura della sorgente degli eventi in modalità coda con modalità provisioning (AWS CLI)

Quanto segue AWS Il comando CLI crea una mappatura della sorgente degli eventi in modalità Queue in modalità provisioning.

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"}]'

Sostituisci i seguenti segnaposto con i tuoi valori:

• subnet-0123456789abcdef0 e subnet-0123456789abcdef1: i tuoi ID di sottorete.

• sg-0123456789abcdef0: l'ID del tuo gruppo di sicurezza.

• MinimumPollers e MaximumPollers: impostato in base alle esigenze di produttività.

Nota

Per passare dalla modalità provisioning alla modalità on-demand, passate un campo vuoto ProvisionedPollerConfig:

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

Selezione della modalità di ridimensionamento

Considerazione	On-demand	Assegnata
Configurazione	Nessuna configurazione aggiuntiva	Configura MinimumPollers e MaximumPollers
Costo	Nessun costo EPU aggiuntivo	Addebiti EPU in base all'utilizzo orario
Ideale per	Low-concurrency carichi di lavoro	Requisiti di latenza rigorosi, traffico intenso
Raggruppamento della mappatura delle sorgenti degli eventi (PollerGroupName)	Non supportata	Supportato. Condividi le EPU tra le mappature delle sorgenti degli eventi di Queue
Low-latency in elaborazione	Non disponibile	Disponibilità
Filtraggio del registro degli schemi (Avro/Protobuf)	Non disponibile	Disponibilità

L'EPU condivide l'isolamento

In modalità provisioned, è possibile raggruppare più mappature delle sorgenti di eventi all'interno dello stesso VPC per condividere la capacità dell'Event Poller Unit (EPU) assegnandole allo stesso modo PollerGroupName. Il raggruppamento della mappatura delle sorgenti di eventi consente di raggruppare le mappature delle sorgenti di eventi a basso rendimento su EPU condivise anziché richiedere ciascuna capacità dedicata, riducendo i costi in modalità provisioning. Per informazioni dettagliate, vedi Ottimizzazione dei costi per la modalità Provisioned.

Le mappature delle sorgenti di eventi in modalità coda condividono le EPU solo con altre mappature delle sorgenti di eventi in modalità coda. A) PollerGroupName utilizzato da una mappatura della sorgente di eventi in modalità Queue non può essere condiviso con una mappatura della sorgente di eventi di Kafka Streams e viceversa. Se si configura un PollerGroupName già utilizzato dall'altra modalità di consumo, CreateEventSourceMapping restituisce un errore di convalida. Per utilizzare il raggruppamento della mappatura delle sorgenti degli eventi con la modalità Queue, crea un PollerGroupName dedicato alle mappature delle sorgenti degli eventi in coda.

Esempio di evento

Quando Lambda richiama la funzione con una sorgente di eventi in modalità Queue, il payload dell'evento è identico a un evento Kafka standard. Ogni record include l'argomento, la partizione, l'offset, il timestamp e la chiave e il valore codificati in base64:

{
  "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": []
      }
    ]
  }
}

È richiesta l'idempotenza

Le mappature delle sorgenti degli eventi Lambda elaborano ogni evento almeno una volta, quindi può verificarsi un'elaborazione duplicata dei record. Poiché la modalità Queue non garantisce l'ordinamento, l'elaborazione duplicata o fuori ordine può produrre risultati errati, ad esempio un doppio conteggio o uno stato incoerente, se il codice funzione non è idempotente. Rendi idempotente il tuo codice funzione.

Limitazioni

• Self-managed Solo cluster Kafka. Amazon MSK non è supportato al momento del lancio.

• Apache Kafka È richiesto 4.2 o versione successiva. Il cluster deve supportare il protocollo Share Consumer con i gruppi di condivisione abilitati.

• La finestra di batch non è supportata. La MaximumBatchingWindowInSeconds il parametro non può essere utilizzato con le mappature delle sorgenti degli eventi di Kafka Queue.

• La posizione iniziale non è supportata. StartingPosition e StartingPositionTimestamp non può essere utilizzato con le mappature delle sorgenti degli eventi in modalità Queue. Il gruppo azionario consuma a partire dalla sua posizione di offset impegnata.

• Isolamento dall'EPU. Le mappature delle sorgenti degli eventi in modalità coda non possono condividere le EPU con le mappature delle sorgenti di eventi di Kafka Streams. Usa separatamente PollerGroupName valori.