• Saltar a la navegación principal
  • Saltar al contenido principal
  • Saltar al pie de página
Bluetab

Bluetab

an IBM Company

  • Soluciones
    • DATA STRATEGY
    • DATA FABRIC
    • AUGMENTED ANALYTICS
  • Assets
    • TRUEDAT
    • FASTCAPTURE
  • Conócenos
  • Oficinas
    • España
    • Mexico
    • Perú
    • Colombia
  • talento
    • España
    • TALENT HUB ALICANTE
    • TALENT HUB BIZKAIA
    • TALENT HUB BARCELONA
  • Blog
  • ES

interes

LakeHouse Streaming en AWS con Apache Flink y Hudi (Parte 2)

octubre 4, 2023 by Bluetab

LakeHouse Streaming en AWS con Apache Flink y Hudi (Parte 2)

Alberto Jaen

AWS Cloud Engineer

Alfonso Jerez

AWS Cloud Engineer

Adrián Jiménez

AWS Cloud Engineer

Introducción

Este artículo es el segundo en una serie de publicaciones que se centran en la creación de un LakeHouse con Hudi a partir de una ingesta en streaming procesada por una aplicación Flink. El primer artículo 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.

El envío de datos que se utiliza como input se mandaba en el anterior artículo desde una máquina en local ejecutando una aplicación de Locust, lo que puede presentar problemas a la hora de escalar y querer procesar un volumen alto de eventos. Además, las aplicaciones de Kinesis Data Analytics con Flink presentan problemas de agilidad en su modo de autoescalado. Todos estos nuevos retos serán resueltos en este artículo.

También se catalogarán estas tablas en Glue, servicio que disponibiliza un catálogo de datos en AWS, para poder acceder a estos y así realizar queries de todo tipo. Como motor de queries que consumirá estos metadatos se utilizará Athena, que proporciona una experiencia escalable, ágil y serverless para poder ejecutar queries con SQL o Spark para nuestras tablas alojadas en S3.

Por otro lado, en este artículo también se han desplegado los componentes necesarios para poder monitorizar nuestras aplicaciones y extraer así 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ún los requisitos que se impongan.

Finalmente se realizará una comparativa en cuanto a rendimiento y latencia de las diferentes aplicaciones de Flink que escriben datos en los formatos de Hudi y JSON para así poder ver las diferentes ventajas e inconvenientes de estos formatos. 

Arquitectura

A continuación se puede ver la arquitectura a alto nivel que se desplegará:

Para un mayor entendimiento vamos a explicarla de izquierda a derecha. Como se puede observar, el cambio más reseñable con respecto al primer artículo es la inclusión de un cluster de Kubernetes para poder escalar los eventos que serán mandados como input de nuestra aplicación de streaming. De esta manera se podrá 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ás, se ha disponibilizado un ALB (Application Load Balancer) que permite acceder a la interfaz de Locust para poder definir el número de usuarios a simular y cómo deben escalar estos con el tiempo. La URL para acceder a esta aparecerá como output al desplegar la infraestructura con Terraform.

Por otro lado se han realizado cambios reseñables en las aplicaciones KDA de Flink y el stream del que leen estas. Cada aplicación lee ahora como consumidores EFO (Enhanced Fan Out), de tal manera que cada una de ellas tiene un ancho de banda dedicado. La razón de este cambio y sus detalles serán explicados más en detalle en el apartado dedicado para Kinesis.

En cuanto a la monitorización y la extracción de métricas en NRT (Near Real Time) se han desplegado unas funciones lambdas que acceden a las tablas apoyándose en Athena gracias a haber registrado los metadatos de estas en el catálogo 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 que registra estas tablas en el catálogo. Este crawler se debe ejecutar manualmente para que esta tabla quede registrada en Glue.

Escalado

Kinesis Stream

Dado que el objetivo es someter la aplicación a una carga considerable de eventos por segundo, es necesario explicar cómo cada una de las piezas de la arquitectura pueden escalar de acuerdo al volumen de datos.

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

Una vez que el stream se encuentra por encima del 100%, aumentará automáticamente el número de shards en un plazo de 15 minutos. La única limitación por tanto es no superar el doble del volumen de escritura admitido en menos de dicho periodo.

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

La solución 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ímite individual de 5 llamadas GetRecord y 2MB por shard por segundo de lectura.

Esta configuración se realiza en el lado del consumidor, en nuestro caso mediante el conector de Kinesis para Flink:

'scan.stream.recordpublisher' = 'EFO',
'scan.stream.efo.registration' = 'EAGER/LAZY',
'scan.stream.efo.consumername' = '{consumer_name}' 

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, de modo que una aplicación consume completamente la cuota de lectura de un stream. Incrementando este valor a 600ms podríamos acomodar las tres aplicaciones, a costa de una mayor latencia:

scan.shard.getrecords.intervalmillis = '600' 

También se hará uso de la opción Adaptive Reads, que modifica dinámicamente el número de eventos recogidos por llamada en función del tamaño de cada record. Esto permite aprovechar los 2 MB/s por shard disponibles para cada consumidor: 

'scan.shard.adaptivereads' = 'true' 

En lo que respecta al escalado en KPUs (Kinesis Processing Unit) de Flink, se ha optado por no hacer uso del autoescalado, ya que cada proceso de escalado incurren en downtime para la aplicación. Debido a los diferentes requerimientos de cada una de las aplicaciones, las acciones de escalado en momentos inesperados podrían interrumpir las pruebas de carga. Además es interesante medir el rendimiento de escritura de cada una de las aplicaciones en igualdad de capacidad de computación.

Hudi

Timeline

Uno de los sistemas base sobre la que se sustenta el funcionamiento y características de Hudi es la timeline. Hudi guarda un registro temporal de todas las acciones que se han realizado sobre la tabla, así como el estado de esta acción.

Las principales acciones que componen la timeline son

  • Commits – escritura atómica de un conjunto de registros en la tabla en formato columnar
  • Delta Commit – similar al commit, representa una escritura de registros en forma de logs en una tabla Merge on Read
  • Compaction – compactación de las escrituras en logs (delta commits) de una tabla MoR a formato columnar
  • Cleans – borrado de versiones antiguas de archivos
  • Rollback – eliminado de los registros escritos por un commit o delta commit fallido
  • Savepoint – marca un conjunto de archivos como “guardados” para que no sean eliminados por el proceso de limpieza. Permite restaurar la tabla a un punto anterior en la timeline

Cualquiera de estas acciones pueden encontrarse en uno de estos tres estados

  1. Requested – una acción ha sido planeada sin iniciar
  2. Inflight – la acción está en proceso
  3. Completed – denota que la acción ha sido completada


Tipos de tabla

Como se ha dejado entrever en el funcionamiento de la timeline de Hudi, existen dos tipos de escritura soportados: columnar y logs. El formato columnar (parquet) constituye la forma final de una tabla de Hudi, junto con los metadatos de la timeline. Sin embargo, es posible hacer uso de las escrituras en logs (avro) para disminuir la latencia de escritura y eventualmente compactarse a formato columnar sin entorpecer la escritura.

El uso de estos métodos de escritura dan lugar a los dos tipos de tabla que Hudi pone a nuestra disposición

  • Copy on Write – las escrituras se realizan exclusivamente en formato columnar, creando un nuevo fichero con los nuevos registros de la tabla. Los datos están disponibles inmediatamente pero incurre en mayor latencia de escritura
  • Merge on Read – hace uso de la escritura en logs. Los nuevos registros son inicialmente escritos como logs, y posteriormente serán transformados a formato columnar por el proceso de compactación. Obtenemos menor latencia de escritura a costa de latencia de lectura; los nuevos registros no estarán disponibles hasta que se realice la compactación.

Tipos de Query

Para poder aprovechar las características de cada tipo de tabla, existen tres tipos de queries que se pueden realizar sobre una tabla de Hudi

  • Snapshot – obtiene la última versión de la tabla. Para las tablas MoR esto implica incurrir en un proceso de compactación para obtener los últimos registros en formato log. 
  • Read Optimized – para tablas MoR, lee sólamente los registros ya expuestos en formato columnar sin incurrir en latencia de lectura adicional.
  • Incremental – recoge únicamente los nuevos registros desde un cierto commit o compactación, facilitando la creación de pipelines incrementales. No está soportada por Athena

