Scopri come Lambda elabora i record di Flusso di dati Amazon Kinesis.

È possibile usare 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 (iteratore standard) del flusso di dati Kinesis o a un consumer di throughput dedicato con fan-out avanzato. Per gli iteratori standard, Lambda esegue il polling di ogni partizione nel flusso Kinesis per i 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 Flusso di dati Amazon Kinesis.

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 singolo shard/flusso dei dati.

Per i flussi di dati Kinesis standard, Lambda esegue il polling per ogni shard nel flusso per i record a una velocità di base di una volta al secondo per ogni shard. Per il fan-out avanzato di Kinesis, Lambda utilizza una connessione HTTP/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.

Lambda non attende il completamento di alcuna estensione 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 uno shard da un fattore di parallelizzazione compreso tra da 1 (predefinito) e 10. Ad esempio, impostando ParallelizationFactor su 2, puoi disporre al massimo di 200 invocazioni Lambda simultanee per elaborare 100 partizioni di dati Kinesis (anche se nella pratica potresti vedere valori differenti per il parametro ConcurrentExecutions). Ciò permette di dimensionare verso l'alto il throughput di elaborazione quando il volume dei dati è volatile e l'IteratorAge è alta. Se si aumenta il numero di batch simultanei per shard, Lambda garantisce comunque l'ordine di elaborazione a livello di partizione-chiave.

Puoi anche utilizzare ParallelizationFactor con l'aggregazione Kinesis. Il comportamento dello strumento di mappatura dell'origine degli eventi dipende dall'utilizzo o meno di un fan-out avanzato:

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