Tutorial: utilizzo di una mappatura delle sorgenti di MSK eventi Amazon per richiamare una funzione Lambda - AWS Lambda
PrerequisitiConfigura la connettività di rete per consentire a Lambda di comunicare con Amazon MSKCrea un IAM ruolo per Lambda da leggere dal tuo argomento su Amazon MSKCrea una funzione Lambda da leggere dal tuo argomento Amazon MSKCrea una mappatura della sorgente degli eventi sulla tua funzione LambdaAggiorna la funzione Lambda per leggere i dati di streamingTesta la tua funzione Lambda per verificare che sia collegata al tuo argomento Amazon MSK

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Tutorial: utilizzo di una mappatura delle sorgenti di MSK eventi Amazon per richiamare una funzione Lambda

In questo tutorial, eseguirai quanto segue:

  • Crea una funzione Lambda nello stesso AWS account di un cluster Amazon MSK esistente.

  • Configura la rete e l'autenticazione per consentire a Lambda di comunicare con Amazon. MSK

  • Configura una mappatura delle sorgenti MSK degli eventi Lambda Amazon, che esegue la funzione Lambda quando vengono visualizzati gli eventi nell'argomento.

Dopo aver completato questi passaggi, quando gli eventi vengono inviati ad AmazonMSK, potrai configurare una funzione Lambda per elaborarli automaticamente con il tuo codice Lambda personalizzato.

Cosa puoi fare con questa funzionalità?

Soluzione di esempio: utilizza una mappatura delle fonti MSK degli eventi per fornire punteggi in tempo reale ai tuoi clienti.

Considerate lo scenario seguente: la vostra azienda ospita un'applicazione web in cui i clienti possono visualizzare informazioni sugli eventi dal vivo, come le partite sportive. Gli aggiornamenti delle informazioni dal gioco vengono forniti al tuo team tramite un argomento di Kafka su Amazon. MSK Vuoi progettare una soluzione che utilizzi gli aggiornamenti dell'MSKargomento per fornire una visione aggiornata dell'evento dal vivo ai clienti all'interno di un'applicazione da te sviluppata. Avete deciso il seguente approccio di progettazione: le vostre applicazioni client comunicheranno con un backend serverless ospitato in. AWS I client si connetteranno tramite sessioni websocket utilizzando Amazon API Gateway WebSocket API.

In questa soluzione, è necessario un componente che legga gli MSK eventi, esegua una logica personalizzata per prepararli per il livello dell'applicazione e quindi inoltrare tali informazioni al Gateway. API API Puoi implementare questo componente con AWS Lambda, fornendo la tua logica personalizzata in una funzione Lambda, quindi chiamandolo con una mappatura della sorgente MSK degli eventi AWS Lambda Amazon.

Per ulteriori informazioni sull'implementazione di soluzioni utilizzando Amazon API Gateway WebSocket API, consulta WebSocket APIi tutorial nella documentazione di API Gateway.

Prerequisiti

Un AWS account con le seguenti risorse preconfigurate:

Per soddisfare questi prerequisiti, ti consigliamo di seguire la sezione Guida introduttiva a usare Amazon MSK nella MSK documentazione di Amazon.

  • Un MSK cluster Amazon. Consulta Creare un MSK cluster Amazon in Guida introduttiva all'uso di Amazon MSK.

  • La seguente configurazione:

    • Assicurati che l'autenticazione IAM basata sui ruoli sia abilitata nelle impostazioni di sicurezza del cluster. Ciò migliora la sicurezza limitando la funzione Lambda all'accesso solo alle risorse MSK Amazon necessarie. Questa opzione è abilitata per impostazione predefinita sui nuovi MSK cluster Amazon.

    • Assicurati che l'accesso pubblico sia disattivato nelle impostazioni di rete del cluster. Limitare l'accesso a Internet del tuo MSK cluster Amazon migliora la tua sicurezza limitando il numero di intermediari che gestiscono i tuoi dati. Questa opzione è abilitata per impostazione predefinita sui nuovi MSK cluster Amazon.

  • Un argomento di Kafka nel tuo MSK cluster Amazon da utilizzare per questa soluzione. Consulta Creare un argomento in Guida introduttiva all'uso di Amazon MSK.

  • Un host di amministrazione Kafka configurato per recuperare informazioni dal tuo cluster Kafka e inviare eventi Kafka al tuo argomento per i test, ad esempio un'istanza Amazon EC2 con l'amministratore Kafka e la libreria Amazon installati. CLI MSK IAM Vedi Creare una macchina client in Guida introduttiva a usare Amazon MSK.

