In che modo Lambda elabora i record di Amazon Kinesis Data Streams

È possibile utilizzare una funzione Lambda per elaborare i record in un flusso di dati Amazon Kinesis. È possibile mappare una funzione Lambda a un consumer a throughput condiviso di Kinesis Data Streams (iteratore standard) o a un consumer a throughput dedicato con fan-out migliorato. Per gli iteratori standard, Lambda analizza ogni shard nel flusso Kinesis alla ricerca di record utilizzando il protocollo. HTTP La mappatura dell'origine eventi condivide il throughput di lettura con altri utenti della partizione.

Per informazioni dettagliate sui flussi di dati Kinesis, consulta Lettura dei dati da Amazon Kinesis Data Streams.

Nota

Kinesis addebita dei costi per ogni partizione, per il fan-out avanzato e per i dati letti dal flusso. Per i dettagli sui prezzi, consulta Prezzi di Amazon Kinesis.

Argomenti

• Flussi di polling e batching
• Esempio di evento
• Elaborazione dei record di Amazon Kinesis Data Streams con Lambda
• Configurazione della risposta batch parziale con Kinesis Data Streams e Lambda
• Conserva i record batch scartati per un'origine di eventi Kinesis Data Streams in Lambda
• Implementazione dell'elaborazione Kinesis Data Streams con stato in Lambda
• Parametri Lambda per le mappature delle sorgenti degli eventi di Amazon Kinesis Data Streams
• Utilizzo del filtraggio degli eventi con una sorgente di eventi Kinesis
• Tutorial: Utilizzo di Lambda con Kinesis Data Streams

Flussi di polling e batching

Lambda legge i record dal flusso di dati e richiama la funzione in modo sincrono con un evento che contiene record di flusso. Lambda legge i record in batch e richiama la funzione per elaborare i record dal batch. Ogni batch contiene registri da una singola partizione/flusso dei dati.

Per i flussi di dati Kinesis standard, Lambda analizza i frammenti del flusso alla ricerca di record a una velocità di una volta al secondo per ogni shard. Per il fan-out avanzato di Kinesis, Lambda utilizza HTTP una connessione /2 per ascoltare i record inviati da Kinesis. Quando sono disponibili dei record, Lambda invoca la funzione e attende il risultato.

Per impostazione predefinita, Lambda richiama la funzione non appena i record sono disponibili. Se il batch che Lambda legge dall'origine eventi contiene un solo record, Lambda invia solo un record alla funzione. Per evitare di richiamare la funzione con pochi record è possibile, configurando un periodo di batch, chiedere all'origine eventi di memorizzare nel buffer i registri per un massimo di 5 minuti. Prima di richiamare la funzione, Lambda continua a leggere i registri dall'origine eventi fino a quando non ha raccolto un batch completo, fino alla scadenza del periodo di batch o fino a quando il batch non ha raggiunto il limite del payload di 6 MB. Per ulteriori informazioni, consulta Comportamento di batching.

avvertimento

Le mappature delle sorgenti degli eventi Lambda elaborano ogni evento almeno una volta e può verificarsi un'elaborazione duplicata dei record. Per evitare potenziali problemi legati agli eventi duplicati, ti consigliamo vivamente di rendere idempotente il codice della funzione. Per ulteriori informazioni, consulta Come posso rendere idempotente la mia funzione Lambda nel Knowledge Center. AWS

Lambda non attende il completamento delle estensioni configurate prima di inviare il batch successivo per l'elaborazione. In altre parole, le estensioni possono continuare a funzionare mentre Lambda elabora il successivo batch di record. Ciò può causare problemi di limitazione in caso di violazione delle impostazioni o dei limiti di simultaneità. Per rilevare se si tratta di un potenziale problema, monitora le tue funzioni e verifica se i parametri di simultaneità per lo strumento di mappatura dell'origine degli eventi sono superiori al previsto. A causa degli intervalli ridotti tra le invocazioni, Lambda potrebbe segnalare brevemente un utilizzo della simultaneità maggiore rispetto al numero di partizioni. Ciò può essere vero anche per le funzioni Lambda senza estensioni.

Configura l'ParallelizationFactorimpostazione per elaborare uno shard di un flusso di dati Kinesis con più di una chiamata Lambda contemporaneamente. È possibile specificare il numero di batch simultanei di cui Lambda esegue il polling da una partizione da un fattore di parallelizzazione compreso tra da 1 (predefinito) e 10. Ad esempio, se si imposta su ParallelizationFactor 2, è possibile avere un massimo di 200 chiamate Lambda simultanee per elaborare 100 shard di dati Kinesis (anche se in pratica, è possibile visualizzare valori diversi per la metrica). ConcurrentExecutions Ciò permette di dimensionare verso l'alto il throughput di elaborazione quando il volume dei dati è volatile e l'IteratorAge è alta. Quando si aumenta il numero di batch simultanei per shard, Lambda garantisce comunque l'elaborazione in ordine a livello di chiave di partizione.

È possibile utilizzarlo anche ParallelizationFactor con l'aggregazione Kinesis. Il comportamento della mappatura delle sorgenti degli eventi dipende dall'utilizzo o meno di un fan-out avanzato:

• Senza fan-out avanzato: tutti gli eventi all'interno di un evento aggregato devono avere la stessa chiave di partizione. La chiave di partizione deve inoltre corrispondere a quella dell'evento aggregato. Se gli eventi all'interno dell'evento aggregato hanno chiavi di partizione diverse, Lambda non può garantire l'elaborazione ordinata degli eventi per chiave di partizione.

• Con fan-out migliorato: in primo luogo, Lambda decodifica l'evento aggregato nei suoi singoli eventi. L'evento aggregato può avere una chiave di partizione diversa dagli eventi che contiene. Tuttavia, gli eventi che non corrispondono alla chiave di partizione vengono eliminati e persi. Lambda non elabora questi eventi e non li invia a una destinazione di errore configurata.

Esempio di evento

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