Integración con Glue Catalog


El conector de Hudi permite una integración nativa con el catálogo de Glue en AWS. Basta con añadir las dependencias de Hive en nuestra aplicación de Flink:

com.amazonaws.aws-java-sdk-glue
org.apache.hive.hive-common
org.apache.hive.hive-exec 

Y especificar la configuración del catálogo en el conector de Hudi:

'hive_sync.enable' = 'true',
'hive_sync.db' = '{glue_database}',
'hive_sync.table' = '{table_name}',
'hive_sync.partition_fields' = '{partition_fields}',
'hive_sync.mode' = 'glue',
'hive_sync.use_jdbc' = 'false' 

Con esta integración, la aplicación creará automáticamente las tablas en el catálogo. Como hemos mencionado anteriormente, existen distintos tipos de query para consultar una tabla de Hudi. Se crearán por tanto en el catálogo distintas tablas para soportar las diferentes consultas.

Para una tabla CoW, la tabla se consultará mediante una query Snapshot. Para MoR en cambio se pondrán a disposición dos tablas, para soportar consultas Read Optimized o Snapshot.

La principal aplicación de Glue es de soporte a las lambdas para que al ejecutar las queries mediante Athena su ejecución pueda realizarse de una forma más eficiente, rápida y segura:

  • Glue Catalog: almacenamiento centralizado de la información acerca de la organización, diseño y formato de los datos, utilizado por Athena para realizar directamente las consultas a S3 sin necesidad de tener que apoyarse en terceros para conseguir esta información
  • Automatización del Esquema: Glue rastrea y cataloga automáticamente los datos en S3, detectando y adaptando los cambios en el esquema. Esto evita posibles errores y permite la lectura de los nuevos campos en caso de que se produzcan alteraciones en los esquemas de los eventos

Configuración de Hudi

Es importante entender las configuraciones que nos ofrece Hudi para optimizar nuestra aplicación, en particular para una aplicación en Near Real Time conviene estar al tanto de las opciones disponibles. Aunque la capacidad de configuración es inmensa [1], se intentará sintetizar las que pueden ser más relevantes para una primera toma de contacto con esta tecnología.

Particionado

Apache Hudi ofrece los tipos de particionado que pueden encontrarse en otras soluciones, se detallarán las principales y se justificara la implementada:

  • Simple: particionado basado en un único campo, en este caso el campo escogido es ‘ticker’ ya que se ha identificado que es el que tiene una cardinalidad menor.
  • Particionado Compuesto: particionamiento basado en múltiples campos, podría resultar interesante escoger un campo de baja cardinalidad (ticker) y otro de cardinalidad media (fecha)
  • Particionado Dinámico: elección de la variable en base de los valores, puede resultar interesante cuando la cardinalidad de las variables puede sufrir variaciones y se quiera una actualización del particionamiento de una forma automática y flexible.

Índices

Apache Hudi cuenta con una múltiples  tipos de indexación[2], comentaremos brevemente los más comunes:

  • Bloom Index – Hace uso de un bloom filter sobre la key de los eventos, adicionalmente se puede complementar con un filtrado por rango de de key. Funciona bien cuando tratamos con una tabla donde la mayoría de cambios ocurren en las particiones más recientes o para deduplicado de eventos.
  • Simple: indexación realizada mediante la combinación de FileID y RecordKey. Recomendado cuando las operaciones Upsert no son tan frecuentes debido a la simplicidad que este ofrece.

Ambos tipos de índices pueden ser usados en su forma global

  • Índice global – Imponen la unicidad de las keys en todas las particiones de la tabla, es decir, garantizan que existirá sólamente un registro con una cierta key.
  • Índice no global – La unicidad de la key sólo es exigida a nivel de partición. Si los datos son consistentes y una key sólo va a existir en una partición, este tipo de índices ofrecen un rendimiento mucho mayor y mejor escalado.

En este caso, se ha optado por un Bloom Index, el cual es el que se toma por defecto en caso de que no se declare expresamente:

"hoodie.index.type" = "BLOOM" 

La elección de este tipo de indexación se debe a que los casos de uso que se han planteado requieren de un procesamiento de datos considerablemente alto y eficiente.

Tipos de operación

Apache Hudi ofrece varios tipos de operaciones[3] que permiten a los usuarios administrar y modificar conjuntos de datos de gran tamaño. A continuación se detallan tanto las principales operaciones realizadas en los Stress Tests como en otros escenarios:

  • Upsert – Es la operación por defecto, y ejecutará un insert o un update dependiendo de si el registro ya existe tras una búsqueda en el índice. Con esta operación la tabla no tendrá duplicados para su clave primaria.
  • Insert – Esta operación ignora la búsqueda en el índice a la hora de insertar eventos. Es la más rápida pero la tabla puede contener duplicados. Aún así es útil si se utilizan métodos auxiliares  de deduplicado, o simplemente la existencia de estos es tolerable en el caso de uso.
  • Delete: Hudi ofrece dos métodos de borrado. Soft Delete convierte a nulos los valores del evento a excepción de la key. Hard Delete ejecuta un borrado físico del evento en la tabla.
  • Bulk Insert Operación similar al Insert pero optimizada para la inserción de un gran volumen de datos, a costa de sacrificar ciertas garantías en el control del tamaño de ficheros. Escala bien para cientos de TBs en caso de bootstrap inicial de una tabla de gran tamaño.

Compactación

En el caso de usar una tabla MoR es posible configurar el ritmo de compactación de logs en parquet para buscar el equilibrio entre latencia de escritura y lectura que más convenga al caso de uso. Se pueden especificar una estrategia de tiempo o número de delta commits (o ambos) que ejecutan un proceso de compactación:

compaction.delta_commits
compaction.delta_seconds
compaction.trigger.strategy 

Acciones asíncronas

Ciertas acciones de la timeline como la compactación, limpieza, archivado y clustering pueden ser realizadas asíncronamente por la aplicación, o incluso ser relegadas a procesos auxiliares a la aplicación de escritura. Para el caso de Flink, puede ayudar a mejorar la latencia de escritura y evitar problemas de BackPressure en la aplicación:

compaction.async.enabled
hoodie.clean.async
hoodie.archive.async
hoodie.clustering.async.enabled 

Stress Tests & Insights

Al desplegar las aplicaciones, se ha procedido a realizar distintos tests variando tanto la carga máxima de eventos como la concurrencia y el grado exponencial de crecimiento de los mismos. Esto ha sido posible  gracias a la flexibilidad ofrecida por Locust al estar levantado sobre un cluster de Kubernetes, pudiendo establecer un límite máximo de concurrencia de eventos y un incremental de los mismos. En los tests se ha establecido un límite máximo de 5 a 15K usuarios simultáneos (Peak Concurrency) escalando la frecuencia de los mismos de forma lineal, desde 5 a 20 usuarios más por segundo (Spawn Rate):

Se ha procedido a monitorizar los distintos test para así sacar conclusiones del rendimiento teniendo en cuenta las características específicas de cada uno de los formatos. Las métricas en las que se han apoyado los análisis son tanto las nativas de CloudWatch Metrics (CPU & Memory Utilization, KPUs, LastCheckpoint SIze & Duration,..), como las métricas obtenidas a partir de las Lambdas que periódicamente consultan el número de eventos disponibles en los buckets y realizan cálculos del promedio de la latencia de los mismos.


Número de Eventos

A la hora de analizar el número total de eventos procesados, los cuales son enviados de forma gradual, es decir, a medida que pasa el tiempo cada vez son más los eventos que se envían por segundo, se identifica una tendencia bastante similar aunque destacan JSON y Hudi MoR sobre Hudi CoW en cuanto a la rendimiento. Cabe destacar que JSON muestra un crecimiento más estable y constante en comparación con Hudi MoR y CoW y esto se debe a que estos últimos son capaces de manejar actualizaciones incrementales en los datos.