Dopo aver configurato queste risorse, raccogli le seguenti informazioni dal tuo AWS account per confermare che sei pronto a continuare.

  • Il nome del tuo MSK cluster Amazon. Puoi trovare queste informazioni nella MSK console Amazon.

  • Il clusterUUID, parte del MSK cluster ARN for your Amazon, che puoi trovare nella MSK console Amazon. Segui le procedure in Listing clusters nella MSK documentazione di Amazon per trovare queste informazioni.

  • I gruppi di sicurezza associati al tuo MSK cluster Amazon. Puoi trovare queste informazioni nella MSK console Amazon. Nei passaggi seguenti, fai riferimento a queste come alle tue clusterSecurityGroups.

  • L'id dell'Amazon VPC contenente il tuo MSK cluster Amazon. Puoi trovare queste informazioni identificando le sottoreti associate al tuo MSK cluster Amazon nella MSK console Amazon, quindi identificando l'Amazon VPC associata alla sottorete nella console Amazon. VPC

  • Il nome dell'argomento Kafka utilizzato nella tua soluzione. Puoi trovare queste informazioni chiamando il tuo MSK cluster Amazon con Kafka topics CLI dal tuo host di amministrazione Kafka. Per ulteriori informazioni sugli argomentiCLI, consulta Aggiungere e rimuovere argomenti nella documentazione di Kafka.

  • Il nome di un gruppo di consumatori per l'argomento Kafka, adatto all'uso con la funzione Lambda. Questo gruppo può essere creato automaticamente da Lambda, quindi non è necessario crearlo con Kafka. CLI Se devi gestire i tuoi gruppi di consumatori, per saperne di più sui gruppi di consumatori, consulta Managing Consumer Groups nella CLI documentazione di Kafka.

Le seguenti autorizzazioni nel tuo account: AWS

  • Autorizzazione a creare e gestire una funzione Lambda.

  • Autorizzazione a creare IAM policy e associarle alla funzione Lambda.

  • Autorizzazione a creare VPC endpoint Amazon e modificare la configurazione di rete nell'Amazon che VPC ospita il tuo MSK cluster Amazon.

Configura la connettività di rete per consentire a Lambda di comunicare con Amazon MSK

AWS PrivateLink Usalo per connettere Lambda e Amazon. MSK Puoi farlo creando VPC endpoint Amazon di interfaccia nella VPC console Amazon. Per ulteriori informazioni sulla configurazione di rete, consultaConfigurazione della rete.

Quando una mappatura delle sorgenti di MSK eventi Amazon viene eseguita per conto di una funzione Lambda, assume il ruolo di esecuzione della funzione Lambda. Questo IAM ruolo autorizza la mappatura per accedere a risorse protette daIAM, come il tuo cluster Amazon. MSK Sebbene i componenti condividano un ruolo di esecuzione, la MSK mappatura Amazon e la funzione Lambda hanno requisiti di connettività separati per le rispettive attività, come illustrato nel diagramma seguente.

Una funzione Lambda interroga un cluster e comunica con Lambda utilizzando. AWS STS

La mappatura delle sorgenti degli eventi appartiene al gruppo di sicurezza MSK del cluster Amazon. In questa fase di networking, crea VPC endpoint Amazon dal tuo MSK cluster Amazon VPC per connettere la mappatura delle sorgenti degli eventi a Lambda e ai servizi. STS Proteggi questi endpoint per accettare il traffico proveniente dal tuo gruppo di sicurezza MSK del cluster Amazon. Quindi, modifica i gruppi di sicurezza del MSK cluster Amazon per consentire alla mappatura delle sorgenti degli eventi di comunicare con il MSK cluster Amazon.

