Cómo usar AWS Kinesis Firehose para almacenar datos en S3 y Amazon Athena para visualizarlos

Para procesamiento de datos en tiempo real, AWS ofrece una serie de servicios de acuerdo a la necesidad específica. En este caso vamos a ver un ejemplo de uso de AWS Kinesis Firehose. Este servicio nos permite recibir datos y almacenarlos en tiempos cortos en otros servicios de Amazon como por ejemplo S3, Redshift, entre otros. Para el ejemplo vamos a enviar datos a AWS Kinesis Firehose a través del CLI y vamos a almacenarlos en un archivo en un bucket de S3, luego vamos a usar Amazon Athena, el cual es un servicio que nos permite ejecutar sentencias SQL sobre datos almacenados en S3.

Para el ejemplo vamos a crear los siguientes recursos:

• Un delivery stream en AWS Kinesis Firehose para recibir los datos.
• Un a función lambda para modificar los datos antes de guardarlos en el bucket de S3.
• Un bucket de S3 donde se almacenarán los datos del delivery stream de Kinesis.
• Una tabla en Athena para consultar los datos del bucket de S3.

A continuación veremos la arquitectura deseada para que quede mas claro la forma como se procesarán y consultarán los datos.

Crear delivery stream de Kinesis Firehose

Para la creación del stream, vamos a buscar el servicio de Kinesis y seguir los siguientes pasos:

Seleccionamos Kinesis Data Firehose.

Agregamos un nombre para el stream.

Seleccionamos la opción “direct PUT” para poder enviar datos a Kinesis desde el CLI.

Habilitamos la opción de transformar datos y seleccionamos crear nueva función lambda.

Seleccionamos el Lambda blueprint de “General Kinesis Data Firehose Processing”. Con esta función vamos a almacenar los datos tal cual los recibimos. Más adelante modificamos la función para almacenar los datos como se requiere.

Agregamos el nombre de la función y la creamos.

Luego de crear la función, la seleccionamos en la creación del stream.

Seleccionamos “Amazon S3” como el destino de nuestros datos, y podemos crear un nuevo bucket o seleccionar uno ya existente.

Dejamos las otras opciones por defecto y creamos el stream.

Esperamos que el estado esté en activo para continuar.

En este momento ya podemos comenzar a enviar datos pero debemos modificar la lambda de transformación para que los datos nos queden en el formato que requerimos. Para esto vamos a la consola de Lambda y buscamos la función que creamos. Vamos a modificar el código para adicionar un salto de línea con cada dato que enviemos.

Envío de datos a Kinesis Firehose a través del CLI

Para realizar el envío de datos vamos a usar el CLI de AWS. Debemos tenerlo configurado con los datos de nuestro usuario con el comando “aws configure”. Una vez teniendo esto configurado vamos a usar el comando aws firehose put-record para enviar los datos a nuestro stream. En el ejemplo vamos a enviar unos datos en formato json.

Una vez enviados los datos, vamos a revisar el bucket de S3 que configuramos para buscar los datos que Kinesis Firehose recibió. Este proceso puede tardar unos minutos.

Al abrir el archivo podemos ver los datos que enviamos.

Consulta de los datos a través de Amazon Athena

Por último vamos a configurar Athena para poder consultar los datos que hemos guardado en el bucket. Buscamos el servicio de Athena e ingresamos a configurarlo.

Vamos a “settings” y configuramos la ubicación donde quedarán los resultados de las consultas que hagamos. Podemos usar el mismo bucket donde almacenamos los datos de Kinesis Firehose.

Ahora vamos a crear la conexión con los datos que vamos a consultar, por lo cual seleccionamos “connect data source” y luego consultar datos en Amazon S3.

Seleccionamos la opción para ingresar los datos de una tabla manualmente.

Ingresamos el nombre de la base de datos, de la tabla y la ubicación donde están los datos que queremos consultar.

Luego seleccionamos el formato de los datos, en este caso JSON.

Por último ingresamos el nombre de las columnas de nuestros datos y finalizamos.

Ahora podemos consultar los datos de nuestra tabla con comandos SQL.

Conclusiones:

Con AWS Kinesis podemos recibir datos en tiempo real y enviarlo a múltiples destinos. De ahí se abren muchas posibilidades para trabajar con estos datos e integrarlos con múltiples servicios de AWS. Depende de nuestra necesidad cual de los servicios vamos a usar.