Trasforma i dati in Amazon Data Firehose

Amazon Data Firehose può richiamare la funzione Lambda per trasformare i dati di origine in entrata e consegnarli alle destinazioni. Puoi abilitare la trasformazione dei dati di Amazon Data Firehose quando crei il tuo stream Firehose.

Comprendi il flusso di trasformazione dei dati

Quando si abilita la trasformazione dei dati Firehose, Firehose memorizza nel buffer i dati in entrata. Il suggerimento sulla dimensione del buffering varia tra 0,2 MB e 3 MB. Il suggerimento predefinito sulla dimensione del buffering Lambda è 1 MB per tutte le destinazioni, tranne Splunk e Snowflake. Per Splunk e Snowflake, l'hint di buffering predefinito è 256 KB. Il suggerimento sull'intervallo di buffering Lambda è compreso tra 0 e 900 secondi. Il suggerimento predefinito per l'intervallo di buffering Lambda è di sessanta secondi per tutte le destinazioni tranne Snowflake. Per Snowflake, l'intervallo di suggerimento di buffering predefinito è di 30 secondi. Per regolare la dimensione del buffering, imposta il ProcessingConfigurationparametro o con il comando e chiamato. CreateDeliveryStreamUpdateDestinationAPIProcessorParameterBufferSizeInMBsIntervalInSeconds Firehose richiama quindi la funzione Lambda specificata in modo asincrono con ogni batch bufferizzato utilizzando la modalità di chiamata sincrona. AWS Lambda I dati trasformati vengono inviati da Lambda a Firehose. Firehose lo invia quindi alla destinazione quando viene raggiunta la dimensione o l'intervallo di buffering di destinazione specificati, a seconda dell'evento che si verifica per primo.

Importante

La modalità di invocazione sincrona di Lambda ha un limite di dimensione del payload di 6 MB sia per la richiesta che per la risposta. Verifica che la dimensione del buffer per l'invio della richiesta alla funzione sia minore o uguale a 6 MB. Verifica anche che la risposta restituita dalla funzione non superi i 6 MB.

Modello di trasformazione e stato dei dati

Tutti i record trasformati da Lambda devono contenere i seguenti parametri, altrimenti Amazon Data Firehose li rifiuta e li considera un errore di trasformazione dei dati.

Per Kinesis Data PUT Streams e Direct:

recordId

L'ID del record viene passato da Amazon Data Firehose a Lambda durante la chiamata. Il record trasformato deve contenere lo stesso ID record. Ogni mancata corrispondenza tra l'ID del record originale e l'ID del record trasformato viene trattato come un errore di trasformazione dei dati.

result

Stato della trasformazione dei dati del record. I valori possibili sono: Ok (il record è stato trasformato correttamente), Dropped (il record è stato rimosso intenzionalmente dalla logica di elaborazione) e ProcessingFailed (non è stato possibile trasformare il record). Se un record ha lo stato di Ok oDropped, Amazon Data Firehose lo considera elaborato correttamente. In caso contrario, Amazon Data Firehose lo considera elaborato senza successo.

dati

Il payload dei dati trasformati dopo la codifica base64.

Di seguito è riportato un esempio di risultato Lambda:

 {
    "recordId": "<recordId from the Lambda input>",
    "result": "Ok",
    "data": "<Base64 encoded Transformed data>"
}

Per Amazon MSK

recordId

L'ID del record viene passato da Firehose a Lambda durante la chiamata. Il record trasformato deve contenere lo stesso ID record. Ogni mancata corrispondenza tra l'ID del record originale e l'ID del record trasformato viene trattato come un errore di trasformazione dei dati.

result

Stato della trasformazione dei dati del record. I valori possibili sono: Ok (il record è stato trasformato correttamente), Dropped (il record è stato rimosso intenzionalmente dalla logica di elaborazione) e ProcessingFailed (non è stato possibile trasformare il record). Se un record ha lo stato Ok oDropped, Firehose lo considera elaborato correttamente. In caso contrario, Firehose lo considera elaborato senza successo.

KafkaRecordValue

Il payload dei dati trasformati dopo la codifica base64.

Di seguito è riportato un esempio di risultato Lambda:

 {
    "recordId": "<recordId from the Lambda input>",
    "result": "Ok",
    "kafkaRecordValue": "<Base64 encoded Transformed data>"
}

Lavora con i progetti Lambda

Questi blueprint dimostrano come è possibile creare e utilizzare funzioni AWS Lambda per trasformare i dati nei flussi di dati di Amazon Data Firehose.