Puoi configurare i seguenti passaggi utilizzando. AWS Management Console

Per configurare l'interfaccia Amazon VPC endpoint per connettere Lambda e Amazon MSK

  1. Crea un gruppo di sicurezza per la tua interfaccia Amazon VPC endpoints, endpointSecurityGroup, che consente il TCP traffico in entrata su 443 da clusterSecurityGroups. Segui la procedura in Creare un gruppo di sicurezza nella EC2 documentazione di Amazon per creare un gruppo di sicurezza. Quindi, segui la procedura in Aggiungere regole a un gruppo di sicurezza nella EC2 documentazione di Amazon per aggiungere le regole appropriate.

    Crea un gruppo di sicurezza con le seguenti informazioni:

    Quando aggiungi le regole in entrata, crea una regola per ogni gruppo di sicurezza in clusterSecurityGroups. Per ogni regola:

    • Per Tipo, seleziona HTTPS.

    • Per Sorgente, selezionate una delle clusterSecurityGroups.

  2. Crea un endpoint che collega il servizio Lambda all'Amazon contenente il tuo VPC cluster AmazonMSK. Segui la procedura descritta in Creare un endpoint di interfaccia.

    Crea un endpoint di interfaccia con le seguenti informazioni:

    • Per il nome del servizio, selezionacom.amazonaws.regionName.lambda, dove regionName ospita la tua funzione Lambda.

    • Per VPC, seleziona l'Amazon VPC contenente il tuo MSK cluster Amazon.

    • Per i gruppi di sicurezza, seleziona endpointSecurityGroup, che hai creato in precedenza.

    • Per Subnet, seleziona le sottoreti che ospitano il tuo cluster Amazon. MSK

    • Per Policy, fornisci il seguente documento di policy, che protegge l'endpoint per l'utilizzo da parte del responsabile del servizio Lambda per l'azione. lambda:InvokeFunction

      {
    "Statement": [
        {
            "Action": "lambda:InvokeFunction",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            },
            "Resource": "*"
        }
    ]
}

    • Assicurati che il DNSnome Enable rimanga impostato.

  3. Crea un endpoint che colleghi il AWS STS servizio all'Amazon VPC contenente il tuo MSK cluster Amazon. Segui la procedura descritta in Creare un endpoint di interfaccia.

    Crea un endpoint di interfaccia con le seguenti informazioni:

    • Per Nome del servizio, selezionare AWS STS.

    • Per VPC, seleziona l'Amazon VPC contenente il tuo MSK cluster Amazon.

    • Per i gruppi di sicurezza, seleziona endpointSecurityGroup.

    • Per Subnet, seleziona le sottoreti che ospitano il tuo cluster Amazon. MSK

    • Per Policy, fornisci il seguente documento di policy, che protegge l'endpoint per l'utilizzo da parte del responsabile del servizio Lambda per l'azione. sts:AssumeRole

      {
    "Statement": [
        {
            "Action": "sts:AssumeRole",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            },
            "Resource": "*"
        }
    ]
}

    • Assicurati che il DNSnome Enable rimanga impostato.

  4. Per ogni gruppo di sicurezza associato al tuo MSK cluster Amazon, ovvero in clusterSecurityGroups, consenti quanto segue:

    • Consenti tutto il TCP traffico in entrata e in uscita su 9098 a tutti clusterSecurityGroups, anche al suo interno.

    • Consenti tutto il TCP traffico in uscita su 443.

    Parte di questo traffico è consentito dalle regole predefinite del gruppo di sicurezza, quindi se il cluster è collegato a un singolo gruppo di sicurezza e tale gruppo ha regole predefinite, non sono necessarie regole aggiuntive. Per modificare le regole del gruppo di sicurezza, segui le procedure in Aggiungere regole a un gruppo di sicurezza nella EC2 documentazione di Amazon.

    Aggiungi regole ai tuoi gruppi di sicurezza con le seguenti informazioni:

    • Per ogni regola in entrata o in uscita per la porta 9098, fornisci

      • Per Tipo, seleziona Personalizzato. TCP

      • Per Port range, fornisci 9098.

      • Per Source, fornisci uno dei clusterSecurityGroups.

    • Per ogni regola in entrata per la porta 443, per Tipo, seleziona. HTTPS

