Transformación de datos en Amazon Data Firehose

Amazon Data Firehose puede invocar la función Lambda para transformar los datos de origen entrantes y entregar los datos transformados a los destinos. Puede habilitar la transformación de datos de Amazon Data Firehose al crear su transmisión de Firehose.

Flujo de transformación de datos

Al activar la transformación de datos de Firehose, Firehose almacena en búfer los datos entrantes. La sugerencia sobre el tamaño del búfer oscila entre 0,2 MB y 3 MB. La sugerencia de tamaño de búfer Lambda predeterminada es de 1 MB para todos los destinos, excepto Splunk y Snowflake. Para Splunk y Snowflake, la sugerencia de almacenamiento en búfer predeterminada es de 256 KB. La sugerencia del intervalo de almacenamiento en búfer de Lambda oscila entre 0 y 900 segundos. La sugerencia de intervalo de almacenamiento en búfer de Lambda predeterminada es de sesenta segundos para todos los destinos excepto Snowflake. Para Snowflake, el intervalo de sugerencia de almacenamiento en búfer predeterminado es de 30 segundos. Para ajustar el tamaño del búfer, defina el ProcessingConfigurationparámetro de la API CreateDeliveryStreamo UpdateDestinationcon la llamada y. ProcessorParameterBufferSizeInMBsIntervalInSeconds A continuación, Firehose invoca la función Lambda especificada de forma asíncrona con cada lote almacenado en búfer mediante el modo de invocación sincrónica. AWS Lambda Los datos transformados se envían de Lambda a Firehose. A continuación, Firehose lo envía al destino cuando se alcanza el tamaño o el intervalo de almacenamiento en búfer de destino especificados, lo que ocurra primero.

importante

El modo de invocación sincrónica de Lambda tiene un límite de tamaño de carga de 6 MB para la solicitud y la respuesta. Asegúrese de que el tamaño de almacenamiento en búfer para enviar la solicitud a la función es inferior o igual a 6 MB. Asegúrese también de que la respuesta devuelta por la función no sea superior a 6 MB.

Modelo de estados y transformación de datos

Todos los registros transformados de Lambda deben contener los siguientes parámetros; de lo contrario, Amazon Data Firehose los rechazará y considerará que se trata de un error de transformación de datos.

En el caso de Kinesis Data Streams y Direct PUT:

recordId

El ID de registro se pasa de Amazon Data Firehose a Lambda durante la invocación. El registro transformado debe contener el mismo ID de registro. Cualquier discrepancia entre el ID del registro original y el del transformado se trata como un error de transformación de datos.

resultado

Es el estado de la transformación de los datos del registro. Los valores posibles son Ok si el registro se ha transformado correctamente, Dropped si la lógica de procesamiento ha omitido el registro intencionadamente y ProcessingFailed si el registro no se ha podido transformar. Si un registro tiene el estado Ok oDropped, Amazon Data Firehose considera que se ha procesado correctamente. De lo contrario, Amazon Data Firehose considerará que no se ha procesado correctamente.

datos

Es la carga útil de datos transformados después codificarlos en base64.

A continuación se presenta un ejemplo de salida de Lambda:

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

En el caso de Amazon MSK:

recordId

El ID del registro se pasa de Firehose a Lambda durante la invocación. El registro transformado debe contener el mismo ID de registro. Cualquier discrepancia entre el ID del registro original y el del transformado se trata como un error de transformación de datos.

resultado

Es el estado de la transformación de los datos del registro. Los valores posibles son Ok si el registro se ha transformado correctamente, Dropped si la lógica de procesamiento ha omitido el registro intencionadamente y ProcessingFailed si el registro no se ha podido transformar. Si un registro tiene el estado Ok oDropped, Firehose considera que se ha procesado correctamente. De lo contrario, Firehose considerará que no se ha procesado correctamente.

KafkaRecordValue

Es la carga útil de datos transformados después codificarlos en base64.

A continuación se presenta un ejemplo de salida de Lambda:

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

Esquemas de Lambda

Estos planos muestran cómo puede crear y utilizar las funciones de AWS Lambda para transformar los datos de las transmisiones de datos de Amazon Data Firehose.

Para ver los planos que están disponibles en la consola AWS Lambda