Per vedere i blueprint disponibili nella console AWS Lambda

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

2. Scegliere Create function (Crea funzione), quindi scegliere Use a blueprint (Usa un piano).

3. Nel campo Blueprints, cerca la parola chiave per firehose trovare i blueprint Amazon Data Firehose Lambda.

Elenco degli schemi:

• Record di processo inviati allo stream Amazon Data Firehose (Node.js, Python)

  Questo modello mostra un esempio di base di come elaborare i dati nel flusso di dati di Firehose utilizzando AWS Lambda.

  Ultima data di rilascio: novembre 2016.

  Note di rilascio: nessuna.

• CloudWatch Registri dei processi inviati a Firehose

  Questo blueprint è obsoleto. Per informazioni sull'elaborazione dei CloudWatch log inviati a Firehose, vedere Writing to Firehose Using Logs. CloudWatch

• Convertire i record di flusso di Amazon Data Firehose in formato syslog in (Node.js) JSON

  Questo modello mostra come convertire i record di input in formato Syslog inRFC3164. JSON

  Ultima data di rilascio: novembre 2016.

  Note di rilascio: nessuna.

Per visualizzare i progetti disponibili in AWS Serverless Application Repository

1. Passa a AWS Serverless Application Repository.

2. Scegli Sfoglia tutte le applicazioni.

3. Nel campo Applications (Applicazioni), cercare la parola chiave firehose.

Puoi inoltre creare una funzione Lambda senza utilizzare uno schema. Vedi Guida introduttiva a AWS Lambda.

Gestione degli errori di trasformazione dei dati

Se la chiamata della funzione Lambda non riesce a causa di un timeout di rete o perché hai raggiunto il limite di invocazione Lambda, Amazon Data Firehose ritenta la chiamata tre volte per impostazione predefinita. Se la chiamata non riesce, Amazon Data Firehose salta quel batch di record. I record ignorati vengono considerati record con errori di elaborazione. È possibile specificare o ignorare le opzioni di riprova utilizzando o. CreateDeliveryStreamUpdateDestinationAPI Per questo tipo di errore, puoi registrare gli errori di chiamata su Amazon CloudWatch Logs. Per ulteriori informazioni, consulta Monitoraggio di Amazon Data Firehose tramite log CloudWatch .

Se lo stato della trasformazione dei dati di un record èProcessingFailed, Amazon Data Firehose considera il record come se fosse stato elaborato senza successo. Per questo tipo di errore, puoi inviare log di errore ad Amazon CloudWatch Logs dalla tua funzione Lambda. Per ulteriori informazioni, consulta Accedere ad Amazon CloudWatch Logs AWS Lambda nella AWS Lambda Developer Guide.

Se la trasformazione dei dati non va a buon fine, i record elaborati non correttamente vengono distribuiti sul bucket S3 nella cartella processing-failed. I record hanno il formato seguente:

{
    "attemptsMade": "count",
    "arrivalTimestamp": "timestamp",
    "errorCode": "code",
    "errorMessage": "message",
    "attemptEndingTimestamp": "timestamp",
    "rawData": "data",
    "lambdaArn": "arn"
}

attemptsMade

Il numero di richieste di invocazioni tentate.

arrivalTimestamp

L'ora in cui il record è stato ricevuto da Amazon Data Firehose.

errorCode

Il codice HTTP di errore restituito da Lambda.

errorMessage

Il messaggio di errore restituito da Lambda.

attemptEndingTimestamp

L'ora in cui Amazon Data Firehose ha smesso di tentare le chiamate Lambda.

rawData

I dati dei record codificati con base 64.

lambdaArn

Amazon Resource Name (ARN) della funzione Lambda.

Comprendi la durata delle chiamate Lambda

Amazon Data Firehose supporta un tempo di chiamata Lambda fino a 5 minuti. Se la funzione Lambda impiega più di 5 minuti per essere completata, viene visualizzato il seguente errore: Firehose ha riscontrato errori di timeout durante la chiamata a Lambda. AWS Il timeout massimo della funzione supportato è di 5 minuti.

Per informazioni su cosa fa Amazon Data Firehose se si verifica un errore di questo tipo, consulta. Gestione degli errori di trasformazione dei dati

Backup dei record di origine

Amazon Data Firehose è in grado di eseguire contemporaneamente il backup di tutti i record non trasformati nel bucket S3 e di consegnare i record trasformati alla destinazione. È possibile abilitare il backup dei record di origine quando si crea o si aggiorna lo stream Firehose. Non puoi disabilitare il backup dei record di origine dopo averlo abilitato.