La similitud entre JSON y Hudi MoR hace que la elección se base completamente en las características del proyecto. En caso de que los datos no sean actualizados JSON puede resultar una solución más interesante debido principalmente a su simplicidad, mientras que si hay una alta frecuencia de actualización de datos históricos, Hudi MoR puede ser una mejor solución. Esto se debe tanto a la mayor eficiencia en las tareas de lectura como por la posibilidad de registrar las distintas versiones de los datos.

 

Latencia

Debido a la dificultad de estandarizar la lógica del cálculo de la latencia entre 3 tipos de almacenamiento distintos, se ha optado por simplificarla calculandolo como la diferencia entre la hora de creación del evento y la del procesamiento en la respectiva aplicación.

Se observa un comportamiento similar entre JSON y Hudi MoR, aunque este primero de una forma más crítica, al tener una latencia inicial muy baja pero a medida que tanto el tiempo de procesamiento como el volumen de carga aumenta, esta latencia se ve negativamente afectada.

La elección entre JSON y Hudi MoR dependerá tanto de la tolerancia de fallo que tenga la aplicación como las propias características de cada uno de los formatos, en caso de que la estructura de los datos sea estable y no cambie con frecuencia,o bien, no dependa de actualizaciones incrementales y pueda lidiar con reescrituras completas, en ese caso JSON puede que sea una mejor opción.

