{"id":15634,"date":"2023-10-04T14:12:51","date_gmt":"2023-10-04T14:12:51","guid":{"rendered":"https:\/\/beta.bluetab.net\/?p=15634"},"modified":"2023-10-04T14:12:51","modified_gmt":"2023-10-04T14:12:51","slug":"lakehouse-streaming-en-aws-con-apache-flink-y-hudi-parte-2","status":"publish","type":"post","link":"https:\/\/beta.bluetab.net\/en\/2023\/10\/lakehouse-streaming-en-aws-con-apache-flink-y-hudi-parte-2\/","title":{"rendered":"LakeHouse Streaming en AWS con Apache Flink y Hudi (Parte 2)"},"content":{"rendered":"
\"\"<\/a><\/figure>\n

Alberto Jaen<\/a><\/h4>\n

AWS Cloud Engineer <\/p>\n

\"\"<\/a><\/figure>\n

Alfonso Jerez<\/a><\/h4>\n

AWS Cloud Engineer <\/p>\n

\"\"<\/a><\/figure>\n

Adri\u00e1n Jim\u00e9nez<\/a><\/h4>\n

AWS Cloud Engineer <\/p>\n

Introducci\u00f3n<\/h2>\n

Este art\u00edculo es el segundo en una serie de publicaciones que se centran en la creaci\u00f3n de un LakeHouse con Hudi<\/b> a partir de una ingesta en streaming procesada por una aplicaci\u00f3n Flink. El primer art\u00edculo<\/a> se centra en sentar una buena base para esta plataforma, donde se desplegaron unas aplicaciones Flink con KDA (Kinesis Data Analytics) para cada tipo de formato (MoR, CoW para Hudi y JSON) que escriben el resultado de este procesamiento en unos buckets<\/i>.<\/p>\n

El env\u00edo de datos que se utiliza como input<\/i> se mandaba en el anterior art\u00edculo desde una m\u00e1quina en local ejecutando una aplicaci\u00f3n de Locust, lo que puede presentar problemas a la hora de escalar y querer procesar un volumen alto de eventos. Adem\u00e1s, las aplicaciones de Kinesis Data Analytics con Flink presentan problemas de agilidad en su modo de autoescalado. Todos estos nuevos retos ser\u00e1n resueltos en este art\u00edculo.<\/p>\n

Tambi\u00e9n se catalogar\u00e1n estas tablas en Glue, servicio que disponibiliza un cat\u00e1logo de datos en AWS, para poder acceder a estos y as\u00ed realizar queries<\/i> de todo tipo. Como motor de queries <\/i>que consumir\u00e1 estos metadatos se utilizar\u00e1 Athena, que proporciona una experiencia escalable, \u00e1gil y serverless<\/i> para poder ejecutar queries<\/i> con SQL o Spark para nuestras tablas alojadas en S3.<\/p>\n

Por otro lado, en este art\u00edculo tambi\u00e9n se han desplegado los componentes necesarios para poder monitorizar nuestras aplicaciones<\/b> y extraer as\u00ed conclusiones sobre la velocidad a la que se ingestan los datos y los posibles problemas a resolver para que el procesamiento tenga la latencia requerida seg\u00fan los requisitos que se impongan.<\/p>\n

Finalmente se realizar\u00e1 una comparativa en cuanto a rendimiento y latencia de las diferentes aplicaciones de Flink<\/b> que escriben datos en los formatos de Hudi y JSON para as\u00ed poder ver las diferentes ventajas e inconvenientes de estos formatos.\u00a0<\/p>\n

Arquitectura
\n<\/h2>\n

A continuaci\u00f3n se puede ver la arquitectura a alto nivel que se desplegar\u00e1:<\/p>\n

\t\t\t\t\t\t\t\t\t\t\t\t\t\"\"\t\t\t\t\t\t\t\t\t\t\t\t\t<\/p>\n

Para un mayor entendimiento vamos a explicarla de izquierda a derecha. Como se puede observar, el cambio m\u00e1s rese\u00f1able con respecto al primer art\u00edculo es la inclusi\u00f3n de un cluster de Kubernetes<\/b> para poder escalar los eventos que ser\u00e1n mandados como input <\/i>de nuestra aplicaci\u00f3n de streaming<\/i>. De esta manera se podr\u00e1 testear de manera exhaustiva el rendimiento de las aplicaciones de Flink dependiendo de su aprovisionamiento y sobre todo del tipo de formato y tabla en el que escriben al LakeHouse. Adem\u00e1s, se ha disponibilizado un ALB (Application Load Balancer) que permite acceder a la interfaz de Locust para poder definir el n\u00famero de usuarios a simular y c\u00f3mo deben escalar estos con el tiempo. La URL para acceder a esta aparecer\u00e1 como output <\/i>al desplegar la infraestructura con Terraform.<\/p>\n

Por otro lado se han realizado cambios rese\u00f1ables en las aplicaciones KDA de Flink y el <\/b>stream <\/i><\/b>del que leen estas. Cada aplicaci\u00f3n lee ahora como consumidores EFO (Enhanced Fan Out), de tal manera que cada una de ellas tiene un ancho de banda dedicado. La raz\u00f3n de este cambio y sus detalles ser\u00e1n explicados m\u00e1s en detalle en el apartado dedicado para Kinesis.<\/p>\n

En cuanto a la monitorizaci\u00f3n y la extracci\u00f3n de m\u00e9tricas en NRT (Near Real Time) se han desplegado unas funciones lambdas que acceden a las tablas apoy\u00e1ndose en Athena gracias a haber registrado los metadatos de estas en el cat\u00e1logo de Glue. Es importante resaltar que los metadatos de las tablas de Hudi son registrados en Glue por Flink pero en el caso de JSON se despliega un crawler <\/i>que registra estas tablas en el cat\u00e1logo. Este crawler se debe ejecutar manualmente para que esta tabla quede registrada en Glue.<\/p>\n

Escalado<\/h2>\n

Kinesis Stream<\/h3>\n

Dado que el objetivo es someter la aplicaci\u00f3n a una carga considerable de eventos por segundo, es necesario explicar c\u00f3mo cada una de las piezas de la arquitectura pueden escalar de acuerdo al volumen de datos.<\/p>\n

Como hemos comentado previamente, se ha optado por un Kinesis Stream On-Demand<\/i> para automatizar el escalado de los shards<\/i> durante las pruebas de carga. Es necesario tener en cuenta que estos streams<\/i> pueden acomodar una tasa de escritura de hasta el 200% de lo especificado por el n\u00famero de shards<\/i> en un momento dado.<\/p>\n

Una vez que el stream<\/i> se encuentra por encima del 100%, aumentar\u00e1 autom\u00e1ticamente el n\u00famero de shards<\/i> en un plazo de 15 minutos. La \u00fanica limitaci\u00f3n por tanto es no superar el doble del volumen de escritura admitido en menos de dicho periodo.<\/p>\n

Por otro lado, dado que se tendr\u00e1n tres aplicaciones de Flink leyendo del mismo stream<\/i>, las limitaciones a nivel de lectura ser\u00e1n el mayor problema. Un Kinesis Stream solo admite 5 llamadas GetRecord<\/i> por shard <\/i>por segundo. Dado que cada aplicaci\u00f3n tiene que leer todo el stream<\/i> (y por lo tanto, todos los shards<\/i>), aumentar el n\u00famero de shards<\/i> no ayuda a solventar este problema.<\/p>\n

La soluci\u00f3n pasa por registrar cada una de las aplicaciones como un consumidor Enhanced Fan-Out. Esta funcionalidad de los Kinesis Stream provee a cada uno de estos consumidores con un l\u00edmite individual de 5 llamadas GetRecord<\/i> y 2MB por shard<\/i> por segundo de lectura.<\/p>\n

\t\t\t\t\t\t\t\t\t\t\t\t\t\"\"\t\t\t\t\t\t\t\t\t\t\t\t\t<\/p>\n

Esta configuraci\u00f3n se realiza en el lado del consumidor, en nuestro caso mediante el conector de Kinesis para Flink:<\/p>\n

'scan.stream.recordpublisher' = 'EFO',\n'scan.stream.efo.registration' = 'EAGER\/LAZY',\n'scan.stream.efo.consumername' = '{consumer_name}' <\/code><\/pre>\n
Conviene mencionar que alternativamente, es posible aumentar la latencia de lectura de nuestras aplicaciones de Flink. Por defecto Flink realiza una lectura cada 200 ms por shard<\/i>, de modo que una aplicaci\u00f3n consume completamente la cuota de lectura de un stream<\/i>. Incrementando este valor a 600ms podr\u00edamos acomodar las tres aplicaciones, a costa de una mayor latencia:<\/p>\n
scan.shard.getrecords.intervalmillis = '600' <\/code><\/pre>\n
Tambi\u00e9n se har\u00e1 uso de la opci\u00f3n Adaptive Reads<\/i>, que modifica din\u00e1micamente el n\u00famero de eventos recogidos por llamada en funci\u00f3n del tama\u00f1o de cada record. Esto permite aprovechar los 2 MB\/s por shard <\/i>disponibles para cada consumidor:\u00a0 <\/p>\n
'scan.shard.adaptivereads' = 'true' <\/code><\/pre>\n
En lo que respecta al escalado en KPUs (Kinesis Processing Unit<\/i>) de Flink, se ha optado por no hacer uso del autoescalado, ya que cada proceso de escalado incurren en downtime<\/i> para la aplicaci\u00f3n. Debido a los diferentes requerimientos de cada una de las aplicaciones, las acciones de escalado en momentos inesperados podr\u00edan interrumpir las pruebas de carga. Adem\u00e1s es interesante medir el rendimiento de escritura de cada una de las aplicaciones en igualdad de capacidad de computaci\u00f3n.<\/p>\n
Hudi<\/h2>\n
Timeline<\/h3>\n
Uno de los sistemas base sobre la que se sustenta el funcionamiento y caracter\u00edsticas de Hudi es la timeline<\/i>. Hudi guarda un registro temporal de todas las acciones que se han realizado sobre la tabla, as\u00ed como el estado de esta acci\u00f3n.<\/p>\n
Las principales acciones que componen la timeline <\/i>son<\/p>\n