Crea un IAM ruolo per Lambda da leggere dal tuo argomento su Amazon MSK

Identifica i requisiti di autenticazione per Lambda da leggere nell'argomento di MSK Amazon, quindi definiscili in una policy. Crea un ruolo, lambdaAuthRole, che autorizza Lambda a utilizzare tali autorizzazioni. Autorizza le azioni sul tuo MSK cluster Amazon utilizzando kafka-cluster IAM le azioni. Quindi, autorizza Lambda a eseguire le azioni Amazon MSK kafka e EC2 Amazon necessarie per scoprire e connettersi al tuo cluster MSK Amazon, CloudWatch nonché le azioni in modo che Lambda possa registrare ciò che ha fatto.

Per descrivere i requisiti di autenticazione per Lambda da leggere da Amazon MSK

  1. Scrivi un documento IAM politico (un JSON documento), clusterAuthPolicy, che consente a Lambda di leggere il tuo argomento Kafka nel tuo MSK cluster Amazon utilizzando il tuo gruppo di consumatori Kafka. Lambda richiede che durante la lettura sia impostato un gruppo di consumatori Kafka.

    Modifica il seguente modello per allinearlo ai tuoi prerequisiti:

    {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:DescribeGroup",
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:ReadData",
                "kafka-cluster:DescribeClusterDynamicConfiguration"
            ],
            "Resource": [
                "arn:aws:kafka:region:account-id:cluster/mskClusterName/cluster-uuid",
                "arn:aws:kafka:region:account-id:topic/mskClusterName/cluster-uuid/mskTopicName",
                "arn:aws:kafka:region:account-id:group/mskClusterName/cluster-uuid/mskGroupName"
            ]
        }
    ]
}

    Per ulteriori informazioni, consultare. Autenticazione basata su ruoli IAM Quando scrivi la tua politica:

    • In region e account-id, fornisci quelli che ospitano il tuo MSK cluster Amazon.

    • In mskClusterName, fornisci il nome del tuo MSK cluster Amazon.

    • In cluster-uuid, UUID forniscili nel ARN tuo MSK cluster Amazon.

    • In mskTopicName, fornisci il nome del tuo argomento su Kafka.

    • In mskGroupName, fornisci il nome del tuo gruppo di consumatori Kafka.

  2. Identifica AmazonMSK, Amazon EC2 e CloudWatch le autorizzazioni necessarie a Lambda per scoprire e connettere il tuo cluster MSK Amazon e registra tali eventi.

    La policy AWSLambdaMSKExecutionRole gestita definisce in modo permissivo le autorizzazioni richieste. Usalo nei seguenti passaggi.

    In un ambiente di produzione, valuta se AWSLambdaMSKExecutionRole limitare la politica del ruolo di esecuzione in base al principio del privilegio minimo, quindi scrivi una politica per il tuo ruolo che sostituisca questa politica gestita.

Per i dettagli sul linguaggio delle IAM politiche, consulta la IAM documentazione.

Ora che hai scritto il tuo documento programmatico, crea una IAM politica in modo da poterla allegare al tuo ruolo. È possibile eseguire questa operazione utilizzando la console con la procedura seguente.