1. Inicie sesión en la AWS Lambda consola AWS Management Console y ábrala en https://console.aws.amazon.com/lambda/.

2. Elija Create function (Crear función) y, a continuación, elija Use a blueprint (Utilizar un proyecto).

3. En el campo Blueprints, busque la palabra clave firehose para encontrar los blueprints de Amazon Data Firehose Lambda.

Lista de esquemas:

• Registros de procesos enviados a Amazon Data Firehose stream (Node.js, Python)

  Este plano muestra un ejemplo básico de cómo procesar los datos del flujo de datos de Firehose con AWS Lambda.

  Fecha de lanzamiento más reciente: noviembre de 2016.

  Notas de la versión: ninguna.

• CloudWatch Registros del proceso enviados a Firehose

  Este modelo está obsoleto. Para obtener información sobre el procesamiento de CloudWatch los registros enviados a Firehose, consulte Cómo escribir en Firehose mediante registros. CloudWatch

• Convierta los registros de transmisión de Amazon Data Firehose en formato syslog a JSON (Node.js)

  En este esquema se muestra cómo convertir los registros de entrada en formato RFC3164 de Syslog en JSON.

  Fecha de lanzamiento más reciente: noviembre de 2016.

  Notas de la versión: ninguna.

Para ver los planos que están disponibles en el AWS Serverless Application Repository

1. Vaya a AWS Serverless Application Repository.

2. Elija Examinar todas las aplicaciones.

3. En el campo Applications (Aplicaciones) busque la palabra clave firehose.

También puede crear una función de Lambda sin utilizar un esquema. Consulte Introducción a AWS Lambda.

Gestión de errores de transformación de datos

Si se produce un error en la invocación de la función Lambda porque se ha agotado el tiempo de espera de la red o porque se ha alcanzado el límite de invocación de Lambda, Amazon Data Firehose vuelve a intentar la invocación tres veces de forma predeterminada. Si la invocación no se realiza correctamente, Amazon Data Firehose omite ese lote de registros. los trata como registros que no se han podido procesar. Puede especificar o anular las opciones de reintento mediante la API o. CreateDeliveryStreamUpdateDestination Para este tipo de error, puede registrar los errores de invocación en Amazon CloudWatch Logs. Para obtener más información, consulte Supervisión de Amazon Data Firehose mediante registros CloudWatch .

Si el estado de la transformación de datos de un registro esProcessingFailed, Amazon Data Firehose considera que el registro no se ha procesado correctamente. Para este tipo de error, puede enviar registros de errores a Amazon CloudWatch Logs desde su función Lambda. Para obtener más información, consulte Acceder a Amazon CloudWatch Logs AWS Lambda en la Guía para AWS Lambda desarrolladores.

Si ocurre un error durante la transformación de datos, los registros que no se hayan podido procesar se entregan en el bucket de S3, en la carpeta processing-failed. Los registros tienen el siguiente formato:

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

attemptsMade

La cantidad de intentos de solicitud de invocación.

arrivalTimestamp

La hora a la que Amazon Data Firehose recibió el registro.

errorCode

Código de error HTTP devuelto por Lambda.

errorMessage

Mensaje de error HTTP devuelto Lambda.

attemptEndingTimestamp

La hora en que Amazon Data Firehose dejó de intentar invocar a Lambda.

rawData

Los datos de registros codificados en base64.

lambdaArn

El nombre de recurso de Amazon (ARN) de la función de Lambda.

Duración de una invocación de Lambda

Amazon Data Firehose admite un tiempo de invocación a Lambda de hasta 5 minutos. Si la función Lambda tarda más de 5 minutos en completarse, aparece el siguiente error: Firehose detectó errores de tiempo de espera al llamar a Lambda. AWS The maximum supported function timeout is 5 minutes.

Para obtener información sobre lo que hace Amazon Data Firehose si se produce un error de este tipo, consulte. Gestión de errores de transformación de datos

Backup de registros de origen

Amazon Data Firehose puede realizar copias de seguridad de todos los registros no transformados en su bucket de S3 de forma simultánea y, al mismo tiempo, entregar los registros transformados al destino. Puedes habilitar la copia de seguridad de los registros de origen al crear o actualizar tu transmisión de Firehose. El backup de los registros de origen no se puede deshabilitar después de haberlo habilitado.