- Saltar a la navegación principal
- Saltar al contenido principal
- Saltar a la barra lateral principal
- Saltar al pie de página
Bluetab
an IBM Company
LakeHouse Streaming en AWS con Apache Flink y Hudi
Introducción
Cada día la ingesta y procesamiento de streams de datos en Near Real Time (NRT) es más necesario. Los requisitos de negocio son cada vez más exigentes en cuanto a tiempos de procesamiento y la disponibilidad de los datos más recientes y este artículo pretende abordar esta cuestión.
Utilizando la nube de AWS y con un enfoque serverless se desplegará en este artículo una aplicación capaz de ingestar streams de datos y procesarlos en NRT, escribiendo su resultado en un LakeHouse de tal manera que se puedan realizar operaciones ACID (Atomicidad, Consistencia, Aislamiento y Durabilidad) sobre estos. Se desplegará una arquitectura en la que se ingestan datos con Locust, se procesan con Flink y finalmente se escriben en Hudi y JSON.
Locust es un framework de Python que sirve para poder realizar Load Testing de una manera fácil y escalable. Las ventajas que ofrece Locust son la capacidad de poder definir este comportamiento de los usuarios con un lenguaje de propósito general y su facilidad de escalado.
Flink se ha convertido en un framework de referencia en el ámbito de procesamiento distribuido sobre streams de datos. Se caracteriza por su orientación al procesamiento de streams (aunque también puede ejecutar procesos batch), su rapidez de procesamiento y su eficiencia en el uso de memoria. Hay otros frameworks populares en el sector, como Spark Streaming y Storm, en el apartado de arquitectura se discutirá por qué en última instancia Flink ha sido el elegido.
Finalmente, Hudi es un formato de fichero transaccional que proporciona las habilidades propias de una base de datos y DataWarehouse al Data Lake. Hudi da la capacidad de dejar atrás los conceptos de batching y sustituirlo con una perspectiva de procesamiento incremental. Como el resto de las tecnologías usadas en este artículo, se describe en detalle más adelante.
Todo el código utilizado en este artículo, tanto IaC como de Python, puede visitarse en nuestro repositorio[1] en Github.
En próximos artículos
Múltiples artículos utilizarán este como base para hablar de los siguientes temas:
- Comparativa en cuanto a eficiencia de procesamiento, escritura y lectura de ficheros y costes en JSON vs Hudi.
- Comparativa de MOR vs COW, además del consumo de estas tablas por los distintos tipos de queries (Snapshot, Read Optimized, Incremental).
- Escalabilidad.
- Otras formas de explotación del dato, como pueden ser Redshift o Pinot.
Arquitectura
A continuación se puede ver la arquitectura a alto nivel que se desplegará:
Como se puede ver, se utiliza Locust como herramienta de Load Testing para enviar datos sintéticos a nuestra aplicación. Estos serán ingestados a través de un Kinesis Stream aprovisionado en modo On Demand, de esta manera el stream escalará de manera automática. La alternativa al modo On Demand es el modo Provisioned, donde debemos especificar el número de shards (componente en los que se divide el stream), con el que queremos aprovisionar el stream. Las diferencias y particularidades de estos dos modos se explicarán más en detalle en el apartado de Kinesis.
Del stream de entrada leen las dos aplicaciones de Kinesis Analytics Flink. Como se mencionó en el apartado de próximos pasos, la razón de tener dos aplicaciones independientes escribiendo en Hudi y JSON respectivamente es para realizar una comparativa en próximos artículos en cuanto a eficiencia. Finalmente los datos se alojarán en S3, el servicio de almacenamiento de objetos de AWS.
La particularidad que tiene la aplicación de Kinesis Analytics Flink es que es serverless, es decir, abstrae al desarrollador de la complejidad de configurar y desplegar un cluster con Flink. A esta aplicación se deben asignar unos KPUs o Kinesis Processing Units y un jar con la librería de Flink y los conectores necesarios para poder desplegarla correctamente. Todos estos conceptos serán explicados en los siguientes apartados.
La alternativa a esta perspectiva serverless con un servicio administrado en AWS es la administración completa de la aplicación por parte del desarrollador, pudiendo utilizar herramientas como Kubernetes o EKS (Kubernetes administrado en AWS) para poder desplegar en un cluster esta aplicación Flink. Las ventajas de esta segunda alternativa sería el poder configurar tanto el cluster (número de nodos, memoria, CPU, disco duro, etc…) como la aplicación Flink (gestión de disaster recovery, gestión de metadatos, etc…) con un grado de detalle mucho mayor. En este artículo se decidió la primera alternativa por su simplicidad y facilidad de uso a la hora de conocer el framework de Flink.
Locust
La primera pieza en la pipeline de ingesta de datos es el componente de Locust escrito en Python. A diferencia de otros frameworks disponibles en el mercado como JMeter, Locust nos da la capacidad de poder escribir un código simple con Python en vez de utilizar un lenguaje específico a un dominio o una interfaz de usuario.
Además, Locust está basado en eventos y utiliza greenlet[2], lo que le da la capacidad de con un solo hilo del procesador poder administrar la capacidad de varios miles de usuarios. Por ejemplo, en el caso de JMeter, se necesita un hilo para cada usuario, lo que supone un problema de escalabilidad para casos en los que se necesite un número alto de estos.
Locust tiene varias posibilidades a la hora de ejecutarse y escalar, pudiendo funcionar en local para aplicaciones con menos exigencias en cuanto a volumen de datos o desplegar en un cluster de Kubernetes al crear una imagen de Docker a raíz del código de Locust.
En cuanto a clientes y sistemas a los que enviar datos, Locust proporciona un cliente HTTP integrado. En el caso de querer enviar eventos a otros sistemas, como el de este artículo, siempre se puede escribir un cliente personalizado gracias a la ventaja de ser un framework de Python.
Además, Locust también proporciona una interfaz web para poder comprobar el progreso de tu envío de datos en tiempo real. Por todas estas razones se ha decidido utilizar esta tecnología en este artículo.
Kinesis Data Analytics
Para la ingesta de datos, se utilizará Kinesis Data Streams, un servicio de streaming de datos completamente administrado y serverless ofrecido por AWS. Un Kinesis Stream está formado por una agrupación lógica de shards, que representan la unidad fundamental de capacidad de un stream y son procesados en paralelo. Cada shard dota al stream de 1 MB/s o 1,000 eventos por segundo de escritura y 2 MB/s de lectura. Los eventos serán distribuidos entre los shards de un stream en función de su clave de partición, por lo que es importante que el particionado sea homogéneo para evitar un sesgo en la distribución y ocurrencia de hot shards. Existen dos modos de aprovisionamiento de capacidad:
- On Demand – el número de shards se gestiona automáticamente para acomodar la carga, asegurando un rendimiento óptimo sin necesidad de ajustes manuales.
- Provisioned – debes especificar el número de shards para el stream en función de la carga esperada.
Por simplicidad, y por ser idóneo para nuestro caso de uso, se optará por el modo On Demand. Esto acomodará automáticamente el número de shards a la cantidad de datos generados por nuestra aplicación de Locust.
Para leer y procesar los datos ingestados a través de Kinesis Data Streams, se usará otro servicio de la familia Kinesis, Kinesis Data Analytics (KDA). Este servicio es ofrecido en dos sabores
- Kinesis Analytics SQL – Permite la creación de aplicaciones de procesamiento de datos en streaming mediante el uso de SQL. Este servicio se considera deprecado en favor del servicio de KDA for Apache Flink.
- Kinesis Analytics for Apache Flink – Proporciona una forma de desplegar un cluster de Flink gestionado por AWS. El uso de Flink faculta la creación de aplicaciones más avanzadas y con mayor rendimiento.
Una aplicación de Flink consta de una serie de tareas de procesado en paralelo, también conocidas como operadores, que se conectan en una Directed Acyclic Graph (DAG). El stream de datos es procesado por esta DAG, con cada operador ejecutando una operación específica sobre el dato.
KDA asigna potencia de computación para nuestra aplicación en forma de Kinesis Processing (KPUs), cada una de ellas equivalente a 1 vCPU y 4GB de RAM. Se determina el número de KPUs para la aplicación mediante la especificación de dos parámetros:
- Parallelism – Número de tareas que se pueden ejecutar concurrentemente.
- ParallelismPerKPU – Número de tareas que pueden ejecutarse en una única KPU.
El número total de KPUs de la aplicación viene dado por Parallelism / ParallelismPerKPU. Es posible desplegar este servicio con autoescalado automático, que ajustará automáticamente el número de KPUs en función del consumo de CPU para acomodar la demanda.
Los costos[3] de Amazon Kinesis Analytics se basan en un modelo pay-per-use, apoyándose en las Kinesis Processing Units consumidas. Además, se asume un coste por el almacenamiento usado por la aplicación y sus copias de seguridad.
Flink
Profundizando más en la aplicación de Flink, una de las características más importantes es la capacidad de ser resiliente a fallos. Para ello, Flink incorpora un sistema de checkpointing mediante el cual se toma un snapshot de la aplicación y su estado que es guardado en un almacenamiento remoto en caso de que sea necesario recuperar la aplicación.
El proceso de checkpointing de una aplicación de Flink está diseñado para ser resiliente y eficiente. Flink puede hacer uso de diferentes backends para guardar el estado de la aplicación. El más simple sería la memoría de la propia Java Virtual Machine, y aunque esto ofrece baja latencia y una gestión más simple, rápidamente pueden surgir problemas de escalado y capacidad que no lo hacen recomendable para entornos de producción. Por eso es común el uso de RocksDB como backend de Flink, una base de datos de clave-valor con alto rendimiento, escalable y con tolerancia a fallos. Adicionalmente KDA guarda estos snapshots en S3 para una capa extra de durabilidad.
Para el propósito de este blog, se ha desarrollado una sencilla aplicación de ingesta de datos en tiempo real y su posterior guardado en S3. Flink ofrece dos APIs mediante las cuales puedes desarrollar una aplicación:
- DataStream API – Es una API basada en el concepto de streams. Ofrece control a bajo nivel de la aplicación con la desventaja de requerir un mayor esfuerzo por parte del desarrollador.
- Table API – Esta API se basa en el concepto de tablas. Ofrece una manera declarativa de desarrollar la aplicación mediante el uso de expresiones SQL. Conlleva una pérdida de control sobre los detalles de la aplicación en favor de ser mucho más sencilla.
Para este caso de uso se usará la Table API por su simplicidad, pero es igualmente compatible con el uso de la DataStream API.
A la hora de desplegar la aplicación con Kinesis Data Analytics sólo es necesario definir el punto de entrada del código de la aplicación y proporcionar un uber jar con todas las dependencias de esta. Conviene explicar las dependencias usadas para esta aplicación, pues suele ser uno de los mayores puntos de fricción a la hora desarrollar una aplicación de Flink:
- SQL connector for Kinesis – Conector fundamental para que nuestra aplicación de Flink sea capaz de leer de un Kinesis Stream.
- S3 Filesystem for Hadoop – Permite a la aplicación operar sobre S3.
- Hudi Bundle – Paquete proporcionado por los desarrolladores de Hudi, con todas las dependencias necesarias para trabajar con la tecnología.
- Hadoop MapReduce Client Core – Dependencia adicional necesaria para que la escritura a Hudi funcione correctamente en KDA. Es posible que en futuras versiones del Hudi Bundle esta dependencia no sea necesaria.
La aplicación está preparada para escribir datos tanto en formato JSON como en tablas de Hudi MoR o CoW (que se explicarán en detalle en la siguiente sección). Tanto el código de la aplicación como la infraestructura están disponibles en el repositorio.
Hudi
Conceptos
Hudi se presenta como una fuente de almacenamiento Open Source a nivel de formato de datos. Al igual que hacen otras soluciones como Iceberg o Delta Lake, ofrece algunas propiedades ya existentes en estas como es el soporte de transacciones ACID (Atomicidad, Consistencia, Aislamiento y Durabilidad), procesos enfocados a la optimización de tareas de lectura/escritura, actualización de datos incrementales y otras que se explicarán a continuación. Es importante resaltar que estas no podrían conseguirse mediante ficheros de formato Avro y Parquet.
Las características que presenta Hudi son las siguientes:
- Transacciones ACID: unas de las principales ventajas que ofrece Apache Hudi es el soporte para transacciones ACID, posibilitando que las operaciones de escritura sean atómicas y consistentes. Además también proporciona que los datos estén aislados y sean duraderos, lo que garantiza la integridad de los datos y la consistencia del sistema. Más adelante se analizará más en detalle cómo las distintas formas de almacenamiento lo hacen posible y las ventajas que estas ofrecen.
- Pipelines Incrementales: la clusterización de los eventos en función de variables de negocio permite que tareas de borrado/actualización de datos se puedan realizar de una forma más eficiente si estas se encuentran indexadas de forma conjunta aunque no se hayan dado en la misma franja temporal.
- Ingesta en Streaming: Hudi permite obtener unos workloads computacionalmente menos pesados a través de Upserts que recurren a una indentación optimizada[4] por grupos de archivos, lo que hace que en tareas de escritura (Update/Append/Delete) sean más eficientes. Esto permite que muchas de las aplicaciones basadas en Hudi no deban ser deduplicadas.
- Queries de estados previos de los datos – Time Travel: Hudi permite actualizar y consultar información de particiones pasadas sin la necesidad de tener que reprocesar ni incluir particiones temporales mayores. De esta manera se asegura que eventos enviados con posterioridad no sean procesados y sean correctamente almacenados.
- Tareas de escritura simultáneas: mediante OCC (Optimistic Concurrency Control[5]) se permite que muchas de las tareas como Upsert e Insert puedan realizarse correctamente aun realizándose de forma simultánea.
A la hora de analizar cómo Hudi procede a realizar el almacenamiento de los eventos ingestados, estos son agrupados por particiones y estas a su vez agrupadas en grupos de archivos. Estos últimos teniendo asignado un file_id único para cada grupo en el cual se encuentra el base file, en formato parquet, el cual surge tras una acción, ya sea un commit o compactación, y el log file que es donde se encuentran registrados todas las actualizaciones realizadas (event version tracking).
Tipos de Tablas y Queries
Hudi ofrece 2 tipos de tablas en función de la necesidad de negocio, esto tiene un impacto a nivel de performance y limitación de ciertas funcionalidades como se verán en más detalle:
Copy on Write (COW)
Sistema de almacenamiento mediante el cual en las tareas de actualización, eliminación o registro de nuevos datos se realizan directamente sobre el archivo de logs (delta file) y se crea una nueva instantánea que incluye una copia completa del conjunto de datos actualizado, incluyendo una nueva versión del base file y un archivo delta que contiene los cambios realizados en esa operación.
No es hasta la compactación de datos (programada o al alcanzar un tamaño de datos definido) cuando se realiza la combinación de los archivos delta con la versión más reciente del conjunto de datos completo.Se crea así un nuevo archivo completo donde se eliminan los archivos delta que ya no son necesarios, actualizando a su vez el archivo de índice para que pueda acceder a los datos del archivo compactado.
Este sistema de almacenamiento está especialmente recomendado para casos de uso en los que las tareas de lectura sean más frecuentes que las de escritura al no requerir de transformaciones de datos adicionales al leer los datos. A continuación se muestra el Timeline de los principales archivos al realizarse las distintas tareas de escritura:
| Acción | NUEVO archivo base | Archivo delta | Archivo de índice | Snapshot |
|---|---|---|---|---|
| Nuevo registro | Se escribe el registro en el archivo base | No se crea un archivo delta | Se actualiza el archivo de índice con el nuevo registro | No se crea un nuevo snapshot |
| Actualización de registro existente | Se escribe el registro actualizando en un nuevo archivo base | Se escribe el registro actualizando en el archivo delta | Se actualiza el archivo de índice con la versión actualizada del registro | No se crea un nuevo snapshot |
| Eliminación de registro | No se escribe el registro eliminado en el nuevo archivo | Se escribe una marca de eliminación en un nuevo archivo delta | Se actualiza el archivo de índice con la marca de eliminación | No se crea un nuevo snapshot |
| Compactación de archivos delta | Se fusionan los archivos delta en un nuevo archivo base | No se crea un nuevo archivo delta | Se crea un nuevo archivo índice que contiene todas las entradas del índice de los archivos fusionados | Se crea un nuevo snapshot que refleja el estado actual de los datos después de la compactación |
Merge On-Read (MOR)
En este caso, no se utilizan delta files separados como en el modelo Copy-on-Write (COW). En su lugar, los cambios se escriben directamente en los archivos de datos existentes (base files). En las tareas en las que se realizan actualizaciones de registros, estos nuevos son añadidos en el base file, y en el caso de eliminación, estos son marcados como tal en el base file, en ambos casos estos cambios son registrados en el archivo de índice, hasta que se realiza la compactación. Es en esta operación donde se aplican todas las actualizaciones a los registros en el archivo base correspondiente y elimina las versiones anteriores de los registros actualizados.
Esta alternativa está especializada en realizar consultas de datos históricos versionados y transformaciones y análisis NRT de grandes volúmenes, ya que es posible realizarlo sin tener que copiar los datos a otra ubicación en el disco. Además de ser óptimo para casos de uso en los que las tareas de escritura son concurrentes al ser más eficiente ya que no es necesario realizar transformaciones de datos adicionales durante la escritura, aunque posee una menor tolerancia al fallo ya que en caso de que el archivo de logs se corrompa puede generar pérdida de las versiones de los datos.
A continuación se muestra el Timeline de los principales archivos al realizarse las distintas tareas de escritura:
| Acción | Archivo base | Archivo delta | Archivo de índice | Snapshot |
|---|---|---|---|---|
| Nuevo registro | Se escribe el registro en el archivo base | No se crea un archivo delta | Se actualiza el archivo de índice con el nuevo registro | No se crea un nuevo snapshot |
| Actualización de registro existente | Se escribe el registro actualizando en un nuevo archivo delta | Se escribe el registro actualizando en el archivo delta correspondiente | Se actualiza el archivo de índice con la versión actualizada del registro | No se crea un nuevo snapshot |
| Eliminación de registro | No se elimina el registro del archivo base | Se escribe una marca de eliminación en un nuevo archivo delta | Se actualiza el archivo de índice con la marca de eliminación | No se crea un nuevo snapshot |
| Compactación de archivos delta | Se fusionan los archivos delta en un nuevo archivo base | Se crea un nuevo archivo delta que contiene las actualizaciones pendientes después de la última compactación | Se crea un nuevo archivo índice que contiene todas las entradas del índice de los archivos fusionados | Se crea un nuevo snapshot que refleja el estado actual de los datos después de la compactación |
Como resumen, se realiza una comparativa de las principales métricas de performance entre Copy on-Write y Merge on-Read:
| COW | MOR | |
|---|---|---|
| Coste de escritura | Mayor | Menor |
| Latencia | Mayor | Menor |
| Rendimiento de consulta | Mayor | Menor antes de compactación Igual tras compactación |
- Escritura: COW tiene un mayor costo de escritura que MOR debido a que cada vez que se realiza una operación de escritura (ya sea añadir un nuevo registro o actualizar uno existente), se crea un nuevo delta file y se deben actualizar los archivos de índice correspondientes. En cambio, en MOR, los registros se escriben directamente en el base file, lo que implica una menor cantidad de operaciones de escritura y, por lo tanto, un menor costo en términos de rendimiento y uso de recursos.
- Latencia: COW tiene un menor data latency que MOR debido a que los registros nuevos o actualizados se escriben primero en un delta file separado, en lugar de actualizar directamente el base file como en MOR.
- Tiempos de consulta: COW tiene un menor tiempo de consulta que MOR debido a que en COW, los datos actualizados se almacenan en los Delta Files y los datos originales se mantienen en el Base File. Esto significa que no es necesario realizar ninguna operación de lectura para obtener la versión actualizada de los datos.
Hudi no solo ofrece distintas formas de almacenamiento, sino también, distintas formas de realizar consultas sobre la información almacenada, dependiendo de nuevo tanto de los casos de negocio como del tipo de almacenamiento escogido:
- Snapshots: consulta la última versión procedente de un commit o compactación. Gracias a este tipo de consultas, se pueden obtener las versiones de los datos en momentos específicos gracias a la combinación del base y delta file (time travel). Misma performance en CoW y MoR.
- Read Optimized: únicamente disponible si el tipo de tabla en el que se almacenan los datos es MoR. Basado en la obtención de vistas optimizadas para lectura de un conjunto de datos grande y distribuido. Esto se consigue mediante indexación optimizada (Bloom Filter Index), lo que permite reducir considerablemente el tiempo de búsqueda de datos. Además se apoya también en la compactación de datos que hace que, de nuevo, las tareas de búsqueda sean menos costosas al disminuir el volumen de los mismos.
- Incremental: Permite leer solo los datos actualizados o agregados desde la última consulta. Esto ayuda a reducir el tiempo de lectura y el uso del almacenamiento en disco.
Conclusiones
En este artículo se ha descrito como desplegar una aplicación que ingesta eventos en tiempo real y forma con la salida un LakeHouse con una arquitectura serverless. Con esto se ha buscado un nivel de abstracción intermedio de tal manera que sea una aplicación simple pero con la suficiente potencia para poder llegar a utilizarse en entornos productivos reales.
Desplegar aplicaciones basadas en la combinación de tecnologías como son Apache Flink y Hudi otorga la capacidad de procesar grandes volúmenes de datos en tiempo real y de manera escalable. Esto combinado con la garantía que aportan las transacciones ACID, hace que la combinación de Apache Flink y Apache Hudi sea una solución sólida para la ingesta y procesamiento de datos en entornos críticos.
A pesar de todas las ventajas que se han descrito cabe resaltar algunos inconvenientes que se han podido detectar desarrollando esta arquitectura. El mayor problema que se ha encontrado ha sido la resolución de dependencias entre las librerías de Flink y los conectores necesarios, como por ejemplo el de Hudi. La falta de comunidad que existe a día de hoy, aunque esta crecerá con el paso del tiempo, supuso un problema inicial considerable para poder formar el paquete final con todas las dependencias necesarias sin que hubiese conflictos entre sí. Además, cabe resaltar que se ha percibido menos comunidad para el lenguaje de Python que para el de Java o Scala. En este artículo se eligió Python ya que existía un conocimiento interno más fuerte pero en el caso de que el stack tecnológico se acerque más a lenguajes soportados por la JVM (Java Virtual Machine) sería aconsejable el uso de Scala o Java.
En los próximos artículos entraremos más en detalle en las particularidades que tienen tanto Hudi como Flink para poder personalizar y ajustar el comportamiento de esta aplicación dependiendo de las necesidades que presente nuestro caso de uso.
Referencias
Autores
Alberto Jaen
AWS Cloud Engineer
Empecé mi carrera laboral con el desarrollo, mantenimiento y administración de bases de datos multidimensionales y Data Lakes. A partir de ahí comencé a estar interesado en plataformas de datos y arquitecturas cloud, estando certificado 3 veces en AWS y 2 con Hashicorp.
Actualmente me encuentro trabajando como un Cloud Engineer desarrollando Data Lakes y DataWarehouses con AWS para un cliente relacionado con la organización de eventos deportivos a nivel mundial.
Alfonso Jerez
AWS Cloud Engineer
Comencé mi carrera como Data Scientist en distintos sectores (banca, consultoría,…) enfocado en la automatización de procesos y desarrollo de modelos. En los últimos años aposté por Bluetab motivado por el interés en especializarme como Data Engineer y comenzar a trabajar con los principales proveedores Cloud (AWS, GPC y Azure) en clientes como Olympics, específicamente en la optimización del procesamiento y almacenamiento del dato.
Colaborando activamente con el grupo de Práctica Cloud en investigaciones y desarrollo de blogs de tecnologías punteras e innovadoras tales como esta, fomentando así el continuo aprendizaje.
Adrián Jiménez
AWS Cloud Engineer
Dedicado al aprendizaje constante de nuevas tecnologías y su aplicación, disfrutando de utilizarlas en la resolución de desafíos tecnológicos. Desarrollo mi carrera como Cloud Engineer diseñando, implementando y manteniendo infraestructura en AWS.
Colaboro activamente en la Práctica Cloud, donde investigamos y experimentamos con nuevas tecnologías, buscando soluciones para los retos que enfrentan nuestros clientes.
Navegación
SOLUCIONES, SOMOS EXPERTOS
Te puede interesar
Footer