Per creare una IAM politica dal documento di policy

  1. Accedi a AWS Management Console e apri la IAM console all'indirizzo https://console.aws.amazon.com/iam/.

  2. Nel riquadro di navigazione a sinistra, seleziona Policies (Policy).

  3. Scegli Create Policy (Crea policy).

  4. Nella sezione Policy editor, scegli l'JSONopzione.

  5. Paste (Incolla) clusterAuthPolicy.

  6. Una volta terminata l'aggiunta delle autorizzazioni alla policy, scegli Successivo.

  7. Nella pagina Verifica e crea, digita i valori per Nome policy e Descrizione (facoltativa) per la policy che si sta creando. Rivedi Autorizzazioni definite in questa policy per visualizzare le autorizzazioni concesse dalla policy.

  8. Seleziona Crea policy per salvare la nuova policy.

Per ulteriori informazioni, consulta Creazione IAM di politiche nella IAM documentazione.

Ora che hai IAM le politiche appropriate, crea un ruolo e assegnale ad esso. È possibile farlo utilizzando la console con la seguente procedura.

Per creare un ruolo di esecuzione nella IAM console

  1. Apri la pagina Ruoli nella IAM console.

  2. Scegliere Crea ruolo.

  3. In Tipo di entità attendibile, scegli Servizio AWS .

  4. In Use case (Caso d'uso), scegli Lambda.

  5. Scegli Next (Successivo).

  6. Selezionare le seguenti policy:

    • clusterAuthPolicy

    • AWSLambdaMSKExecutionRole

  7. Scegli Next (Successivo).

  8. Per il nome del ruolo, inserisci lambdaAuthRole e quindi scegli Crea ruolo.

Per ulteriori informazioni, consulta Definizione delle autorizzazioni della funzione Lambda con un ruolo di esecuzione.

Crea una funzione Lambda da leggere dal tuo argomento Amazon MSK

Crea una funzione Lambda configurata per utilizzare il tuo IAM ruolo. Puoi creare la tua funzione Lambda utilizzando la console.

Per creare una funzione Lambda utilizzando la configurazione di autenticazione

  1. Apri la console Lambda e seleziona Crea funzione dall'intestazione.

  2. Scegli Crea da zero.

  3. Per il nome della funzione, fornisci un nome appropriato a tua scelta.

  4. Per Runtime, scegli l'ultima versione supportata di Node.js per utilizzare il codice fornito in questo tutorial.

  5. Scegli Cambia il ruolo di esecuzione predefinito.

  6. Seleziona Usa un ruolo esistente.

  7. Per Ruolo esistente, seleziona lambdaAuthRole.

In un ambiente di produzione, di solito è necessario aggiungere ulteriori policy al ruolo di esecuzione per la funzione Lambda per elaborare in modo significativo gli eventi Amazon. MSK Per ulteriori informazioni sull'aggiunta di politiche al tuo ruolo, consulta Aggiungere o rimuovere le autorizzazioni di identità nella documentazione. IAM

Crea una mappatura della sorgente degli eventi sulla tua funzione Lambda

La mappatura delle sorgenti MSK degli eventi Amazon fornisce al servizio Lambda le informazioni necessarie per richiamare la tua Lambda quando si verificano gli eventi Amazon appropriati. MSK Puoi creare una MSK mappatura Amazon utilizzando la console. Crea un trigger Lambda, quindi la mappatura della sorgente dell'evento viene impostata automaticamente.

Per creare un trigger Lambda (e una mappatura dell'origine degli eventi)

  1. Vai alla pagina di panoramica della tua funzione Lambda.

  2. Nella sezione panoramica delle funzioni, scegli Aggiungi trigger in basso a sinistra.

  3. Nel menu a discesa Seleziona una fonte, seleziona Amazon MSK.

  4. Non impostare l'autenticazione.

  5. Per il MSKcluster, seleziona il nome del cluster.

  6. Per Dimensione del Batch, inserire 1. Questo passaggio semplifica il test di questa funzionalità e non rappresenta un valore ideale in fase di produzione.

  7. Per il nome dell'argomento, inserisci il nome del tuo argomento Kafka.

  8. Per l'ID del gruppo di consumatori, fornisci l'ID del tuo gruppo di consumatori Kafka.

Aggiorna la funzione Lambda per leggere i dati di streaming

Lambda fornisce informazioni sugli eventi di Kafka tramite il parametro event method. Per un esempio di struttura di un MSK evento Amazon, vedi Esempio di evento. Dopo aver capito come interpretare gli eventi MSK Amazon inoltrati da Lambda, puoi modificare il codice della funzione Lambda per utilizzare le informazioni fornite.

Fornisci il seguente codice alla tua funzione Lambda per registrare il contenuto di un MSK evento Amazon Lambda a scopo di test:

.NET
AWS SDK for .NET
Nota

C'è di più su. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumazione di un MSK evento Amazon con Lambda utilizzando. NET.

using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KafkaEvents;


// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]

namespace MSKLambda;

public class Function
{
    
    
    /// <param name="input">The event for the Lambda function handler to process.</param>
    /// <param name="context">The ILambdaContext that provides methods for logging and describing the Lambda environment.</param>
    /// <returns></returns>
    public void FunctionHandler(KafkaEvent evnt, ILambdaContext context)
    {

        foreach (var record in evnt.Records)
        {
            Console.WriteLine("Key:" + record.Key); 
            foreach (var eventRecord in record.Value)
            {
                var valueBytes = eventRecord.Value.ToArray();    
                var valueText = Encoding.UTF8.GetString(valueBytes);
                
                Console.WriteLine("Message:" + valueText);
            }
        }
    }
    

}
Go
SDKper Go V2
Nota

C'è altro da fare. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumare un MSK evento Amazon con Lambda utilizzando Go.


package main

import (
	"encoding/base64"
	"fmt"

	"github.com/aws/aws-lambda-go/events"
	"github.com/aws/aws-lambda-go/lambda"
)

func handler(event events.KafkaEvent) {
	for key, records := range event.Records {
		fmt.Println("Key:", key)

		for _, record := range records {
			fmt.Println("Record:", record)

			decodedValue, _ := base64.StdEncoding.DecodeString(record.Value)
			message := string(decodedValue)
			fmt.Println("Message:", message)
		}
	}
}

func main() {
	lambda.Start(handler)
}
Java
SDKper Java 2.x
Nota

C'è altro su. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumo di un MSK evento Amazon con Lambda utilizzando Java.


import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent.KafkaEventRecord;

import java.util.Base64;
import java.util.Map;

public class Example implements RequestHandler<KafkaEvent, Void> {

    @Override
    public Void handleRequest(KafkaEvent event, Context context) {
        for (Map.Entry<String, java.util.List<KafkaEventRecord>> entry : event.getRecords().entrySet()) {
            String key = entry.getKey();
            System.out.println("Key: " + key);

            for (KafkaEventRecord record : entry.getValue()) {
                System.out.println("Record: " + record);

                byte[] value = Base64.getDecoder().decode(record.getValue());
                String message = new String(value);
                System.out.println("Message: " + message);
            }
        }

        return null;
    }
}
JavaScript
SDKper JavaScript (v3)
Nota

C'è di più su. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumazione di un MSK evento Amazon con Lambda utilizzando. JavaScript


exports.handler = async (event) => {
    // Iterate through keys
    for (let key in event.records) {
      console.log('Key: ', key)
      // Iterate through records
      event.records[key].map((record) => {
        console.log('Record: ', record)
        // Decode base64
        const msg = Buffer.from(record.value, 'base64').toString()
        console.log('Message:', msg)
      }) 
    }
}
PHP
SDK per PHP
Nota

C'è altro da fare. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumazione di un MSK evento Amazon con Lambda utilizzando. PHP

<?php
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0

// using bref/bref and bref/logger for simplicity

use Bref\Context\Context;
use Bref\Event\Kafka\KafkaEvent;
use Bref\Event\Handler as StdHandler;
use Bref\Logger\StderrLogger;

require __DIR__ . '/vendor/autoload.php';

class Handler implements StdHandler
{
    private StderrLogger $logger;
    public function __construct(StderrLogger $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @throws JsonException
     * @throws \Bref\Event\InvalidLambdaEvent
     */
    public function handle(mixed $event, Context $context): void
    {
        $kafkaEvent = new KafkaEvent($event);
        $this->logger->info("Processing records");
        $records = $kafkaEvent->getRecords();

        foreach ($records as $record) {
            try {
                $key = $record->getKey();
                $this->logger->info("Key: $key");

                $values = $record->getValue();
                $this->logger->info(json_encode($values));

                foreach ($values as $value) {
                    $this->logger->info("Value: $value");
                }
                
            } catch (Exception $e) {
                $this->logger->error($e->getMessage());
            }
        }
        $totalRecords = count($records);
        $this->logger->info("Successfully processed $totalRecords records");
    }
}

$logger = new StderrLogger();
return new Handler($logger);
Python
SDKper Python (Boto3)
Nota

C'è di più su. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumo di un MSK evento Amazon con Lambda usando Python.


import base64

def lambda_handler(event, context):
    # Iterate through keys
    for key in event['records']:
        print('Key:', key)
        # Iterate through records
        for record in event['records'][key]:
            print('Record:', record)
            # Decode base64
            msg = base64.b64decode(record['value']).decode('utf-8')
            print('Message:', msg)
Ruby
SDKper Ruby
Nota

c'è altro da fare. GitHub Trova l'esempio completo e scopri come eseguire la configurazione e l'esecuzione nel repository di Esempi serverless.

Consumare un MSK evento Amazon con Lambda utilizzando Ruby.


require 'base64'

def lambda_handler(event:, context:)
  # Iterate through keys
  event['records'].each do |key, records|
    puts "Key: #{key}"

    # Iterate through records
    records.each do |record|
      puts "Record: #{record}"

      # Decode base64
      msg = Base64.decode64(record['value'])
      puts "Message: #{msg}"
    end
  end
end

Puoi fornire il codice della funzione alla tua Lambda utilizzando la console.

Per aggiornare il codice della funzione Lambda

  1. Vai alla pagina di panoramica della tua funzione Lambda.

  2. Scegli la scheda Codice.

  3. Inserisci il codice fornito nel codice sorgenteIDE.

  4. Nella barra di navigazione del codice sorgente, scegli Deploy.

Testa la tua funzione Lambda per verificare che sia collegata al tuo argomento Amazon MSK

Ora puoi verificare se la tua Lambda viene richiamata o meno dall'origine dell'evento CloudWatch controllando i registri degli eventi.

Per verificare se la funzione Lambda viene richiamata

  1. Usa il tuo host di amministrazione Kafka per generare eventi Kafka utilizzando. kafka-console-producer CLI Per maggiori informazioni, consulta Scrivere alcuni eventi nell'argomento della documentazione di Kafka. Invia un numero sufficiente di eventi per riempire il batch definito dalla dimensione del batch per la mappatura dell'origine degli eventi definita nel passaggio precedente, altrimenti Lambda aspetterà che vengano richiamate ulteriori informazioni.

  2. Se la funzione viene eseguita, Lambda scrive cosa è successo a. CloudWatch Nella console, vai alla pagina dei dettagli della tua funzione Lambda.

  3. Selezionare la scheda Configurazione.

  4. Dalla barra laterale, seleziona Strumenti operativi e di monitoraggio.

  5. Identifica il gruppo di CloudWatch log in Configurazione della registrazione. Il gruppo di log dovrebbe iniziare con/aws/lambda. Scegli il link al gruppo di log.

  6. Nella CloudWatch console, controlla gli eventi di registro per gli eventi di registro che Lambda ha inviato al flusso di log. Identifica se ci sono eventi di registro contenenti il messaggio del tuo evento Kafka, come nell'immagine seguente. In tal caso, hai collegato correttamente una funzione Lambda ad Amazon MSK con una mappatura della sorgente degli eventi Lambda.

    Un evento di registro CloudWatch che mostra le informazioni sugli eventi estratte dal codice fornito.