La elección de Hudi CoW sobre MoR puede darse cuando se necesite una alta tolerancia a errores y una alta capacidad de recuperación de eventos de escritura fallidos o corrompidos.`


Uso de CPU

Al analizar el uso de CPU, se ha identificado cierta homogeneidad entre los distintos tests aun trabajando con distintas cargas de trabajo. JSON Y Hudi MoR destacan por tener los niveles de uso de CPU más bajos, ambos por distintos motivos. JSON destaca por la simplicidad al incluir directamente los nuevos datos sin necesidad de tener que lidiar con versionado de datos, mientras que MoR no consume tanta CPU ya que por sus características, el consumo mayor de CPU se hace al realizar consultas de lectura, en las tareas de escritura únicamente identifica los cambios que serán aplicados al consultarlos.

Recordar que las métricas nativas de CloudWatch únicamente nos permiten monitorizar las aplicaciones, que corresponden a las tareas de escritura. La monitorización de las tareas de lectura corresponde a las Lambdas mencionadas anteriormente. 

En este caso MoR es más beneficioso respecto a CoW, dado que el mayor consumo de CPU en MoR se produce al consultar los datos almacenados mientras que en CoW tiene lugar al actualizar los datos.

La elección entre los formatos más eficientes se deben a las necesidades del proyecto, en caso de que se requiera una mayor tolerancia al fallo, versionado de los datos y una mayor eficiencia de lectura, se optara por MoR frente a JSON, entre los dos formatos de Hudi, de nuevo, la elección dependerá de las características del proyecto, en caso de que las consultas requieran transformaciones pesadas y/o complejas se optaría por MoR, si en cambio, el proyecto requiera de una mayor integridad de datos y/o la ingesta de datos sea en batch,  resultaría más interesante CoW debido a que al trabajar con esos volúmenes de datos, el contar con copias de seguridad, en caso de surgir errores, el impacto en término de costes y tiempo de recuperación es menor.

 

Memory Utilization

JSON de nuevo destaca por tener los valores de uso de memoria más bajos aunque para la operativa de transformaciones que se realizan son relativamente altos y más teniendo en cuenta que no tiene que lidiar con la administración de versiones o la combinación de datos. Estos valores se deben a que no tiene capacidades de compresión optimizadas ni manejo eficiente de esquemas.

Respecto a Hudi, se pueden obtener unas conclusiones similares a las del apartado de uso de CPU, MoR tiene una utilización de memoria mayor que JSON debido al procesamiento de logs delta y la administración de versiones y una menor a CoW ya que la consolidación real de los datos no ocurre durante la escritura.


Last Checkpoint Size

Destacar, nuevamente, la estabilidad de JSON frente a las aplicaciones Hudi, ya que no solo muestra en los test realizados un valor inferior a ambos, si no una estabilidad que no se consigue ni con MoR ni CoW, ya que como puede apreciarse, al monitorear el tamaño de los Checkpoints, se percibe una volatilidad considerable.

La volatilidad percibida en las aplicaciones Hudi se debe principalmente a fallos surgidos en Checkpoints lo que conlleva que el Checkpoint posterior al fallido, tenga un volumen mayor. Además de esto, la volatilidad en los tamaños de los Checkpoints puede estar relacionado con las operaciones de optimización y compactación realizadas internamente que puede conllevar la compactación del estado y que esto reduzca considerablemente el tamaño del mismo.

Desafíos en el desarrollo

Read Throughput de Kinesis y EFO

Para no sobrepasar el límite de lectura sobre el Kinesis Stream se ha optado por suscribir los consumidores como Enhanced Fan-Out. En algunas pruebas en conjunto con Autoscaling esto ha dado problemas con el conector de Kinesis de Flink siendo incapaces de cerrar conexiones a la hora de escalar el cluster.


Configuración de Hudi

La configuración de Hudi ha sido otro de los puntos de fricción durante el desarrollo. Bajo cargas elevadas los procesos de compactación y limpieza son más propensos a causar problemas de Backpressure y causar errores en la aplicación. Aunque configurar estos procesos para que ocurran de forma asíncrona puede aliviar este problema, pueden surgir conflictos y desalineación entre procesos bajo cargas elevadas. Un equilibrio entre estas configuraciones y la capacidad del cluster de la aplicación son claves para el buen funcionamiento de la aplicación.

Heterogeneidad de formato

Al hacer un análisis del rendimiento de las 3 aplicaciones, se cuenta con una dificultad adicional debido a la naturaleza de los tipos de formato, teniendo esto tanto un impacto a la hora de plantear la arquitectura como en el planteamiento de las lógicas.
El distinto comportamiento de los formatos en la ingesta, complica el desarrollo de las lógicas a la hora de calcular la latencia. MoR escribe en logs previa compactación, por lo que los datos no están disponibles inmediatamente como ocurre con CoW o JSON.  Esto implica que la métrica común medible para todos los formatos es la de disponibilidad de lectura, la cual no es el principal objetivo de una tabla MoR.  


Sincronización con el Glue Catalog

Una de las grandes ventajas que nos hemos encontrado con Hudi es su capacidad para sincronizarse con el catálogo de Glue, creando las tablas y manteniéndose actualizadas sin necesidad de un crawler. Esto permite una aplicación y arquitectura más limpia que para el caso de JSON, para el cual debe ejecutarse manualmente al desplegar las aplicaciones.

Conclusiones

Los resultados de los tests muestran diferencias considerables entre los formatos JSON, Hudi MoR y CoW en términos de eficiencia, capacidad de respuesta y utilización de recursos. Se procede a analizar cada uno de los aspectos más en detalle:

  • Eficiencia de Procesamiento: JSON y Hudi MoR destacan en la mayoría de las métricas, mostrando un desempeño óptimo en términos de Latencia, CPU & Memory Utilization. Sin embargo, el comportamiento de JSON es más estable y predecible, aunque MoR cuente con ventajas sobre JSON, como por ejemplo, en la gestión de actualizaciones incrementales.
  • Resiliencia y Tolerancia a Fallos: la tolerancia a fallos es un factor muy importante en la decisión sobre la elección entre Hudi y JSON. En el caso de  MoR y CoW, dependerá del grado de criticidad, ya que a nivel general el rendimiento en tareas de escritura para MoR es superior.
  • Uso de Recursos: JSON se muestra como el más ligero, con baja utilización de CPU y memoria, debido a su simplicidad inherente. Mientras que Hudi MoR y CoW, por la naturaleza de su diseño y gestión de datos, requieren más recursos, especialmente en operaciones que involucran el manejo de versiones y la compactación de datos.

Para finalizar, resulta interesante identificar en quéque casos de uso o proyectos puede resultar más recomendable cada uno de los formatos en función de las características de los mismos y las red flags que puedan establecerse:

  • JSON: Recomendado para aplicaciones con estructuras de datos estables que no requieren actualizaciones incrementales y donde la simplicidad y la estabilidad son clave.
  • Hudi MoR: Adecuado para proyectos que requieren una gestión eficiente de actualizaciones incrementales y donde la latencia y la eficiencia en la escritura son cruciales.
  • Hudi CoW: Ideal para contextos donde la integridad de los datos es esencial, y se necesita una robusta recuperación de errores, especialmente en escenarios de ingestas en batch. 

Referencias

[1] Configuraciones Tablas Hudi. [link]

[2] Tipos de Indexacion Hudi. [link]

[3] Tipos de Operaciones Hudi. [link]

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

Apasionado de los datos y las nuevas tecnologías, especializado como AWS Cloud Engineer en la optimización de DataWarehouses y procesos de ingesta y transformación de Data Lakes. Motivado por la mejora continua y automatización de la integración de servicios.

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

¿Quieres saber más de lo que ofrecemos y ver otros casos de éxito?
DESCUBRE BLUETAB

SOLUCIONES, SOMOS EXPERTOS

DATA STRATEGY
DATA FABRIC
AUGMENTED ANALYTICS

Te puede interesar

LakeHouse Streaming en AWS con Apache Flink y Hudi (Parte 2)

octubre 4, 2023
LEER MÁS

Big Data e IoT

febrero 10, 2021
LEER MÁS

Del negocio físico a la explosión del On-Line

abril 7, 2021
LEER MÁS

¿Cómo gobernar los productos de datos? Hacia un gobierno de los datos federado

mayo 12, 2022
LEER MÁS

Algunas de las capacidades de Matillion ETL en Google Cloud

julio 11, 2022
LEER MÁS

Snowflake, el Time Travel sin DeLorean para unos datos Fail-Safe.

febrero 23, 2023
LEER MÁS

Publicado en: Blog, Blog, Destacado, interes, Destacado, Practices

Snowflake: Zero-Copy clone, o cómo librarte del duplicado de datos al clonar.

marzo 22, 2023 by Bluetab

Snowflake: Zero-Copy clone, o cómo librarte del duplicado de datos al clonar.

Roberto García Parra

Technical Delivery Manager

Gabriel Gallardo Ruiz

Enterprise Architect

Introducción

Cómo continuación a la serie de artículos que estamos haciendo sobre las funcionalidades avanzadas que se derivan de la forma en la que se almacenan los datos en Snowflake, presentamos este nuevo artículo sobre el Zero-copy clone, que permite mediante diferentes operaciones a nivel metadato poder tener diferentes copias o versiones de la información, sin tener que duplicar datos en la mayoría de las ocasiones.

¿Qué es Zero-Copy Clone?

Uno de los casos de uso más frecuente que implica gran consumo de tiempo, recursos y almacenamiento, especialmente si hablamos de grandes dataset, es el copiado de datos. Para la realización de copias de objetos, snowflake ofrece zero-copy clone. Esta operación se realiza sobre la metadata, lo que permite realizar clonado de objetos rápidamente sin tener que duplicar los datos.

¿Cómo funciona?

Snowflake realmente lo que realiza es una copia de la metadata asociada al objeto que se va a clonar. Como podemos ver en el ejemplo de clonación de la tabla ‘Events’ en la siguiente imagen, simplemente duplica la metadata sin realizar ningún cambio en la parte de almacenamiento.

Una vez realizado el clon, los objetos clonados tienen su propio ciclo de vía, lo que permite que se puedan realizar cambios sobre los datos sin afectar al objeto original, de igual modo los cambios realizados sobre el objeto original tampoco serán reflejados sobre el objeto clonado.

Zero-copy clone permite la realización de clones prácticamente de cualquier objeto de Snowflake siendo especialmente útil en bases de datos, esquemas y tablas.

¿Qué coste tiene?

Al tratarse de una operación exclusiva de metadata, no se repercuten costes ni de procesamiento ni de almacenamiento, ni siquiera es necesario realizar la operación con un virtual data warehouse activo.

¿Cómo se puede clonar una tabla?

Privilegios: Para poder clonar una tabla, el ROLE que va a realizar la clonación tiene que tener privilegios de SELECT sobre la tabla que se va a clonar, además como es lógico, privilegios de CREATE TABLE sobre el esquema destino en el que se va a crear el clon de la tabla.

Sentencia: La sentencia utilizada para la clonación de tablas es similar a la de creación pero añadiendo la cláusula CLONE. A continuación, vamos a clonar la tabla “events»:

USE ROLE INGESTA_HUB_ROLE;
USE SCHEMA WEATHER.HISTORICAL;
CREATE TABLE EVENTS_CLONE CLONE EVENTS;

Podemos comprobar que la clonación de la tabla se realiza de inmediato, ya que como se comentó anteriormente únicamente se opera sobre la metadata.

Además, podemos observar en la siguiente tabla que todas las propiedades de la tabla origen se han clonado en la nueva tabla. Únicamente en el caso en que la tabla origen tenga asignado una cluster key, la nueva tabla se creará con automatic_clustering suspendido.

EVENTS EVENTS
cluster_by
LINEAR (COUNTRY,CITY)
LINEAR (COUNTRY,CITY)
rows
7,479,165
7,479,165
bytes
105,110,528
105,110,528
owner
INGESTA_HUB_ROLE
INGESTA_HUB_ROLE
retention_time
30
30
automatic_clustering
ON
OFF
change_tracking
OFF
OFF
search_optimization
OFF
OFF
is_external
N
N

Con respecto a los privilegios, por defecto no serán clonados. Esto lo podemos comprobar con las sentencias siguientes:

SHOW GRANTS ON TABLE WEATHER.HISTORICAL.EVENTS;

SHOW GRANTS ON TABLE WEATHER.HISTORICAL.EVENTS_CLONE;

Para que se clonen los privilegios asignados a la tabla origen, hay que añadir COPY GRANTS en la sentencia de clonado:

CREATE TABLE EVENTS_CLONE_1 CLONE EVENTS COPY GRANTS;

Ahora podemos comprobar que los privilegios han sido clonados:

SHOW GRANTS ON TABLE WEATHER.HISTORICAL.EVENTS_CLONE;

Clonación usando time travel

Snowflake permite realizar la clonación de una tabla para un momento histórico determinado, para ello tendremos que utilizar la cláusula AT o BEFORE en la sentencia de clonado.

Para la ejecución de la prueba, vamos a hacer cambios en la tabla de EVENTS y después realizaremos el clonado con un time travel anterior al cambio.

DELETE FROM EVENTS WHERE AIRPORTCODE=’KS47′;

Clonamos la tabla con un time travel anterior a la realización del borrado

CREATE TABLE EVENTS_CLONE_TIME_TRAVEL CLONE EVENTS at (offset => -60*5);

Si consultamos la información referente a ambas tablas, podemos comprobar que el clonado se ha realizado en el momento anterior en el que la tabla EVENTS tenía 9.062 filas más.

Consideraciones del clonado de tablas

  • Actualmente las tablas externas no pueden ser clonadas.
  • La tabla clonada tiene su propio ciclo de vida con lo que no tiene acceso a los datos históricos de la tabla origen utilizando time travel.
  • Una tabla clonada no incluye el historial de carga(LOAD_HISTORY) de la tabla de origen.
  • Si se clona una tabla con una secuencia asignada como valor por defecto a una columna, ésta seguirá referenciando a la secuencia original. En el caso de clonación de base de datos o esquemas que contengan tanto la secuencia como la tabla, la columna referenciará a la secuencia clonada (esto lo veremos con un ejemplo en la parte de clonado de Esquemas y Bases de dato)
  • Si clonamos una tabla que contiene a una foreign key, esta seguirá haciendo referencia a la tabla con al primary key. Como pasaba en el caso de las secuencias, si la clonación se realiza sobre un esquema o una base de datos y contiene ambas tablas, las referencias se realizan sobre las clonadas. En el caso de que la referencia de la foreign key sea sobre otra base de datos, seguirá realizándose sobre la tabla que contiene la primary key.

¿Cómo realizar la clonación de Esquemas y Base de datos?

Privilegios: Para poder clonar una base de datos o un esquema en Snowflake el role que va a realizar la operación tiene que tener permisos USAGE sobre los objetos que se van a clonar y los privilegios adecuados para la creación de los objetos en el destino.

La realización de clonado de un esquema o de una base de datos se realiza de manera recursiva, clonando todos los objetos hijos con la única excepción de las tablas externas , stages internos y snowpipes internos que no serán clonados.

A diferencia de la clonación de tablas, cuando se realiza la clonación de un esquema o una base de datos todos los permisos son heredados, por tanto, todos los objetos de la base de datos o del esquema clonado tendrán los mismos privilegios que tenían en el original.

Sentencia

USE ROLE ACCOUNTADMIN;
USE DATABASE WEATHER;
CREATE SCHEMA HISTORICAL_CLON CLONE HISTORICAL;

Al igual que sucedía con la clonación de tablas, la operación de clonado se realiza únicamente sobre la metadata, lo que permite que se realice en un tiempo reducido y sin necesidad de tener un virtual warehouse activo.

Para comprobar que la clonación se ha realizado de la forma esperada, podemos observar los objetos de cada una de los esquemas. Comprobamos tablas e internal stages.

SHOW TABLES IN HISTORICAL;

SHOW TABLES IN HISTORICAL_CLON;

Observamos que las tablas del esquema original y del clonado son iguales, además, se han heredado tanto owner como resto de propiedades. Como sucedía en el caso de la clonación de tablas, automatic_clustering está desactivado en las tablas del esquema clonado.

A continuación, vamos a comprobar que los internal stage del esquema original no se han clonado en el nuevo esquema

SHOW STAGES IN HISTORICAL;

SHOW STAGES IN HISTORICAL_CLON;

Clonación usando time travel

Como sucedía con el clonado de tablas, Snowflake también permite realizar el clonado de bases de datos y esquemas usando la opción de time travel.

En este caso vamos a realizar la clonación de la base de datos en un tiempo anterior a la clonación del esquema “HISTORICAL” que hemos realizado anteriormente.

CREATE DATABASE WEATHER_CLONE CLONE WEATHER at (offset => -60*60);

SHOW SCHEMAS IN WEATHER;

SHOW SCHEMAS IN WEATHER_CLONE;

Podemos comprobar en la base de datos clonada que no se encuentra el esquema que hemos clonado anteriormente.

Secuencias y foreign key:

 Como se comentó anteriormente en el clonado de tablas, si se clona un esquema que contiene una tabla con una columna con un valor por defecto de una secuencia o una foreign key y están en el mismo esquema o base de datos, la referencia de la secuencia apuntará a la misma referencia en el esquema o base de datos clonada.

Se ha añadido al esquema “HISTORICAL” una tabla “event_temperature” que contiene una secuencia y una foreign key a otra tabla. Se realiza la clonación:

CREATE SCHEMA HISTORICAL_CLON_2 CLONE HISTORICAL;

Si se observa la definición de la table, podemos comprobar cómo se ha cambiado la referencia tanto de la secuencia como de la foreign key.

Consideraciones del clonado de esquemas y bases de datos

  • Para el clonado de esquemas y bases de datos hay que tener en cuenta las mismas consideraciones observadas en la parte de las tablas.
  • Cuando se clona una base de datos o un esquema que contiene tareas, las tareas del clon se suspenden de forma predeterminada.
  • La clonación es rápida, pero no instantánea, especialmente para objetos grandes. Por tanto, si se ejecutan comandos de DDL en los objetos de origen mientras la operación de clonación está en curso, es posible que los cambios no sean reflejados en objeto clonado.

Conclusiones

Como vimos también en los artículos anteriores, Snowflake nos ofrece muchas características avanzadas, es muy importante comprender el funcionamiento de cada una de ellas para poder sacar el máximo partido siendo este el objetivo principal de esta serie de artículos. En este caso, comprender correctamente el clonado de datos nos va a ayudar a poder utilizar esta característica de manera correcta cuando sea necesario como puede ser en la creación de entornos de prueba o en la realización de snapshot.

Finalmente, hay que destacar que Snowflake nos ofrece un potente mecanismo de clonado de objetos, permitiéndonos la clonación de una forma sencilla, apenas incurriendo en costes y sin duplicación de datos. Estas características pueden ser muy importantes cuando vayamos a seleccionar un datawarehouse para nuestro entorno analitico.

Navegación

Introducción

¿Qué es Zero-Copy Clone?

¿Cómo funciona?

¿Qué coste tiene?

¿Cómo se puede clonar una tabla?

Clonación usando time travel

Consideraciones del clonado de tablas

¿Cómo realizar la clonación de Esquemas y Base de datos?

Clonación usando time travel

Consideraciones del clonado de esquemas y bases de datos

Conclusiones

Autores

¿Quieres saber más de lo que ofrecemos y ver otros casos de éxito?
DESCUBRE BLUETAB

Roberto García Parra

Technical Delivery Manager

Gabriel Gallardo Ruiz

Enterprise Architect

SOLUCIONES, SOMOS EXPERTOS

DATA STRATEGY
DATA FABRIC
AUGMENTED ANALYTICS

Te puede interesar

Cómo preparar la certificación AWS Data Analytics – Specialty

noviembre 17, 2021
LEER MÁS

LA BANCA Y LA ERA DEL OPEN DATA

abril 19, 2023
LEER MÁS

Tenemos Plan B

septiembre 17, 2020
LEER MÁS

Desplegando una plataforma CI/CD escalable con Jenkins y Kubernetes

septiembre 22, 2021
LEER MÁS

Bluetab se incorporará a IBM

julio 9, 2021
LEER MÁS

Serverless Microservices

octubre 14, 2021
LEER MÁS

Publicado en: Blog, interes, Practices, Tech

Databricks sobre Azure – Una perspectiva de Arquitectura (parte 2)

marzo 24, 2022 by Bluetab

Databricks sobre Azure - Una perspectiva de arquitectura (parte 2)

Francisco Linaje

AWS Solutions Architect

Gabriel Gallardo Ruiz

Senior Data Architect

En esta segunda entrega nos centraremos en analizar los diferentes servicios que ofrece Databricks para asegurar el escalado de nuestros servicios y la recuperación ante fallas del sistema, así como otros aspectos relativos a la seguridad como encriptación de los datos tanto reposo como en tránsito.

Primera entrega (link):

  • Arquitectura alto nivel
  • Planes y tipos de carga de trabajo
  • Networking
  • Identidad y Gestión de accesos

Segunda entrega:

  • Disaster Recovery
  • Escalabilidad
  • Seguridad
  • Logging y monitorización

Glosario

  • All Purpose Compute: Diseñado para entornos colaborativos en los que se recurra de forma simultánea al clúster por parte de Data Engineers y Data Scientist
  • Azure Data Lake: Permite almacenar múltiples formatos de datos en un mismo lugar para su explotación y análisis, actualmente Azure dispone la versión Gen2 .
  • Azure Key Vault: Servicio administrado de Azure que permite el almacenamiento seguro de secretos.
  • Azure Virtual Network (VNET): Red virtual aislada lógicamente en Azure.
  • DBFS (Databricks File Systen): Sistema de archivos de Databricks que se monta sobre los sistema de archivos distribuido de los Cloud Providers.
  • Data Lake: Paradigma de almacenamiento distribuido de datos provenientes de multitud de fuentes y formatos, estructurados, semi estructurados y sin estructurar.
  • Identity Provider (IdP): Entidad que mantiene la información de identidad de los individuos dentro de una organización.
  • Infraestructura como código o IaC: gestión y aprovisionamiento de la infraestructura a partir de código declarativo.
  • Jobs Compute: Enfocado a procesos orquestados mediante pipelines gestionados por data engineers que puedan conllevar autoescalado en ciertas tareas
  • Jobs Light Compute: Diseñado para procesos cuya consecución no sea crítica y no conlleve una carga computacional muy elevada
  • Network Security Group o NSG: Especifican las reglas que regulan el tráfico de entrada y salida de la red y los clusters en Azure
  • Private Link: Permite el acceso privado (IP privada) a Azure PaaS a través de tu VNET, de la misma forma que los service endpoints el tráfico se enruta a través del backbone de Azure.
  • SQL Compute: Cluster reservados a queries para la visualización de la información almacenada en el Data Lake
  • Secret scope: Colección de secretos identificados por un nombre.
  • Secure Cluster Connectivity (SCC):  Comunicación a través de túnel inverso SSH entre Control Plane y cluster. Permite no tener puertos abiertos ni IPs públicas en las instancias.
  • Security Assertion Markup Language (SAML): Estándar abierto utilizado para la autenticación. Basado en XML, las aplicaciones web utilizan SAML para transferir datos de autenticación entre dos entidades, el Identity Provider y el servicio en cuestión.
  • Service endpoints: Componente de red que permite conectar una VNET con los diferentes servicios dentro de Azure a través de la propia red de Azure.
  • TLS/ TLS1.2 (Transport Layer Security): es un protocolo de cifrado y comunicación que proporciona comunicaciones seguras por una red, comúnmente Internet.
  • Workspace: Entorno compartido para acceder a todos los activos de Databricks. En este se organizan los diferentes objetos (notebooks, librerias, etc…) en carpetas y se administran los accesos a recursos computacionales como clusters y jobs.

Disaster Recovery

Entendemos por Disaster Recovery al conjunto de políticas, herramientas y procedimientos que permiten la recuperación de la infraestructura cuando el sistema en su conjunto cae, como por ejemplo una caída de una región de Azure.

No debemos confundir estas políticas y herramientas con las empleadas en materia de alta disponibilidad de nuestro sistema (mínimo nivel de servicios).

Para ello, cuando implementamos una solución en la nube, una de las principales preguntas que debemos plantearnos a la hora de diseñar e implementar nuestra solución es:

  • ¿Qué piezas son críticas en nuestro sistema?
  • ¿Qué daños pueden provocar en el servicio?
  • ¿Cómo puede el sistema adaptarse y recuperarse ante estos errores?

Dar respuesta a estas preguntas es de vital importancia si deseamos que nuestra solución pueda cumplir adecuadamente el estándar de calidad que hayamos planteado.

Para este punto debemos analizar en que ámbito de nuestra solución opera Databricks y que herramientas o pautas debemos seguir para que la plataforma pueda cumplir con su servicio.

Debemos recordar que Databricks ofrece soluciones en materia de transformación y almacenamiento de datos tanto batch como en streaming, utilizando Azure Blob storage como capa de persistencia de datos no estructurados, como asimismo diferentes herramientas relacionadas con orquestación de jobs o análisis ad-hoc de datos vía SQL como servicio de analitica. Por lo tanto en este punto veremos que diferentes herramientas pueden ser propuestas para sincronizar nuestros workspaces,activos/recursos involucrados entre nuestras regiones.


Conceptos DR

Para poder comprender que es Disaster Recovery, deberemos primero comprender dos conceptos importantes:

Recovery Point Objective (RPO)

Hace referencia a la cantidad de datos máxima pérdida (medida en minutos) aceptable después de una caída del sistema. En este caso al disponer de Azure Blob Storage como sistema de persistencia distribuido, el concepto aplicaría a los datos de usuario temporales almacenados por Databricks, como por ejemplo cambios realizados en nuestros notebooks.

Recovery Time Objective (RTO)

Entendemos por RTO al periodo de tiempo desde la caída del sistema hasta la recuperación del nivel de servicio marcado.

En la siguiente imagen, podemos observar ambos conceptos de una forma visual:

Esquema conceptual sobre los conceptos RPO/RTO (fuente: Databricks)

Es importante indicar que la corrupción existente en los datos no se verá mitigada por las políticas asociadas a DR, sin embargo Databricks ofrece Delta time travel como sistema de versionado.

Tipos de región y redundancia

Una vez comprendido los conceptos de RPO y RTO, deberemos comprender los diferentes tipos de regiones en los que operará nuestra solución:

  • Región primaria: Región principal donde opera el sistema de forma normal.
  • Región secundaria: Región alternativa que entrará en operativa en caso de caída de la región primaria.

En nuestro caso de uso, estamos implementando un workspace de Databricks, por lo tanto emplearemos como capa de persistencia principal Blob Storage. Este servicio ofrece diferentes posibilidades a la hora de replicar nuestros datos entre regiones, vamos a verlas.

Region primaria

  • Almacenamiento con redundancia local (LRS): se realizan tres copias síncronas dentro de una única ubicación física en la región primaria, reduciendo así el coste, pero afectando a la disponibilidad y durabilidad (once nueves) de los datos.
Diagrama de replicación LRS (fuente: Azure)
  • Almacenamiento con redundancia de zona (ZRS): copia síncrona de los datos en tres zonas de alta disponibilidad en la región primaria (doce nueves).
Diagrama de replicación ZRS (fuente: Azure)

Region primaria y secundaria

  • Almacenamiento con redundancia geográfica (GRS): Se realiza una copia LRS en la región primaria y secundaria.
Diagrama de replicación GRS (fuente: Azure)
  • Almacenamiento con redundancia de zona geográfica (GZRS): Se realiza una copia con ZRS en la región primaria y mediante LRS en la región secundaria.
Diagrama de replicación GZRS (fuente: Azure)

En ambos casos, el acceso a los datos en la región secundaria no estará disponible salvo activación de la opción de lectura RA.

Dadas estas configuraciones, en la siguiente imagen se pueden ver los escenarios planteados en los que nuestros datos dejarían de ser accesibles.

Disponibilidad y acceso al dato según la redundancia configurada (fuente: Azure)

Deberemos configurar el nivel de replicación y redundancia entre zonas con el fin de disponer de nuestros datos sincronizados y disponibles en las regiones secundarias con el fin de que estás puedan estar operativas.


Tipos de despliegue

Dentro de los tipos de despliegue, podemos encontrar diferentes combinaciones según la necesidad de respuesta y los costes que deseamos asumir por su disponibilidad.

  • Activo: Despliegue principal que ejecuta las funcionalidad y servicios propios del sistema.
  • Pasivo: Procesos que no operan en el despliegue principal y permanecen inactivos/pasivos hasta que el despliegue activo deje de funcionar por una caída.

Es posible encontrar combinaciones de estos: activo-pasivo, activo-activo. De forma general:

Backup Restore
Es la estrategia más económica y lenta que podemos implementar. El objetivo principal es tener un conjunto de puntos de restauración en ambas regiones que podamos emplear para recuperar el servicio, sin necesidad de aprovisionar elementos core del sistema en otras regiones. 

Pilot Light 
Las piezas más importantes de nuestro sistema se encuentran desplegadas de forma activa pero bajo mínimos dentro de nuestra región secundaria, de forma que ante una caída del sistema los servicios principales podrían estar operativos y podrían escalarse de forma gradual (activo-pasivo).

Warn Standby
Estaríamos en un escenario muy similar a Pilot Light pero donde no solo tendríamos activos nuestros sistemas principales sino también una buena parte de los secundarios funcionando bajo mínimos pero listos para ser escalados (activo-pasivo).

Multi-site
Este plan ofrece el mayor grado de respuesta ya que implica disponer de forma activa todas nuestras piezas en una región secundaria, listas para dar servicio en caso de caída de la región principal (activo-activo)

Deberemos elegir la estrategia que mejor se adapte a nuestro caso de uso que dependerá principalmente del nivel de respuesta y coste asumible.


Workflow típico de recuperación

Dentro de los diferentes procedimientos, encontramos la estrategia activa-pasiva como la solución más sencilla y barata pero a la vez efectiva a la hora de ofrecer respuesta y servicio en el caso donde tras una caída del sistema en la región principal, el sistema pasivo entra en funcionamiento dando soporte al servicio.

La estrategia podría ser implementada de forma unificada para toda la organización o por grupos/departamentos de forma independiente basados en sus propias reglas y procedimientos.

De una forma global nos encontraremos que el procedimientos típico a alto nivel sería el siguiente:

  • Caída de un servicio crítico en la región primaria: red, origen de datos, etc
  • Se levanta el servicio en la segunda región si ésta no está afectada.
    • Se deben parar todas las actividades relacionadas con el workspace que sigan en funcionamiento en la región primaria y realizar un backup de los cambios recientes si es posible.
    • Se inicia el proceso de recuperación de los servicios sobre la región secundaria. Actualizando el enrutamiento y direcciones de dominio a la nueva región.
  • Se verifica que el servicio funciona correctamente y con normalidad.
  • En algún punto, la incidencia en la región primaria se ve resuelta y los servicios de Azure vuelven a un funcionamiento normal. Por lo tanto se deberá restablecer el sistema sobre la región primaria.
    • De forma idéntica al punto 2.a se deben parar todos los servicios y cargas de trabajo en la región secundaria.
    • Además se deben de volver a actualizar el enrutamiento y las direcciones de dominio a la región primaria.
    • Por último se debe de realizar un backup de los datos generados durante la caída de la región primaria para ser replicados en esta.
  • Finalmente se verifica que el servicio vuelva a funcionar correctamente y con normalidad en la región primaria.

Una vez nos hacemos una idea general de como sería un workflow típico de recuperación activo-pasivo, estudiaremos como podemos aplicarlo dentro de Databricks en nuestros workspaces.


Disaster Recovery en Azure Databricks

Databricks como plataforma de Data Analytics, tiene los datos como principal activo. Por ello se deben de definir las estrategias que permitan no solo poder seguir operando los servicios de la plataforma y workflows productivos en la región de soporte, sino la estrategia que permita generar consistencia en la propia replicación de los diferentes data sources.

En la siguiente imagen se especifican a modo de diagrama los diferentes activos que se verían involucrados en la replicación del plano de control o de datos.

Estrategia y herramientas en la sincronización.

Una vez realizado un análisis de nuestro sistema, deberemos analizar pieza por pieza como podemos realizar el procedimiento de réplica y sincronización.

Existen dos principales estrategias:

  • Un cliente que sincroniza los datos productivos y activos de la región primaria a la secundaria en un flujo programado.
  • Herramientas de integración/despliegue continuo (CI/CD) para el despliegue de forma paralela de la infraestructura, código y otros recursos principales del sistema en ambas regiones, de forma que la región secundaria se encuentre sincronizada con todos los cambios y desarrollos para ser operativa en caso necesario.
Esquema de sincronización vía CI/CD (fuente: Databricks)

Herramientas

Databricks ofrece en la siguiente tabla un resumen del conjunto de estrategias que se podrían aplicar según el recurso/activo involucrado de nuestro workspace. 

Es importante señalar que a día de hoy no hay ningún servicio oficial por parte de Databricks que permita administrar e implementar una política activa-pasiva de los workspaces en Azure.

Herramientas de replicación
FEATURE
Sync Client
CI/CD
Código fuente, notebooks, librerías
Sincronización con la región secundaria
Despliegue en ambas regiones
Usuarios y grupos
Empleo SCIM para la sincronización en ambas regiones
Control de los metadatos de los usuarios y grupos a través de GIT.
Configuración de los pools
Empleo del CLI o API para la creación en la segunda región
Empleo de templates. Configurar la región secundara con min_idle_instances a 0
Configuración de los jobs
Empleo del CLI o API para la sincronización con la segunda región
Empleo de templates. Configurar la región secundaria con concurrencia a 0
ACLs
Mediante la API de Permisos 2.0 es posible replicar los controles de acceso sobre los recursos copiados
Empleo de templates.
Librerias
DBFS
Repositorio central
Scripts de inicialización del cluster
Replicar de una región a otra a través del almacenamiento en el workspace
Repositorio central
Metadata
Incluir las DDL en el código fuente.
Secretos
Replicacion via API o CLI en el momento de creación
Configuraciones del cluster
Replicacion via API o CLI en el momento de creación
Empleo de templates en GIT.
Permisos de Notebooks, jobs y directorios
Replicación mediante la API de Permisos 2.0
Empleo de templates en GIT.

Implementación

Una vez, tenemos clara nuestra estrategia deberemos estudiar como podemos implementarla, para ello disponemos un conjunto de herramientas que van desde IaC, librerías de sincronización de data sources y migración de workspaces. Sin embargo, ninguna de las librerías de sincronizado/migración es oficial y aún se encuentran en desarrollo.

  • Módulo Databricks de Terraform [1]: para replicar la infraestructura, workspaces, metadatos, etc
  • Databricks Workspace Migration Tools [2]: paquete de librerías para generar un punto de restauración y migración de nuestros workspaces en otras regiones e incluso otros proveedores cloud.
  • Databricks Sync (DBSync) [3]: especializado en la sincronización, creación de copias de seguridad y  restauración de workspaces.

Escalabilidad

En este punto, veremos las diferentes opciones que ofrece Databricks en materia de escalabilidad, debido a que este punto ya ha sido tratado profundamente por nuestros compañeros dentro de la entrada Databricks sobre AWS – Una perspectiva de arquitectura (parte 2), nos limitaremos a comentar las características equivalentes en Azure.

Auto Escalado de workers

De la misma forma que en AWS, Databricks ofrece sobre Azure la posibilidad de escalar horizontalmente de una forma dinámica el número de workers dependiendo el mínimo y máximo que hayamos definido, permitiendo mejorar el tiempo de los trabajos sin sobre asignar recursos y por lo tanto reduciendo el coste global por trabajo en hasta un 30%.

Por lo general, en la forma tradicional cuando se definían las políticas de escalado para nuestros clusters se tenían que establecer una serie de umbrales estáticos donde si estos son rebasados se aprovisionan recursos extra, en forma de nodos de cómputo de bajo coste y efímeros (Spot). En muchos casos el escalado in/out de estos recursos no es lo suficientemente rápido, generando una ralentización global del job y una utilización subóptima de los recursos.

Para ello Databricks propone un  nuevo tipo de escalado optimizado [6], donde a partir de la información de los ejecutores es capaz de adaptar rápidamente los recursos del trabajo a sus necesidades de una forma rápida y eficiente, sin necesidad de esperar a que el trabajo completo termine para comenzar el desescalado.

Auto escalado tradicional: Ejecutores activos vs total de ejecutores (fuente: Databricks)
Auto escalado optimizado de Databricks: Ejecutores activos vs total de ejecutores (fuente: Databricks)

Caracteristicas:

  • Posibilidad de escalado desde el mínimo al máximo en dos pasos.
  • Posibilidad de desescalado aun cuando el cluster no está en idle viendo el shuffle file
  • Desescalado en base al porcentaje de nodos trabajando
  • En cluster del tipo job, el desescalado puede producirse si estos están infrautilizados tras 40 segundos, en all-purpose tras 150 segundos. 
  • Posibilidad de configurar la frecuencia de escalado mediante la propiedad spark.databricks.agressiveWindowDownS


Pools

Para reducir al máximo el tiempo de lanzamiento de una nueva instancia, Databricks permite mantener un set de clusters o pool pre-inicializado en estado idle listo para su empleo en nuestros trabajos o en los procesos de escalado. Si se llega al caso de que todo el pool de instancias se ha consumido, de forma automática se asignarán nuevas instancias al pool.

De la misma forma al escalado de los clusters, podremos definir un número máximo y mínimo de instancias que el pool podrá tener en estado idle para su posterior asignación al trabajo demandante y el tiempo que estas pueden permanecer desasignadas hasta su eliminación.

Respecto al tipo de instancias asignado al pool, no podrán cambiarse, tanto el driver como los workers del trabajo consumirán el mismo tipo de instancias.


Auto escalado del almacenamiento

Databricks ofrece la posibilidad de asignar un auto escalado en el almacenamiento local en disco del cluster con el fin de acotar la necesidad de dimensionado de estos. 

Databricks monitoriza el espacio libre en el disco de forma que en caso necesario se montará un disco externo sobre éste. Es importante señalar que estos discos una vez asignados no podrán desmontarse hasta que el cluster no sea eliminado, por ello se recomienda emplearlos en instancias Spot o que en instancias tengan una política de auto finalizado

Seguridad

Encriptación de datos databricks

Uno de los aspectos más importantes cuando vamos a seleccionar una plataforma para  el tratamiento de datos es la seguridad de los mismos. Debe ofrecer mecanismos de encriptación de datos tanto en los sistemas de almacenamiento, comúnmente conocido como datos en reposo (at rest), como cuando están en movimiento (in-transit).


En transito

Databricks encripta todos los datos que circulan por cada uno de sus diferentes componentes  y orígenes con TLS.  Además de la encriptación de datos, se encriptan con TLS todas las comunicaciones que se realizan entre el plano de control y el plano de datos, por tanto los comandos, consultas y meta-data viajan también encriptados.

Para plataformas que requieran un nivel alto de protección, se puede realizar la encriptación entre los nodos del cluster utilizando la encriptación RPC de Spark [7]. Está se realiza con cifrado AES de 128 bits a través de una conexión TLS 1.2. Está opción solo está disponible con el plan premiun y es necesario establecer los parámetros de configuración de Spark en el script de init del cluster o en el global si necesitamos que se aplique a todos los cluster del workspace. Es importante que tengamos en cuenta que la encriptación entre los nodos del cluster puede suponer una disminución en el rendimiento de los procesos y dado que la red privada de los nodos suele estar aislada, en la mayoría de los casos no será necesario este tipo de encriptación.

 
En reposo

Para el cifrado de los datos en reposo se utiliza SSE [8] (server-side encryption), cifra automáticamente los datos cuando se guardan en el almacenamiento distribuido (blob storage, ADLS y ADLS2).

Por defecto DBFS está encriptado usando claves administradas por Microsoft pero también permite la opción de usar claves administradas por el cliente, comúnmente conocidas como (CMK), permitiendo de este modo utilizar tu propia clave de cifrado para cifrar la cuenta de almacenamiento del DBFS. Además, tanto si se usa clave administradas como tu propia clave, también se ofrece la posibilidad de una capa adicional de cifrado utilizando un algoritmo/modo de cifrado diferente en la capa de infraestructura utilizando claves de cifrado administradas por la plataforma.

Para tener un completo cifrado de los datos en reposo, además del cifrado datos en el almacenamiento distribuido, se puede habilitar la encriptación de los disco locales de los nodos del clúster con lo que se permite la encriptación de los datos temporales que se guardan en las ejecuciones de los procesos. Actualmente está característica se encuentra en en versión preliminar pública y sólo está disponible para la creación del cluster desde el api REST utilizando la configuración siguiente:

{"enable_local_disk_encryption": true} 

También hay que tener en cuenta que activar esta opción puede suponer cierto impacto en el rendimiento de los procesos.

Diagrama de Arquitectura Databricks (fuente: Azure)

Logging

Para el correcto gobierno de una plataforma de ejecución de datos es necesario disponer de las herramientas necesarias para poder realizar el seguimiento y comprobación de ejecución de los workloads. Databricks integra en su plataforma todos elementos necesarios para realizar el mismo en un entorno de Spark. A continuación, vamos a resumir las opciones que integra Databricks out of the box aunque se pueden realizar monitorizaciones más avanzadas utilizando otras herramientas o servicios.

 

Cluster logs

Para cada uno de los cluster o job cluster creados en la plataforma podemos consultar de forma visual:

Captura Workspace - Sección Clusters

Event log: Se muestran todos los eventos relacionados con el ciclo de vida del cluster que han sucedido, como pueden ser, creación, terminación, cambios en la configuración…

Spark UI: Permite el acceso a la GUI ofrecida por Spark. Esta GUI es fundamental para poder detectar y solventar los problemas de performance en las aplicaciones de Spark.

Driver Logs : Permite ver los logs de ejecución tanto de la salida estándar , error y  log4j. Databricks también permite que se realice el volcado de logs en un filesystem determinado, para ellos es necesario configurarlo en las opciones avanzadas del cluster o indicándolo en la creación del cluster si se realiza desde crea desde API o CLI.

Metrics: Databricks proporciona acceso a Ganglia Metrics para obtener un mayor detalle del rendimiento que está ofreciendo el cluster

Captura Workspace - Ganglia Metrics

Registro de diagnóstico en Azure Databricks 

Azure Databricks nos ofrece la posibilidad de descargar los registros de las actividades realizadas por los usuarios a través del registro de diagnóstico [9]. Activando esta opción se enviarán los registros de la actividad de usuario a un destino seleccionado, Azure tiene disponibles 3 opciones para el envío de los registros: Cuenta de Almacenamiento, Event y Log Analytics.

Estos son los servicios que se pueden seleccionar para obtener registros de diagnóstico.

SERVICIOS DISPONIBLES PARA DIAGNÓSTICO
DBFS
sqlanalytics
modelRegistry
clusters
genie
repos
accounts
globalInitScripts
unityCatalog
jobs
iamRole
instancePools
notebook
mlflowExperiment
deltaPipelines
ssh
featureStore
sqlPermissions
workspace
RemoteHistoryService
databrickssql
secrets
mlflowAcledArtifact

La activación se puede realizar desde Azure Portal, API REST, CLI, ó powershell. Los registros están disponibles en un plazo de 15 minutos después de la activación.

Este sería el esquema de un registro de diagnóstico de salida 

Campo Descripción
operationversion
Versión del esquema del formato del registro de diagnóstico.
time
Marca de tiempo UTC de la acción.
properties.sourceIPAddress
Dirección IP de la solicitud de origen.
properties.userAgent
Explorador o cliente de API usado para realizar la solicitud.
properties.sessionId
Identificador de sesión de la acción.
identities

Información sobre el usuario que realiza las solicitudes:
* * : dirección de correo electrónico del usuario.

category
Servicio que registró la solicitud.
operationName
La acción, como el inicio de sesión, el cierre de sesión, la lectura, la escritura, etc.
properties.requestId
Identificador de solicitud único.
properties.requestParams

Pares clave-valor de parámetro usados en el evento.

El requestParams campo está sujeto a truncamiento. Si el tamaño de su representación JSON supera los 100 KB, los valores se truncan ... truncated y la cadena se anexa a las entradas truncadas. En raras ocasiones, cuando un mapa truncado sigue siendo mayor que 100 KB, TRUNCATED en su lugar hay una sola clave con un valor vacío.

properties.response
Respuesta a la solicitud: * * : mensaje de error si se ha producido un error. * * : resultado de la solicitud. * * : código de estado HTTP que indica si la solicitud se realiza correctamente o no.
properties.logId
Identificador único de los mensa jes de registro.

Tabla Esquema Registro Salida (fuente: Azure)

Para la explotación de los registros, si se ha seleccionado la opción de Logs Analytics, podremos explotarlos de forma sencilla utilizando Azure Monitor. Pero si lo que se desea es explotar estos registros con cualquier otra plataforma, servicio o herramienta es posible tomando estos registros JSON del lugar del envio seleccionando en la activación.

Referencias

[1] Databricks Terraform Provider. [link]

[2] Databricks Workspace Migration Tools. [link]

[3] Databricks Sync. [link]

[4] Databricks Disaster Recovery [link]

[5] Cifrado entre nodos de trabajo[link]

[6] Optimized AutoScaling [link]

[7] Spark Security [link]

[8] Azure encriptación discos [link]

[9] Registro de diagnostico [link]

Navegación

Glosario

Disaster Recovery

Escalabilidad

Seguridad

Logging

Referencias

Autores

¿Quieres saber más de lo que ofrecemos y ver otros casos de éxito?
DESCUBRE BLUETAB

Francisco Linaje

AWS Solutions Architect

Gabriel Gallardo Ruiz

Senior Data Architect

SOLUCIONES, SOMOS EXPERTOS

DATA STRATEGY
DATA FABRIC
AUGMENTED ANALYTICS

Te puede interesar

Conceptos básicos de AWS Glue

julio 22, 2020
LEER MÁS

CDKTF: Otro paso en el viaje del DevOps, introducción y beneficios.

mayo 9, 2023
LEER MÁS

Databricks sobre AWS – Una perspectiva de arquitectura (parte 2)

enero 27, 2022
LEER MÁS

¿Existe el Azar?

noviembre 10, 2021
LEER MÁS

Hacia un Gobierno del Dato 2.0: Del Gobierno del Dato al Gobierno de los Productos de datos

enero 31, 2022
LEER MÁS

Starburst: Construyendo un futuro basado en datos.

mayo 25, 2023
LEER MÁS

Publicado en: Blog, interes, Practices, Tech

Footer

LegalPrivacidadPolítica de cookies

Patrono

Patrocinador

© 2022 Bluetab Solutions Group, SL. All rights reserved.