Cuando hablamos de Data mesh podemos decir que se ha trasladado la filosofía de la arquitectura de microservicios de software al mundo de los datos. Al igual que en el desarrollo software los microservicios son entidades independientes que se exponen para ser consumidos sin mostrar sus entresijos, en Data mesh los Data as a Product (DaaP) se convierten en microservicios de datos para su consumo por otros equipos en la organización en forma de APIs o BBDD. Podríamos decir que los Data products son datamarts vitaminados con dos grandes diferencias respecto a los tradicionales: los encargados de crearlos y mantenerlos son las unidades de negocio a las que pertenecen los datos, no equipos técnicos centrales; y por otro lado, son entidades autónomas que integran la seguridad, reglas de calidad, gobierno, consumo, etc. Para asumir esta nueva responsabilidad, los áreas de negocio se dotan de recursos técnicos con conocimientos en data capaces de crear y mantener los productos de datos.

El Data Mesh se inspira en principios similares al «Domain Driven Design» DDD al enfatizar la importancia de comprender a fondo el dominio o área de negocio que se está tratando de abordar. En lugar de tratar exclusivamente los datos como un recurso aislado, el Data Mesh reconoce la necesidad de involucrar a expertos en el dominio y diseñar soluciones de datos que se ajusten a las necesidades y desafíos específicos de ese dominio. Data Mesh se alinea con una perspectiva «domain-driven», pero no es una implementación directa del Domain Driven Design.

La gestión de datos ha experimentado una gran evolución en las últimas décadas, y esto ha sido posible gracias a la creciente cantidad de información generada y al surgimiento de nuevas tecnologías.

Data Warehouse (finales de 1980) fue uno de los primeros modelos de arquitectura de datos en surgir. Se trata de un sistema centralizado diseñado para almacenar y gestionar grandes cantidades de datos relacionales, permitiendo la consolidación y el análisis de información de diferentes fuentes. Sin embargo, este modelo se enfrenta a limitaciones, como la complejidad y el tiempo requerido para la integración de nuevos datos.

El surgimiento de Data Lake (2000) trajo un nuevo enfoque en la arquitectura de datos. Este modelo se basa en el almacenamiento no estructurado de datos en su formato original, lo que permite una mayor flexibilidad y escalabilidad. Además, un Data Lake permite un procesamiento en paralelo de grandes cantidades de datos, mejorando la velocidad de procesamiento y la eficiencia. La evolución dentro de este paradigma fue trasladar la infraestructura on-premise a la nube, por lo que hablaríamos de Cloud Data Lake (2011).

El concepto de Data Lakehouse (finales del 2010) surge como híbrido entre los Data Warehouse y Data Lake. Este modelo combina la capacidad de gestión, estructura, gobierno y análisis de datos relacionales de los Data Warehouse con la capacidad, flexibilidad y escalabilidad de los Data Lake. Se crea una capa semántica entre los datos estructurados y no-estructurados almacenados en un repositorio (data lake) para consumirlos como si se tratase de una DWH.

Data Mesh (2019) es un enfoque más reciente en la gestión de datos que se basa en la descentralización y la colaboración. Este modelo promueve la gestión de datos como un producto, con equipos de negocio responsables de sus propios conjuntos de datos, otorgando mayor flexibilidad, escalabilidad en la gestión de datos y reduciendo el time-to-market de las iniciativas de datos al desaparecer las dependencias con otros actores (equipos centrales de datos). A finales de esta década también surge otro paradigma, Data Fabric, que a diferencia de Data Mesh, da continuidad a la arquitectura de datos centralizada pero poniendo foco en la tecnología que facilita el consumo de los datos implementando una capa de abstracción por encima de todos los orígenes de datos (virtualización de datos).

En conclusión, la arquitectura de datos ha evolucionado desde modelos centralizados y estructurados hasta enfoques más flexibles y descentralizados. Cada modelo tiene sus fortalezas y limitaciones, y la elección del modelo adecuado dependerá de las necesidades y objetivos específicos de cada organización. Sin embargo, es seguro decir que la arquitectura de datos continuará evolucionando a medida que surjan nuevas tecnologías y desafíos.

Data Mesh

En un Data Mesh los datos se organizan en Data Domains que representan áreas de negocio específicas, cada uno representa un conjunto de datos relacionados con un área de negocio o función específica de la organización. Por ejemplo, en un banco, podríamos tener Data Domains para transacciones bancarias, préstamos, seguridad, marketing, etc. Cada dominio es gestionado por un equipo de dominio autónomo, responsable de la calidad y disponibilidad de los datos compuesto por expertos en el área y responsables de la recopilación, transformación y gestión de los datos dentro de ese dominio. Estos equipos son independientes y tienen la propiedad de sus datos. Tienen la responsabilidad de garantizar la calidad y la disponibilidad de los datos en su dominio. Crean Data Products, subconjuntos de datos listos para su consumo que se presentan en forma de APIs, bases de datos, o servicios. Los Data Products son diseñados para ser utilizados por otros equipos en la organización. Para facilitar el acceso y su consumo se pueden implementar capas de servicios de datos. Estos servicios proporcionan una interfaz estandarizada para acceder a los datos en los Data Domains. Los equipos de dominio pueden exponer sus Data Products a través de estos servicios con una infraestructura que puede incluir tecnologías como Data Lakes, Data Warehouses, sistemas de streaming y es compartida y administrada de manera centralizada. La gobernanza y seguridad de los datos se aplican de manera uniforme en toda la organización para garantizar la integridad y privacidad de los datos.

Por ejemplo, en una empresa de comercio electrónico divide sus datos en Data Domains, como historiales de compras, preferencias de los clientes, inventario, y análisis de marketing. Cada Data Domain es gestionado por un equipo de dominio compuesto por expertos en esas áreas, quienes se encargan de recopilar, cocinar y gestionar los datos en su dominio. Estos equipos crean Data Products, como recomendaciones personalizadas para los clientes o análisis de tendencias de compras, que son compartidos a través de Data Services estandarizados para su uso en toda la organización. La infraestructura de datos subyacente se gestiona centralmente para garantizar la eficiencia y la seguridad, y las políticas de gobernanza y seguridad de datos se aplican de manera consistente en toda la empresa para proteger la integridad y la privacidad de la información.

Principios de Data Mesh

Una arquitectura Data Mesh orbita alrededor de estos cuatro principios:

1. Propiedad de datos descentralizada: Las áreas de negocio aka dominios tienen la propiedad y responsabilidad de los datos que generan y utilizan, lo que les permite tomar decisiones informadas sobre su uso y gestión. Influenciado por el concepto de Diseño orientado al Dominio o Domain-driven design (DDD). 
2. Diseño basado en productos: Los datos son tratados como productos aka Data as a Product (DaaP), con equipos de desarrollo de datos responsables de su calidad, seguridad y valor. Los DaaPs se exponen al resto de la organización para su consumo.
3. Plataforma de auto servicio: infraestructura transversal que articula la democratización del uso de los DaaPs, así como la gestión del ciclo de vida completo de cada uno. Este framework de componentes técnicos se gestiona desde un equipo central para facilitar la adopción del Data Mesh a las unidades de negocio.
4. Gobierno federado: crea una capa de gobierno federada entre los dominios con un equipo formado por representantes de cada uno. Equilibra entre la autonomía y agilidad de los dominios frente a la interoperabilidad del propio data mesh.

¿Que es un Data as a Product?

En lugar de tratar los datos como un recurso interno, los datos se entregan como un producto independiente por parte de cada data domain dedicado a su desarrollo y mantenimiento. Como adelantábamos en la introducción del artículo, se trata de microservicios de datos y se denominan Data as a Product o DaaP. Para que un conjunto de datos se considere un producto de datos debe cumplir los siguientes atributos:

1. Secure: Los datos deben estar protegidos contra accesos no autorizados y cumplir con las regulaciones y estándares de seguridad aplicables.
2. Discoverable: Los datos deben ser fácilmente descubribles y accesibles para los usuarios autorizados. Cualquier usuario debe ser capaz de poder buscar y encontrar los Data as a Product creados en la organización para identificar cual de ellos debe utilizar, o si se diera el caso, solicitar al data domain propietario de los datos que necesita, la creación de uno nuevo.
3. Addressable: Los datos deben tener una identidad única y una dirección para facilitar su acceso y uso.
4. Understandable: Los datos deben ser comprensibles para los usuarios y contener metadatos para facilitar su uso.
5. Trustworthy: Los datos deben ser precisos y confiables. Se pueden definir SLAs o como se refiere Zhamak en su literatura, SLOs (objetivos de nivel de servicio, o en inglés service level objectives) para determinar el nivel de disponibilidad y calidad de un Data as a Product. Por ejemplo, según la propia definición de Google, un SLO puede medir la latencia y la disponibilidad.
6. Natively accessible: Los datos deben ser fácilmente accesibles en su formato nativo, sin la necesidad de transformaciones previas.
7. Interoperable: Los datos deben ser interoperables con otros sistemas y formatos. Además, muchos Data as a Product consumidores se generan como DaaP derivados cruzando varios DaaP, por lo que es necesario que las claves o identificadores sean cross a toda la organización.
8. Valuable: Los datos deben tener un valor comprobado para la organización y ser utilizados para tomar decisiones y mejorar los procesos de negocio.

Comparativa entre arquitecturas de datos

Infraestructura de un Data Mesh

La plataforma donde se almacenan, transforman y distribuyen los datos entre los data domains es administrada de forma central por un equipo técnico. Los Data Domains solo deben preocuparse de explorar y trabajar los datos, no de cómo persistirlos o distribuirlos.

• Almacenamiento de datos: Los datos se almacenan en múltiples instancias independientes, cada una enfocada en un dominio específico. Estas instancias pueden ser bases de datos (con esquemas o datamarts específcos por dominio), data lakes (distribución por directorios) o sistemas de archivos.
• Servicios de datos: Cada instancia de datos proporciona servicios de datos independientes, como integración, seguridad y acceso, a través de una interfaz de programación de aplicaciones (API). Estos servicios pueden ser desarrollados y mantenidos por los equipos de datos correspondientes.
• Virtualización de datos: Se puede utilizar una capa de virtualización de datos para aislar los servicios de datos de los sistemas operativos subyacentes y permitir un acceso independiente a los datos. Esta capa puede ser proporcionada por herramientas como Denodo.
• Red de servicios: Los diferentes servicios de datos se comunican y colaboran entre sí a través de una red de servicios. Esta red puede ser implementada utilizando tecnologías como Kubernetes o Docker.
• Automatización: Se utilizan herramientas y procesos automatizados para garantizar la seguridad, calidad, cumplimiento y escalabilidad de la arquitectura.

Sin embargo, implementar un Data Mesh también presenta algunos desafíos. Uno de los principales es la necesidad de establecer una estrategia clara para la gestión de datos en la organización, establecer procesos y herramientas para garantizar la seguridad, calidad y cumplimiento, y asegurar una comunicación y colaboración efectiva entre los diferentes equipos de datos.

Desafíos de la arquitectura Data Mesh

Implementar un Data Mesh presenta varios desafíos, algunos de los cuales son:

• Cambio en la cultura organizacional: implica un cambio en la forma en que se gestionan los datos y en la estructura de la organización. Es importante involucrar a todos los departamentos y equipos en el proceso de cambio para garantizar su éxito. Se debe dedicar mucho esfuerzo a la gestión del cambio.
• Comunicación y colaboración entre equipos: La descentralización de los datos y la responsabilidad de su gestión entre múltiples equipos independientes requiere una buena comunicación y colaboración entre ellos. Es necesario establecer procesos y herramientas para garantizar una comunicación y colaboración efectiva.
• Seguridad y cumplimiento: Es importante establecer procesos y herramientas para garantizar la seguridad y cumplimiento de los datos en una arquitectura descentralizada.
• Escalabilidad: Es importante garantizar que la arquitectura sea escalable para manejar grandes cantidades de datos y poder crecer con las necesidades de la organización.
• Integración de los datos: Un Data Mesh implica la integración de múltiples fuentes de datos, lo que puede ser desafiante debido a la variedad de formatos y sistemas de origen de los datos.
• Capacitación y habilidades: Es necesario capacitar al equipo en las nuevas tecnologías y procesos necesarios para implementar un Data Mesh. Al distribuir la responsabilidad end-to-end de los datos a los dominios, éstos se deben dotar de recursos técnicos.
• Monitoreo y medición: Es importante establecer procesos y herramientas para monitorear y medir el rendimiento y la disponibilidad de los servicios de datos, así como detectar problemas.
• Cambio en los procesos de negocio: puede requerir cambios en los procesos de negocio existentes para aprovechar al máximo los servicios de datos independientes.

Implementación Data Mesh

Una cosa es la literatura de Zhamak en Data Mesh: Deliverig Data-Driven value at scale y otra la realidad. Os recomiendo esta lectura donde se desarrollan los motivos por los que una organización no estaría preparada para implantar una arquitectura Data Mesh. Como tener un greenfield para implementar una arquitectura Data Mesh es utópico, el único camino para no ahogarse en el intento es definir un roadmap e ir adaptando poco a poco el entorno. En nuestro caso, nos hemos juntado varios equipos técnicos y de negocio para diseñar un framework que sea el habilitador a la adopción de la nueva cultura de datos al resto de la organización. Va a nacer acoplado a los sistemas que ya hay implantados para ir poco a poco evolucionando a herramientas o servicios independientes e interoperables para todos los dominios.

Por ahora los DaaP se están generando y manteniendo desde este pequeño grupo, con la idea de que a medida que los dominios de datos asuman su gestión se doten de recursos técnicos especializados para asumir el mantenimiento y desarrollo. Tenemos hitos alcanzables, el objetivo no es hacer un big bang; queremos ir implementando cambios poco a poco y llevar de la mano a las demás áreas. Además del trabajo técnico y de diseño, esta transformación lleva consigo una inmensa tarea evangelizadora. Esto implica que a corto y medio plazo se tenga que asumir mucha deuda técnica. Algunos comentarios sobre cómo lo estamos abordando:

• Definición clara del roadmap, tanto en el corto plazo como el objetivo final ¿qué queremos conseguir? ¿cómo? y ¿cuándo?
• Esfuerzo inicial en la definición de un DaaP, qué consideramos un Data as a product y las propiedades que debe cumplir (ver ¿Qué es un data as a product?). Para ello hemos desarrollado las 6 características principales (seguro, descubrible, accesible, entendible, confiable e interoperable), dejando de lado las de valioso y nativamente accesible, ya que son inherentes al objetivo a su creación. Se ha trabajado en definir qué significa para nosotros cada uno de estos atributos y como llevarlo al mundo real. Por ejemplo, como organización ¿qué necesitamos para que un DaaP sea seguro? ¿qué mecanismos podemos construir para conseguirlo? en nuestro caso, hemos creado dos capas de seguridad definidas por roles de nuestro LDAP y sobre la BD donde se almacenan los data products como esquemas de base de datos.
• Se ha comenzado con DaaP pilotos. Es necesario partir el elefante en trocitos, por lo que se han identificado cuáles son los que podríamos empezar por resultar más sencillos de implementar y tener un impacto en negocio. Se han aprovechado las nuevas necesidades de datos en la organización para que nazcan ya como data products.
• Hemos definido varios tipos de DaaP en función de cuál es el origen y como se persisten los datos en base a nuestra arquitectura beta:
  • Tabla de BBDD
  • Modelo de estrella en BBDD (tabla de hechos y sus dimensiones)
  • Vista de BBDD
  • Entidades virtualizadas
• Seguimos manteniendo equipos de data centralizados encargados de hacer nacer los primeros DaaP y construir el framework de la plataforma de auto servicio.
• Los propietarios de los data products siguen siendo los equipos de desarrollo que están empujando la iniciativa. Los dominios aun no están lo suficientemente maduros para asumir el desarrollo, mantenimiento y en general el ciclo de vida de los DaaP.
• Los DaaPs se consumen a través de esquemas de BBDD (como si fueran datamarts). El objetivo a largo plazo no está claro, podría ser crear una capa de API REST como propone Zhamak. No sólo los propios datos, todo el ecosistema que define un DaaP podría ser accesible o consumible por microservicios. Por ejemplo, la gobernanza, metadatos, monitorización, etc. Se está explorando la virtualización de datos para resolver este paradigma.
• La seguridad de acceso a datos, la interoperabilidad, etc. por ahora se gestiona a nivel dominio activo y de configuración de base de datos (roles) ya que los datos están almacenados en datamarts de BBDD.
• Con el fin de dar servicio al descubrimiento, el catálogo de Data as a Products o data martketplace se está construyendo aprovechando las herramientas que ya están implementadas en la organización. En el catálogo se describe cada DaaP en detalle con las evidencias a los atributos descritos previamente: qué datos contiene, SLA/SLO, ejemplos, linaje, ownership, etc.

La entrada Data Mesh: gestión de datos descentralizada se publicó primero en Marcos Torregrosa.

Llegamos a la última semana del bootcamp para Data Engineers organizado por DataTalksClub. Si la semana pasada vimos el procesamiento de datos por lotes, en esta vamos a abordar el procesamiento en tiempo real con Apache Kafka. En primer lugar analizando cómo funciona internamente para luego practicar levantando un clúster Spark y Kafka en local con Docker o utilizando el SaaS de Confluence que ejecuta un clúster de Kafka en cloud.

• Repo Bootcamp: https://github.com/DataTalksClub/data-engineering-zoomcamp/tree/main/week_6_stream_processing
• Repo personal con ejercicios: https://github.com/hegdehog/data-engineering-zoomcamp-2023

Última actualización: 18/03/2023

Prerequisitos

En mi caso estoy trabajando en Windows, os recomiendo seguir la guía de instalación de los prerequesitos. Es importante tener en cuenta que si estamos usando Git bash o MINGW64 el catálogo de variables de entorno no se replica en Windows, por lo que debemos arrancar los cuadernos de jupyter desde el command shell de unix para que coja correctamente los path configurados.

• Apache Kafka: podemos instalarlo de forma manual siguiendo las instrucciones de la documentación oficial o bien levantar un contenedor de Docker preconfigurado. Puedes consultar cómo configurarlo y desplegarlo más adelante.
• Revisa las variables de entorno en Windows, además de tenerlas declaradas de forma independiente, es necesario añadirlas al PATH:
  • HADOOP_HOME=C:\tools\hadoop-3.2.0
  • JAVA_HOME=C:\Program Files\Java\jdk-11.0.17
  • SPARK_HOME=C:\tools\spark-3.3.2-bin-hadoop3
  • PYSPARK_PYTHON=python

Stream processing (procesamiento de datos en tiempo real)

El stream processing (procesamiento de datos en tiempo real) es un método que implica el procesamiento continuo de datos en tiempo real a medida que se generan y se transmiten a través de un flujo (stream) en lugar de procesarlos como un lote (batch) estático.

Los datos se procesan a medida que se generan y se transmiten, lo que significa que se pueden analizar y tomar decisiones en tiempo real. Esto es especialmente útil en aplicaciones donde la velocidad y la precisión son críticas, como el monitoreo de transacciones financieras, la detección de fraudes, la gestión de inventario, análisis de redes sociales y seguimiento de la cadena de suministro, entre otros.

La mayoría de los sistemas de stream processing están diseñados para ser escalables, tolerantes a fallos y distribuidos, lo que permite el procesamiento de grandes cantidades de datos en tiempo real en múltiples nodos o servidores. Algunas herramientas populares de stream processing incluyen Apache Kafka, Apache Flink, Apache Storm, Spark Streaming, y AWS Kinesis.

Apache Kafka

Apache Kafka es una plataforma de streaming de datos de código abierto desarrollada por la Apache Software Foundation. Se utiliza para la transmisión de datos en tiempo real a través de diferentes aplicaciones y sistemas, lo que permite a los usuarios procesar, analizar y almacenar grandes cantidades de datos en tiempo real.

Kafka se basa en el modelo de publicación-suscripción, donde los datos son enviados por los productores (producers) y recibidos por los consumidores (consumers) en tiempo real. Está diseñado para ser escalable y tolerante a fallos, lo que significa que es capaz de manejar grandes volúmenes de datos y puede continuar operando incluso si algunos nodos fallan. Ofrece una arquitectura distribuida, lo que significa que se puede implementar en múltiples servidores y se puede escalar horizontalmente según las necesidades. Además, cuenta con una API, lo que facilita su integración con diferentes sistemas y lenguajes de programación.

Productores y Consumidores

Los productores y los consumidores son los componentes principales de Kafka. Los productores son los encargados de enviar los datos a Kafka, mientras que los consumidores los reciben y los procesan. Es como si los productores fueran los que hablan y los consumidores los que escuchan. Los consumidores se pueden agrupar en grupos de consumidores (consumer.group.id) para leer un mismo topic.

Acks (confirmaciones) es una configuración de Apache Kafka que se refiere a la cantidad de réplicas de un mensaje que deben confirmar la recepción antes de que se considere que el mensaje ha sido correctamente procesado y se envíe una respuesta de confirmación al productor.

Cuando un productor envía un mensaje a un topic en Kafka, se envía a los brokers que son responsables de ese topic que tienen una o más réplicas del mismo. La configuración acks indica cuántas réplicas deben confirmar la recepción del mensaje antes de que el productor reciba una respuesta de confirmación. Hay tres valores posibles:

• 0: El productor no espera confirmación de recepción. Esto significa que el mensaje se envía al broker y se considera enviado.
• 1: El productor espera la confirmación de recepción de al menos un broker. Si uno confirma la recepción del mensaje se envía una respuesta de confirmación al productor.
• all» o -1: El productor espera la confirmación de recepción de todas las réplicas. Cuando han confirmado la recepción del mensaje, se envía una respuesta de confirmación al productor.

La configuración acks es importante porque afecta la disponibilidad y la durabilidad de los datos en Kafka. Si se establece en 0 se corre el riesgo de perder mensajes si se produce una incidencia en el broker antes de que se hayan replicado los datos. Si se establece en all se asegura la durabilidad de los datos, pero impacta en la entrega del mensaje debido a la necesidad de que todas las réplicas confirmen la recepción.

Brokers

Los brokers son los nodos de Kafka que se encargan de almacenar y distribuir los datos. Cada broker es capaz de almacenar uno o más topics y es responsable de asegurarse de que los datos sean entregados correctamente a los consumidores. Si un broker falla, los otros brokers pueden continuar operando sin interrupciones.

Clúster de Kafka

Un clúster de Kafka es un conjunto de servidores o brokers de Kafka que trabajan juntos para proporcionar una plataforma de streaming de datos distribuida y tolerante a fallos. Permite escalar la plataforma horizontalmente para procesar grandes cantidades de datos en tiempo real. Cada broker en el clúster de Kafka es responsable de almacenar y distribuir un subconjunto de los datos del clúster. Los productores envían mensajes a los brokers, que los almacenan en particiones y los distribuyen a los consumidores.

Uno de los broker se convierte en el controlador (controller) para coordinar la asignación de particiones a todos los brokers. El controlador supervisa el estado de los brokers y las particiones y se asegura de que las réplicas de las particiones estén distribuidas de manera uniforme entre los brokers. Por otro lado, utiliza grupos de consumidores (consumer groups) para equilibrar la carga de lectura de mensajes. Los consumidores de un grupo se asignan a diferentes particiones de un topic para leer los mensajes de forma paralela y procesarlos.

KRaft (sustituto de Zookeeper)

KRaft es un protocolo de consenso de Apache Kafka que fue introducido en la versión 2.4.0 (aunque se liberó oficialmente en la 3.1.1) y que tiene como objetivo reemplazar a ZooKeeper para la gestión de los metadatos del clúster de Kafka. ZooKeeper gestiona el estado del clúster y almacena los metadatos de los diferentes componentes, como los topics, particiones, brokers, etc. con la misión de gestionar su estado estado y coordinar los consumidores. Sin embargo, puede ser una fuente de problemas, ya que su arquitectura centralizada lo hace propenso a fallos y cuellos de botella.

Con el objetivo de mejorar la fiabilidad y escalabilidad del sistema, Kafka introdujo el protocolo KRaft, que permite almacenar los metadatos en los propios brokers de Kafka a través de un topic en lugar de depender de ZooKeeper. Kraft utiliza un sistema distribuido de almacenamiento de metadatos que permite a los brokers mantener una copia de los metadatos del clúster, lo que mejora la disponibilidad y elimina la necesidad de un sistema de gestión de estado centralizado como ZooKeeper.

Históricamente, el plano de control de Kafka se gestionaba a través de un servicio de consenso externo llamado ZooKeeper. Uno de los brokers se designa como controlador y es responsable de comunicarse con ZooKeeper y los demás brokers del clúster. Los metadatos del clúster se almacenan en ZooKeeper. En KRaft se designa un grupo de brokers como controladores, que proporcionan los servicios de consenso que solían ser proporcionados por ZooKeeper. Todos los metadatos del clúster se almacenan en topics de Kafka y se gestionan internamente.

Topics

Un topic es una categoría o flujo de mensajes que se almacenan en un clúster de Kafka. Los productores envían mensajes a un topic específico y los consumidores pueden leerlos. Cada topic se compone de uno o más logs, que son la persistencia física de los mensajes en una secuencia ordenada. Cada log se almacena en uno o más brokers. Un topic divide los mensajes en múltiples particiones, lo que permite distribuir la carga de datos entre los nodos del clúster (brokers). Cada mensaje tiene asignado un par clave-valor, un timestamp y se identifica con un offset, número único que indica la posición del mensaje en la partición. Los consumidores utilizan los offsets para saber qué mensajes ya han sido procesados y cuáles están pendientes.

Particiones

Cada topic puede tener una o varias particiones, que son secuencias ordenadas y duraderas de mensajes que se almacenan en los brokers de Kafka. Cada partición se asigna a un único broker y se denomina líder de la partición, que es el responsable de recibir y servir las solicitudes de L/E de los productores y consumidores para esa partición. Una partición sólo se puede asignar a un consumidor, por lo que lo ideal es que haya tantas particiones como consumidores.

Los mensajes se asignan a las particiones en un topic mediante su clave: las claves de los mensajes se hashean y se dividen a partes iguales por las particiones. Las particiones mejoran el rendimiento y son una herramienta de escalabilidad de los productores y consumidores de Kafka. Al tener varias particiones, se pueden escribir varios mensajes de forma concurrente entre varios brokers.

El Consumer Group id es un parámetro clave en Kafka, ya que determina cómo se distribuyen los mensajes entre los consumidores. Si varios consumidores pertenecen al mismo grupo y comparten la misma suscripción al mismo topic, Kafka garantiza que cada mensaje se entregue a un solo consumidor en el grupo, lo que ayuda a distribuir la carga de procesamiento de mensajes entre los consumidores y asegura que cada mensaje se procese exactamente una vez.

La elección de la clave de partición adecuada en Kafka es una tarea crítica para el rendimiento y la escalabilidad del sistema. A continuación, se presentan algunos factores a considerar:

1. Cardinalidad: la cardinalidad de la clave de partición es importante porque afecta la capacidad de distribuir uniformemente los mensajes a través de los diferentes consumidores. Si la clave de partición tiene baja cardinalidad, los mensajes se agruparán en un conjunto reducido de particiones, lo que puede generar un desequilibrio en la carga y un bajo rendimiento del sistema. Si la clave de partición tiene alta cardinalidad, los mensajes se distribuirán uniformemente en diferentes particiones.
2. Ordenación: si el sistema requiere que los mensajes se procesen en un orden específico, la clave de partición debe elegirse de tal manera que garantice que se entreguen en el orden correcto.
3. Distribución de carga: si el sistema tiene muchos consumidores , es importante elegir una clave de partición que permita una distribución uniforme de la carga a través de éstos. Si la carga está desequilibrada, algunos consumidores pueden estar sobrecargados, mientras que otros pueden estar inactivos.
4. Reutilización de particiones: es posible que se puedan reutilizar particiones para diferentes flujos de mensajes. En este caso, la clave de partición debe elegirse de tal manera que permita la reutilización de particiones en diferentes flujos de mensajes.

Replicación (Replication)

Kafka utiliza la replicación de particiones para garantizar la tolerancia a fallos y la alta disponibilidad de los datos. Cada partición tiene una o varias réplicas (seguidores o followers), que son copias exactas de la partición líder en otros brokers de Kafka. Las réplicas garantizan que los datos de la partición estén disponibles incluso si el broker donde está ubicado el líder de la partición falla, en tal caso, uno de los seguidores en otro broker se convierte en el nuevo líder.

El factor de replicación se configura a nivel de topic y configura la cantidad de copias de una partición que se deben almacenar en diferentes brokers para garantizar la tolerancia a fallos y la disponibilidad de los datos en caso de que falle un broker. Por ejemplo, si un topic tiene un factor de replicación de 2, cada partición tendrá un líder y dos réplicas. Si el líder falla, una de las réplicas se elegirá como el nuevo líder y continuará gestionando las L/E en la partición.

Un factor de replicación 1 implica que no habrá copias de las particiones, es decir, solo habrá una instancia de cada partición en el broker líder que la aloja. Si falla, esa partición no estará disponible hasta que se restaure el broker o se cree un nuevo líder para esa partición en otro. Esta casuística puede provocar una interrupción de servicio y pérdida de datos en caso de que la partición no esté respaldada en otra ubicación. En general, se recomienda utilizar un factor de replicación mayor que 1 para garantizar la tolerancia a fallos y la disponibilidad de los datos en caso de que falle un broker.

Es importante tener en cuenta que un factor de replicación muy alto significa que se necesita más espacio de almacenamiento y ancho de banda de red para replicar los datos en los diferentes brokers, lo que también impacta en el rendimiento y velocidad.

Política de retención (retention policy)

La política de retención en Kafka determina cómo se manejan los registros en un topic después de cierto tiempo o tamaño. En otras palabras, es la configuración que determina cuánto tiempo se deben mantener los mensajes antes de ser eliminados.

Hay dos tipos de políticas de retención en Kafka: basadas en tiempo y basadas en tamaño.

• Retención basada en tiempo: Los logs se eliminan después de un período de tiempo especificado, independientemente de su tamaño. Para habilitar esta política, se establece la propiedad de configuración log.retention.ms en el archivo de configuración de Kafka.
• Retención basada en tamaño: Los registros se eliminan del topic después de que se haya alcanzado un tamaño máximo especificado. Para habilitar esta política, se establece la propiedad de configuración log.retention.bytes.

Otros parámetros que podemos configurar de la política de retención son:

• retention.ms: este parámetro de configuración controla el tiempo máximo de retención de un log antes de que sea eliminado para liberar espacio. Si se indica a -1 no se aplica límite de tiempo (no se borran nunca los logs).
• cleanup.policy: determina la política de retención de los logs. Por defecto es delete y borra todos los logs una vez superado el tiempo designado en la política de retención. Otra opción es configurarlo como compact para comprimirlos. La última opción combinar ambas separadas por coma delete, compact, que elimina los logs antiguos y comprime los logs retenidos.

Offsets

__consumer_offsets es un topic interno que se utiliza para almacenar el estado de compensación de los consumidores de Kafka. Este topic almacena los offsets de los mensajes usados por cada consumidor de Kafka en un grupo de consumidores determinado.

Los offsets son números que indican la posición del último mensaje que un consumidor ha leído en un topic determinado. Almacenar los offsets en __consumer_offsets permite que los consumidores de Kafka puedan reanudar la lectura de los mensajes desde el último punto en que se detuvieron, incluso después de un reinicio del consumidor o un fallo del mismo. También se utiliza para mantener la coordinación entre los miembros de un grupo de consumidores de Kafka. Cuando un consumidor se une o abandona un grupo, se actualiza la información en __consumer_offsets para asegurar que los offsets de los mensajes se mantengan consistentes entre todos los miembros del grupo.

auto.offset.reset es una configuración en Apache Kafka que especifica qué sucede cuando un consumidor se une a un grupo de consumidores para leer mensajes de un topic y no tiene un offset válido para comenzar a leerlos (porque es la primera vez que se conecta o porque ha perdido su posición de lectura). Puede tener uno de los siguientes valores:

• earliest: Si no hay offset válido para el consumidor, el consumidor comenzará a leer mensajes desde el comienzo del topic.
• latest: el consumidor comenzará a leer mensajes desde el final del topic.
• none: se lanzará una excepción.

Es importante tener en cuenta que la configuración auto.offset.reset solo se aplica cuando un consumidor se une a un grupo de consumidores. Si un consumidor está leyendo desde un topic sin unirse a un grupo de consumidores, debe especificar el offset inicial de forma explícita.

Timestamps

Los timestamps se utilizan para registrar la hora en que se produce un evento en un topic de Kafka. Son importantes porque permiten ordenar los eventos en función del tiempo en que se produjeron, fundamental para procesar flujos de datos en tiempo real, ya que es necesario garantizar que los eventos se procesen en el orden correcto para generar resultados precisos, por ejemplo, si usamos funciones de ventana (windowing).

En Kafka, existen dos tipos de timestamps:

• Timestamp del productor: se asigna al evento cuando se escribe en un topic. Por defecto, Kafka utiliza el timestamp del sistema en el momento en que se escribe el evento, pero se puede especificar uno personalizado si es necesario.
• Timestamp del registro: se asigna al evento cuando se lee del topic. Por defecto, Kafka utiliza el del productor, pero se puede configurar para que utilice el del sistema en el momento en que se lee el evento.

Por último, existen tres conceptos de tiempo importantes:

• Event time: momento en que un evento realmente ocurrió en el mundo real. Se puede incluir como un campo en el registro de Kafka o se puede inferir a partir de otros campos, como la marca de tiempo del sistema del dispositivo que generó el evento.
• Processing time: momento en que un evento es procesado por una aplicación o un sistema de Kafka. Es decir, es el tiempo en que se procesó el evento. Es determinado por el sistema que está procesando el evento y se puede incluir como un campo en el registro de Kafka.
• Ingestion time: momento en que un evento es recibido por un sistema de Kafka, es decir, es el tiempo en que el evento llega al broker y se escribe en un topic. Se puede incluir como un campo en el registro de Kafka o se puede inferir a partir de otros campos, como la marca de tiempo del sistema del broker que recibió el evento.

Schema Registry

El Schema Registry es un componente opcional de Kafka que se utiliza para gestionar los esquemas en formato JSON de los datos que se envían (productores) y reciben (consumidores). Usar esquemas en Kafka es como tener una plantilla para el tipo de datos. Esto ayuda a prevenir errores y garantiza que los datos sean consistentes y estén en el formato correcto. Además, permite agregar nuevos campos o cambiar los tipos de datos sin romper la compatibilidad con aplicaciones antiguas. Usar esquemas en Kafka hace que sea más fácil enviar y recibir datos de manera confiable entre diferentes tecnologías y aplicaciones.

Cuando un productor envía un mensaje a un topic, Kafka puede validar el mensaje utilizando el esquema asociado. Si el mensaje es válido según el esquema, se procesa normalmente. Si el mensaje no cumple con el esquema, se rechaza o envia una notificación de error. Cuando un consumidor lee un mensaje de un tema, Kafka puede recuperar el esquema asociado y utilizarlo para deserializar el mensaje. Esto ayuda a garantizar que el mensaje se procese correctamente y que los datos estén en el formato correcto.

El Schema Registry también puede manejar la evolución de los esquemas. Si se actualiza un esquema, Kafka puede almacenar la versión anterior y garantizar que los consumidores que utilizan el esquema anterior aún puedan procesar los mensajes. Esto permite una evolución más flexible de los datos a medida que cambian con el tiempo.

Avro

Avro es un formato de serialización de datos que se utiliza para estructurar y almacenar datos. La serialización es el proceso de convertir un objeto o estructura de datos en un formato que se pueda transmitir o almacenar, y que luego se pueda volver a convertir en el objeto original. Es similar a JSON en el sentido de que ambos son formatos de texto que se pueden leer y escribir en varios lenguajes de programación. Sin embargo, hay algunas diferencias clave entre los dos. En primer lugar, Avro es más compacto que JSON, lo que significa que utiliza menos espacio de almacenamiento y ancho de banda. Además, Avro admite la serialización de datos complejos, como registros y uniones, mientras que JSON solo admite tipos de datos simples, como cadenas y números.

Una de las mayores ventajas de Avro es que se puede utilizar en diferentes lenguajes de programación. Esto significa que un mensaje serializado en Avro en un lenguaje de programación se puede deserializar en otro lenguaje sin problemas de compatibilidad. En otras palabras, si estás usando Java para enviar un mensaje a través de Kafka y otro equipo está usando Python para recibir ese mensaje, ambos equipos pueden leer el mensaje en Avro sin ningún problema. Otra ventaja de Avro es que se integra bien con el Schema Registry para garantizar que todos los mensajes que se envían a través de Kafka se ajusten a un esquema específico.

Cuando se utiliza Avro para serializar datos, es posible que los esquemas cambien con el tiempo, lo que puede causar problemas si no se manejan correctamente. La evolución de Avro gestiona los cambios en los esquemas de datos. Básicamente, Avro tiene un sistema que permite manejar estos cambios de manera eficiente, sin afectar la capacidad de los clientes para leer registros más antiguos. Por ejemplo, cuando se serializa un registro en Avro este incluye su propio esquema. Si éste cambia en el futuro, Avro utilizará el esquema almacenado en el registro para interpretar los datos, lo que garantiza que los datos se lean correctamente.

Además, Avro permite agregar nuevos campos a los esquemas sin romper la compatibilidad hacia atrás. Esto significa que se pueden agregar nuevos campos a los registros sin afectar la capacidad de los clientes más antiguos para leer registros más antiguos. Sin embargo, si se eliminan campos, esto puede romper la compatibilidad hacia atrás y puede requerir una conversión de esquema.

• La compatibilidad hacia atrás (backward compatibility) es la capacidad de nuevos consumidores de leer datos generados por productores antiguos. En otras palabras, si un consumidor más nuevo puede leer los datos que fueron producidos por un productor más antiguo sin problemas, entonces se dice que existe compatibilidad hacia atrás.
• La compatibilidad hacia adelante (forward compatibility) se refiere a la capacidad de productores más nuevos para producir datos que pueden ser leídos por consumidores más antiguos. Es decir, si un productor más nuevo puede producir datos que un consumidor más antiguo puede leer sin problemas, entonces se dice que existe compatibilidad hacia adelante.
• La compatibilidad mixta (mixed compatibility) se refiere a la capacidad de sistemas antiguos y nuevos para interactuar juntos de manera adecuada.

Kafka Connect

Kafka Connect es un framework que se utiliza para conectar Kafka con otros sistemas de forma que los usuarios pueden enviar y recibir datos desde y hacia Kafka utilizando conectores preconstruidos. Utiliza una arquitectura de plugin que permite a los usuarios conectar diferentes sistemas de origen y destino. Los plugins de origen son responsables de extraer los datos de la fuente y enviarlos a Kafka, mientras que los plugins de destino son responsables de recibir los datos y enviarlos al sistema de destino.

Proporciona una amplia gama de características que mejoran la usabilidad, escalabilidad y flexibilidad de la integración de datos:

1. Conectores preconstruidos: tiene una variedad de conectores preconstruidos para integrar diferentes sistemas, se incluyen conectores para bases de datos, sistemas de archivos, servicios web y más.
2. Configuración sencilla: utiliza una configuración simple en formato JSON para los conectores, lo que facilita el proceso y la personalización de los conectores.
3. Escalabilidad: altamente escalable y puede manejar grandes volúmenes de datos de manera eficiente. Además, admite la configuración de múltiples conectores para la misma tarea, lo que permite una mayor escalabilidad.
4. Tolerancia a fallos: está diseñado para ser tolerante a fallos. Si un proceso de Kafka Connect falla, otro proceso lo reemplaza automáticamente sin interrupción del flujo de datos.
5. Integración con Kafka: está diseñado para integrarse sin problemas con Kafka y utiliza el mismo sistema de particiones y replicación que Kafka para garantizar la tolerancia a fallos y la alta disponibilidad de los datos.

Kafka Streams

Kafka Streams es una herramienta open source para procesar y analizar datos en tiempo real mientras almacenados en Apache Kafka, donde se consumen datos de un topic de Kafka y son enviados a otro topic para su procesamiento. Permite realizar realizar operaciones de transformación, filtrado, combinación (join) y hacer cálculos en los flujos de datos en tiempo real para detectar patrones y tendencias. Es muy útil para aplicaciones que requieren análisis en tiempo real de grandes volúmenes de datos, como el análisis de datos de IoT, la detección de fraude, migración de datos o monitoreo y alertas.

Imagina Kafka Streams es como un cocinero que recibe datos de uno o varios lugares (producers), los mezcla y los prepara de una manera especial para después enviarlos a otro lugar (consumer). En este caso, los datos son como los ingredientes, los topics de Kafka son los lugares donde se encuentran estos ingredientes, y Kafka Streams es el chef que usa los ingredientes, los mezcla y transforma en algo diferente y sabroso para luego enviarlo a otro lugar donde pueda ser utilizado.

En Kafka Streams un thread es como un trabajador que se encarga de procesar datos de Kafka en tiempo real. Cada thread puede procesar varias tareas llamadas tasks» que son como paquetes de trabajo que contienen datos de una parte del topic de Kafka. Para procesar grandes cantidades de datos en tiempo real, se dividen las tareas en subconjuntos más pequeños y se distribuyen en varios threads. De esta manera, se pueden procesar varias tareas al mismo tiempo y de manera eficiente.

El número de threads y tasks se puede ajustar según las necesidades de procesamiento de la aplicación y los recursos disponibles. De esta manera, se puede asegurar que Kafka Streams tenga suficientes workers para procesar grandes cantidades de datos en tiempo real de manera rápida y eficiente.

Algunas de las principales funciones de Kafka Streams incluyen:

• Transformación de datos: transformar, filtrar y enriquecer flujos de datos en tiempo real.
• Procesamiento de ventanas de tiempo: analizar datos en ventanas de tiempo definidas, lo que facilita la detección de patrones y tendencias en los flujos de datos.
• Join: combinar flujos de datos de múltiples fuentes para obtener información más completa y precisa.
• Agregación: realizar cálculos en los flujos de datos en tiempo real, como el recuento de eventos o la sumatoria de valores.

Stream

Un stream es una corriente continua de datos que se actualiza constantemente y cuyo tamaño es desconocido o ilimitado. Es como un río que fluye y nunca se detiene. En Kafka Streams, una stream puede estar compuesto por una o más particiones que son secuencias de datos ordenados, inmutables, y que se pueden reproducir y recuperar en caso de fallos.

Cada registro de datos en un Stream es un par de valores llamado (key-value) donde key es una etiqueta que se utiliza para identificar el registro y value es el valor real que contiene el registro.

Processor Topology

El processor topology es un grafo formado por Streams (nodos) conectados por Streams processors (aristas) o la State Store que representa las distintas etapas de transformación de los datos en tiempo real de un topic de Kafka.

Los stream processors procesan un flujo de datos de entrada y producen uno de salida. Pueden realizar diversas operaciones en los datos, como filtrar, transformar, combinar o agregar. Cada stream processor puede recibir y generr uno o varios streams, lo que le permite realizar múltiples operaciones de procesamiento en los datos. Hay dos tipos de stream processors:

• Source processor: Es un stream processor especial que no tiene ningún nodo anterior, es el primer procesador que lee datos directamente de uno o varios topics.
• Sink processor: por contra, este tipo de stream processor sería el último de la topología. Se encarga de enviar todos los datos recibidos de los stream processor previos a un topic de Kafka.

Para trabajar con una topología de procesamiento de datos podemos utilizar la Processor API para trabajar a bajo nivel con mayor flexibilidad con los streams e interactuar con las state stores. Sin embargo, la opción más recomendada es utilizar la API Kafka Streams DSL (Domain Specific Language), que construida sobre el Processor API permite trabajar con los streams de una forma mucho más sencilla en muy pocas líneas de código, aunque menos personalizada.

KStreams vs KTables

La principal diferencia entre un kStream y una KTable (o State) es que el primero representa un flujo continuo de datos en tiempo real, mientras que una tabla es una vista instantánea de los datos almacenados en un momento dado. En otras palabras, los streams permiten procesar datos a medida que fluyen, mientras que las tablas permiten realizar consultas y análisis sobre los datos almacenados en un momento específico en el tiempo. Los streams pueden ser infinitos, lo que significa que pueden procesar y transmitir datos continuamente, mientras que las tablas tienen un tamaño finito y se actualizan en tiempo real a medida que fluyen los datos.

Las KTables se almacenan en la state store, una caché en memoria que se utiliza para proporcionar un acceso más rápido y eficiente a los datos. Cada instancia de la aplicación de Kafka Streams tiene su propia copia del state store en la memoria, lo que permite que cada instancia procese su propia partición de los datos. También admite la persistencia de KTables en un almacenamiento externo como un sistema de archivos o una base de datos, con lo que incluso podríamos acceder a los datos aunque nuestra aplicación ya no esté corriendo.

Transformaciones Stateful y Stateless

Algunas aplicaciones de stream processing no requieren estado (stateless), lo que significa que el procesamiento de un mensaje es independiente del resto mensajes y no necesitan almacenar datos adicionales en Kafka. Un ejemplo sería cuando sólo necesitas transformar un único mensaje a la vez o filtrar mensajes en función de alguna condición.

Por otro lado, nuestra aplicación necesitará estado (stateful) cuando realiza operaciones como unir, agregar o dividir registros que necesitan información de los registros previos, de forma que los datos del Stream son almacenados en kafka.

Uniones (join)

Los joins de streams son muy similares a los joins de tablas en bases de datos relacionales. Sin embargo, hay algunas diferencias importantes debido a la naturaleza de los flujos de datos en Kafka. En Kafka Streams, los joins de streams se realizan mediante la unión de dos o más streams que se producen en tiempo real a través de las claves de los eventos.

Los joins en Kafka Streams utilizan un modelo de tiempo de evento, es decir, los eventos se procesan en orden de llegada, en lugar de utilizar un modelo de tiempo de procesamiento. Esto garantiza que los joins se realicen de manera precisa y coherente a medida que llegan los eventos. Los joins de streams también pueden ser realizados con ventanas de tiempo (windowed joins), lo que permite unir streams en ventanas de tiempo específicas en lugar de en todo el flujo. Esto es útil en situaciones donde se quiere hacer uniones con datos históricos o en eventos que ocurrieron en un período de tiempo específico.

En general, hay dos tipos de joins de streams en Kafka Streams:

1. Stream-Stream Join: Este tipo de join se utiliza para combinar dos o más streams en uno nuevo. Para realizar esta unión los streams deben estar particionados por las mismas claves. Se puede utilizar una ventana de tiempo para limitar el alcance del join y unir solamente los eventos que caen dentro ésta. Puede ser útil para limitar la cantidad de eventos que se procesan y reducir el costo de procesamiento.
2. Table-Stream Join: se utiliza para combinar un stream con una tabla utilizando una clave común. El resultado es un nuevo stream enriquecido con la información de la tabla. La ventana de tiempo se puede utilizar para definir el período de tiempo durante el cual se debe realizar el join entre la tabla y el stream. Esto permite realizar uniones en la tabla basados en un período de tiempo específico, en lugar de en la tabla completa.
3. Table-table join: unión similar a la de dos tablas relacionales que se combinan cruzando por la misma clave. Este tipo de unión no admite ventanas de tiempo.

from pyspark.sql.functions import from_json, col, struct
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType

# Configuración de Kafka
kafka_bootstrap_servers = "localhost:9092"
stream1_topic = "stream1"
stream2_topic = "stream2"
stream1_starting_offsets = "earliest"
stream2_starting_offsets = "earliest"

# Definición del esquema de los datos
stream1_schema = StructType([
    StructField("id", IntegerType()),
    StructField("producto", StringType()),
    StructField("cantidad", IntegerType())
])

stream2_schema = StructType([
    StructField("producto", StringType()),
    StructField("precio", DoubleType())
])

# Creación de los DataFrames que representan los streams
stream1_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", stream1_topic) \
    .option("startingOffsets", stream1_starting_offsets) \
    .load() \
    .selectExpr("CAST(value AS STRING)") \
    .select(from_json("value", stream1_schema).alias("data")) \
    .select("data.*")

stream2_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", stream2_topic) \
    .option("startingOffsets", stream2_starting_offsets) \
    .load() \
    .selectExpr("CAST(value AS STRING)") \
    .select(from_json("value", stream2_schema).alias("data")) \
    .select("data.*")

Una vez que tenemos los DataFrames, podemos hacer el join utilizando la función join de PySpark:

# Realizamos el join entre los dos streams
joined_df = stream1_df.join(stream2_df, "producto")

# Escribimos el resultado en la consola para visualizarlo
query = joined_df \
    .writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

Funciones de ventana (windowing)

Las funciones de ventana (windowing) en Kafka Streams son una herramienta poderosa que permiten procesar flujos de datos en ventanas de tiempo discretas. Básicamente, una función de ventana divide el flujo de datos en segmentos de tiempo llamados ventanas, y aplica una operación de agregación (como sumar, contar, promediar) sobre los eventos que caen dentro de cada ventana.

Las funciones de ventana son muy útiles para realizar análisis en tiempo real de flujos de datos, como calcular promedios móviles, contar eventos por hora o sumar valores por día. Al procesar datos en ventanas de tiempo, se pueden obtener resultados más precisos y oportunos que al procesar datos en tiempo real sin una ventana de tiempo definida. Utilizan cinco unidades de tiempo: DD (día), HH (hora), MI (minuto), SS (segundo) y MS (milisegundo).

En Kafka Streams, existen varios tipos de funciones de ventana, entre las que se incluyen:

Tumbling windows
Son ventanas de tiempo fijas y no solapadas. Por ejemplo, una ventana de 10 segundos que inicia en los segundos 0, 10, 20, etc.

SELECT count(*) FROM demo GROUP BY ID, TUMBLINGWINDOW(ss, 10);

Hopping windows
Son ventanas de tiempo que se solapan. Por ejemplo, una ventana de 10 segundos que se mueve cada 5 segundos, lo que resulta en ventanas que cubren los segundos 0-10, 5-15, 10-20, etc. Un evento puede estar en dos ventanas de tiempo.

SELECT count(*) FROM demo GROUP BY ID, HOPPINGWINDOW(ss, 10, 5);

Sliding windows
Se desplazan a lo largo del tiempo. Por ejemplo, una ventana de 10 segundos que se desplaza cada segundo, lo que resulta en ventanas que cubren los segundos 0-10, 1-11, 2-12, etc. Un evento puede pertenecer a más de una ventana. Solo generan un output si se produce un evento y cada ventana tiene al menos uno. Os recomiendo esta lectura de Amazon AWS.

SELECT count(*) FROM demo GROUP BY ID, SLIDINGWINDOW(mi, 1);

Session windows
Se definen en función de una brecha de tiempo entre eventos. Si no se reciben eventos durante un período de tiempo determinado, se cierra la ventana actual y se abre una nueva ventana. Una ventana de sesión comienza cuando ocurre el primer evento. Si otro evento ocurre dentro del timeout especificado desde la ingesta del último evento, la ventana se amplía para incluirlo.

SELECT count(*) FROM demo GROUP BY ID, SESSIONWINDOW(mi, 2, 1);

Cada tipo de ventana tiene sus propias características y se utiliza en diferentes situaciones. Además, en Kafka Streams se pueden configurar ventanas con diferentes duraciones y períodos de desplazamiento según las necesidades del caso de uso. Os recomiendo esta lectura de la certificación Data Engineer Associate de Azure.

Ejecutar Spark y Cluster Kafka en docker (PySpark Streaming)

Vamos a seguir las instrucciones del bootcamp para levantar un contenedor Docker con Kafka.

1. Si aun no lo has hecho, clona la repo del bootcamp en local.

2. Arrancamos Docker desktop

3. Ejecutamos el shell week_6_streaming_processing\docker\spark\build.sh para descargar las imágenes de docker que vamos a necesitar para construir el contendor de Spark (spark-master, spark-worker y jupyterlab). Tardará un ratito ;).

bash build.sh

4. Creamos la red para que los contenedores Docker de Kafka y Spark que vamos a crear tengan conectividad entre sí:

docker network  create kafka-spark-network

docker volume create --name=hadoop-distributed-file-system

5. Arrancamos los contenedores de Kafka y Spark (están ubicados en las carpetas con este nombre de la repo que hemos clonado en el primer paso).

docker compose up -d

6. Comprobamos si se han levantado todos los servicios accediendo a los frontales web:

• JupyterLab en localhost:8888
• Spark master en localhost:8080
• Spark worker I en localhost:8083
• Spark worker II en localhost:8084
• Confluent Control center en localhost:9021 (kafka)

7. Para nuestra primera demo de PySpark y Kafka vamos a usar la repo del bootcamp, si la clonas, realmente usaremos los ficheros ubicados en los directorios streams-example y resources. Creamos un nuevo entorno de Python y nos ayudamos de un requirements.txt para instalar las librerías necesarias:

Creamos el nuevo entorno de Python:

virtualenv venv-kafka
virtualenv/scripts/activate

El fichero requirements.txt lo puedes copiar en la carpeta raíz del proyecto:

kafka-python
confluent_kafka
requests
avro
pyspark

Instalamos todas las librerías:

pip install -r requirements.txt

Iniciamos las pruebas lanzando el script del productor producer.py:

python producer.py

Y observamos como se empiezan a generar topics!!

Si lanzamos el script del consumer.py observamos cómo se cargan:

python consumer.py

Podemos comprobar desde el Confluent Control Center (http://localhost:9021/)el comportamiento de nuestro topic:

Producer.py

Este script se encarga de conectarse al fichero de datos de origen (rides.csv) con los datos y alimentar el topic con mensajes:

import csv
from time import sleep
from typing import Dict
from kafka import KafkaProducer

from settings import BOOTSTRAP_SERVERS, INPUT_DATA_PATH, PRODUCE_TOPIC_RIDES_CSV


def delivery_report(err, msg):
    if err is not None:
        print("Delivery failed for record {}: {}".format(msg.key(), err))
        return
    print('Record {} successfully produced to {} [{}] at offset {}'.format(
        msg.key(), msg.topic(), msg.partition(), msg.offset()))


class RideCSVProducer:
    def __init__(self, props: Dict):
        self.producer = KafkaProducer(**props)
        # self.producer = Producer(producer_props)

    @staticmethod
    def read_records(resource_path: str):
        records, ride_keys = [], []
        i = 0
        with open(resource_path, 'r') as f:
            reader = csv.reader(f)
            header = next(reader)  # skip the header
            for row in reader:
                # vendor_id, passenger_count, trip_distance, payment_type, total_amount
                records.append(f'{row[0]}, {row[1]}, {row[2]}, {row[3]}, {row[4]}, {row[9]}, {row[16]}')
                ride_keys.append(str(row[0]))
                i += 1
                if i == 5:
                    break
        return zip(ride_keys, records)

    def publish(self, topic: str, records: [str, str]):
        for key_value in records:
            key, value = key_value
            try:
                self.producer.send(topic=topic, key=key, value=value)
                print(f"Producing record for <key: {key}, value:{value}>")
            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"Exception while producing record - {value}: {e}")

        self.producer.flush()
        sleep(1)


if __name__ == "__main__":
    config = {
        'bootstrap_servers': [BOOTSTRAP_SERVERS],
        'key_serializer': lambda x: x.encode('utf-8'),
        'value_serializer': lambda x: x.encode('utf-8')
    }
    producer = RideCSVProducer(props=config)
    ride_records = producer.read_records(resource_path=INPUT_DATA_PATH)
    print(ride_records)
    producer.publish(topic=PRODUCE_TOPIC_RIDES_CSV, records=ride_records)

En este caso, se ha generado un settings.py para guardar toda la configuración:

import pyspark.sql.types as T

INPUT_DATA_PATH = 'resources/fhv_tripdata_2019-01.csv'
BOOTSTRAP_SERVERS = 'localhost:9092'

TOPIC_WINDOWED_VENDOR_ID_COUNT = 'vendor_counts_windowed'

PRODUCE_TOPIC_RIDES_CSV = CONSUME_TOPIC_RIDES_CSV = 'fhv_csv'

RIDE_SCHEMA = T.StructType(
    [T.StructField("dispatching_base_num", T.StringType()),
     T.StructField('pickup_datetime', T.TimestampType()),
     T.StructField('dropOff_datetime', T.TimestampType()),
     T.StructField("PUlocationID", T.IntegerType()),
     T.StructField("DOlocationID", T.FloatType()),
     T.StructField("SR_Flag", T.IntegerType()),
     T.StructField("Affiliated_base_number", T.FloatType()),
     ])

Consumer.py

Por otro lado, desde este script vamos a ir recibiendo los mensajes del topic:

import argparse
from typing import Dict, List
from kafka import KafkaConsumer

from settings import BOOTSTRAP_SERVERS, CONSUME_TOPIC_RIDES_CSV


class RideCSVConsumer:
    def __init__(self, props: Dict):
        self.consumer = KafkaConsumer(**props)

    def consume_from_kafka(self, topics: List[str]):
        self.consumer.subscribe(topics=topics)
        print('Consuming from Kafka started')
        print('Available topics to consume: ', self.consumer.subscription())
        while True:
            try:
                # SIGINT can't be handled when polling, limit timeout to 1 second.
                msg = self.consumer.poll(1.0)
                if msg is None or msg == {}:
                    continue
                for msg_key, msg_values in msg.items():
                    for msg_val in msg_values:
                        print(f'Key:{msg_val.key}-type({type(msg_val.key)}), '
                              f'Value:{msg_val.value}-type({type(msg_val.value)})')
            except KeyboardInterrupt:
                break

        self.consumer.close()


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Kafka Consumer')
    parser.add_argument('--topic', type=str, default=CONSUME_TOPIC_RIDES_CSV)
    args = parser.parse_args()

    topic = args.topic
    config = {
        'bootstrap_servers': [BOOTSTRAP_SERVERS],
        'auto_offset_reset': 'earliest',
        'enable_auto_commit': True,
        'key_deserializer': lambda key: int(key.decode('utf-8')),
        'value_deserializer': lambda value: value.decode('utf-8'),
        'group_id': 'consumer.group.id.csv-example.1',
    }
    csv_consumer = RideCSVConsumer(props=config)
    csv_consumer.consume_from_kafka(topics=[topic])

PySpark Structured Streaming

La última sección de la semana es sobre PySpark structured streaming, módulo de la biblioteca PySpark que permite el procesamiento en tiempo real de grandes conjuntos de datos estructurados. Utiliza una estructura de datos en forma de tabla, lo que significa que puedes procesar y analizar los datos con mucha más facilidad y profundidad. Además, es muy resistente a fallos y escalable, por lo que puede manejar grandes volúmenes de datos sin problemas.

Ha sido la parte en la que más he sufrido para poder lanzar el script streaming.py con spark-submit. Tras reinstalar Python y reconfigurar las variables de entorno, he logrado ejecutarlo en Windows 10 con Powershell. A continuación indico las instrucciones:

1. Revisa las variables de entorno en Windows, además de tenerlas declaradas de forma independiente, es necesario añadirlas al PATH:

• HADOOP_HOME=C:\tools\hadoop-3.2.0
• JAVA_HOME=C:\Program Files\Java\jdk-11.0.17
• SPARK_HOME=C:\tools\spark-3.3.2-bin-hadoop3
• PYSPARK_PYTHON=python

2. Arranca los contenedores de Docker que hemos comentado previametne con el clúster de Kafka y el Standalone Spark (ubicados en las carpetas correspondietnes).

docker compose up -d

3. Ahora sí, dirígete a la carpeta donde está ubicado el script streaming.py (\week_6_streaming_processing\streams-example\pyspark) y ejecuta el spark-submit:

spark-submit --master spark://localhost:7077 --num-executors 2 --executor-memory 512M --executor-cores 1  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.1,org.apache.spark:spark-avro_2.12:3.3.1,org.apache.spark:spark-streaming-kafka-0-10_2.12:3.3.1 streaming.py

WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources

Tras pelearme mucho con este error finalmente lo he solucionado añadiendo un nuevo worker al clúster de Spark. Como no tengo claro por qué con los dos workers que creamos en el contenedor no es suficiente, estoy pendiente de la respuesta en Slack. En este artículo indican dos opciones para resolverlo, pero ninguna me ha funcionado. El error indica que no hay memoria o cores suficientes ejecutar un nuevo job en el clúster, aunque si revisamos el Spark UI vemos que sí.

Solución

Crear un nuevo worker desde línea de comandos. Nos ubicamos en la carpeta donde hemos descomprimido spark (C:\tools\spark-3.3.2-bin-hadoop3) y ejecutamos:

./bin/spark-class org.apache.spark.deploy.worker.Worker spark://localhost:7077

Vemos en Spark UI cómo se ha creado el nuevo worker:

Y como devuelve los resultados de las dos operaciones de streaming:

La entrada Data Engineering Zoomcamp – Semana 6 se publicó primero en Marcos Torregrosa.

Entramos en la recta final del bootcamp para Data Engineers. Las dos semanas que quedan están dedicadas al procesamiento de datos, comenzando esta con el batch o procesamiento por lotes empleando tecnología spark y la semana que viene con el procesamiento de datos en tiempo real o streaming con Apache Kafka.

• Repo Bootcamp: https://github.com/DataTalksClub/data-engineering-zoomcamp/tree/main/week_5_batch_processing
• Repo personal con ejercicios: https://github.com/hegdehog/data-engineering-zoomcamp-2023

Última actualización: 07/03/2023

Prerequisitos

En mi caso estoy trabajando en Windows, os recomiendo seguir la guía de instalación de los prerequesitos. Es importante tener en cuenta que si estamos usando Git bash o MINGW64 el catálogo de variables de entorno no se replica en Windows, por lo que debemos arrancar los cuadernos de jupyter desde el command shell de unix para que coja correctamente los path configurados.

• Java: Mínimo la versión 11 del JDK de java desde su página oficial para trabajar con Spark. Recuerda que además de instalarlo, es necesario añadir a la variable de entorno PATH el JAVA_HOME con la ubicación de Java.
• Apache Hadoop: Descargamos los binarios de la versión 3.2 desde aquí. Es necesario añadir también en el PATH la ubicación del directorio donde extraigamos los archivos y crear la variable HADOOP_HOME.
• Apache Spark: vamos a descargar la versión 3.3.2 de Spark desde su repositorio. Si estás usando Git bash en Windows y no tienes instalado wget para descargar paquetes, debes hacerlo desde este enlace y copiar el archivo .exe en la ruta bin donde tengas instalado git bash (C:\Program Files\Git\mingw64\bin). Puedes acceder al GUI de Spark publicado en el puerto 4040: http://localhost:4040/jobs/. Creamos la variable de entorno SPARK_HOME y la añadimos también al PATH de Windows.
• PySpark: librería de Python para trabajar con Spark. Podemos seguir esta guía para su instalación. Si nos da problemas, específicamente el error Module not found cuando añadimos al PATH la ruta de PySpark, revisa la nota indicada en la repo del bootcamp (hay que cambiar el nombre del fichero en la ruta PATH si hemos descargado otra versión). Añadimos la variable de entorno PYTHONPATH según las instrucciones.

Procesamiento de datos

El procesamiento de datos es el conjunto de operaciones y técnicas utilizadas para transformar datos brutos o sin procesar en información útil y significativa. Esta información puede ser utilizada para tomar decisiones, realizar análisis, predicciones o automatizar procesos. Existen distintas técnicas o enfoques para procesar los datos donde destacan los procesos batch (por lotes) y streaming (en tiempo real), que a su vez pueden clasificarse en procesamiento distribuido, paralelo, online, transaccional o analítico. Esta semana del bootcamp está centrada en el batch, por lo que vamos a dejar por aquí una breve pincelada del procesamiento streaming donde profundizaremos la semana que viene.

Batch

El procesamiento por lotes (batch) se utiliza para realizar periódicamente trabajos de datos repetitivos y de gran tamaño. Las tareas de transformación, filtrado y clasificación pueden ser intensivas en cálculo e ineficientes si se ejecutan en transacciones individuales. En su lugar se procesan estas tareas en lotes, a menudo en momentos de menor actividad cuando los recursos informáticos están más disponibles, como al final del día o durante la noche.

Consideremos una tienda online que recibe pedidos durante todo el día. En lugar de procesar cada pedido a medida que ocurre, el sistema podría recopilar todos los pedidos al final de cada día y compartirlos en un bloque con el equipo de cumplimiento de pedidos.

Streaming

El procesamiento de datos en tiempo real o streaming (streaming processing / real-time processing) es un método que se realiza en tiempo real, a medida que se generan o se reciben los datos. Cuando la cantidad de datos es desconocida o infinita es preferible aplicar el procesamiento por streaming en lugar de batch.

Este enfoque es adecuado para tareas que requieren una respuesta rápida, como la detección de fraudes, la monitorización de sistemas en línea, el análisis de datos de sensores (IoT) o de logs.

El procesamiento de datos en streaming permite tomar decisiones rápidas basadas en la información más reciente disponible.

Batch processing (procesamiento de datos por lotes)

Como hemos indicado en la introducción, el procesamiento batch o por lotes trabaja con conjuntos de datos que procesa a intervalos de tiempo. Por ejemplo, si un banco o aseguradora debe cumplir sus compromisos regulatorios hacia entidades de control de forma mensual, existirá un proceso batch mensual que agrupe toda la información, la transforme, agregue y cocine para generar el modelo de datos del reporting. Habitualmente los procesos batch tienen intervalos:

• Mensual (más habitual)
• Semanal (más habitual)
• Diario (más habitual)
• Cada hora
• 3 por hora
• Cada 5 minutos
• …

A nivel de tecnología podemos crear un proceso batch en cualquier lenguaje de programación (por ejemplo, Python, Java, Scala…), modelar los datos con dbt y orquestar los scripts con Apache Airflow, Prefect, Control-M, etc.

Apache Spark

¿Qué es Apache Spark?

Apache Spark es un motor de procesamiento de datos de código abierto utilizado para realizar análisis y transformación de grandes volúmenes de datos en clústeres de servidores distribuidos (paralelizando el procesamiento en distintos nodos). Fue desarrollado originalmente en la Universidad de California, Berkeley, y ahora es mantenido por la Apache Software Foundation. Básicamente lo que hace Spark es dividir una carga de trabajo en varias porciones que distribuye entre distintos nodos o máquinas para que trabajen de forma paralela y cuando finalizan, agrupan los resultados y lo devuelven.

Spark es conocido por su velocidad, ya que puede procesar grandes conjuntos de datos mucho más rápido que otras herramientas con el mismo objetivo, como Hadoop MapReduce. Spark también admite múltiples lenguajes de programación, como Scala, Java, Python y R. Además, proporciona una variedad de bibliotecas y herramientas para diferentes tareas de procesamiento de datos, como batch, streaming, procesamiento de gráficos y aprendizaje automático (ML o machine learning).

Dentro de un ecosistema Data Lake utilizar Spark nos va a ayudar en el proceso de transformación de los datos. En un Data lake los datos se almacenan como ficheros, habitualmente csv o parquet, que podemos consultar como si fuera un modelo de datos SQL mediante herramientas tales como Hive, Presto o Athena (en cloud Amazon AWS), o BigQuery (en Google Cloud Platform). En el caso de que la lógica sea más compleja y no podamos resolverla mediante SQL, entra en juego Spark. En un mismo flujo de trabajo podemos combinar ambas opciones, cuando los datos puedan transformarse por SQL usaremos este camino, y cuando sean transformaciones complejas lo haremos con Spark.

Arquitectura Spark

Spark se basa en una arquitectura de procesamiento distribuido, lo que significa que utiliza un clúster o grupo de ordenadores para procesar datos. Consta de varios componentes que se comunican entre sí para ejecutar las tareas.

Un clúster de Spark consiste en un proceso Driver que se ejecuta dentro de un nodo Master y procesos Executor que se ejecutan dentro de cada uno de los nodos Worker. Cuando se envía un trabajo a Spark, el Driver particiona y distribuye el trabajo en forma de tareas a los procesos Executor (en diferentes nodos Worker) para su procesamiento adicional. A medida que se ejecuta el trabajo de la aplicación, los Executor informan al Driver sobre el estado de la tareas, y así éste mantiene el estado general del trabajo de la aplicación. Cada Worker tiene su propia memoria y CPU, y está conectado a otros Workers a través de una red de alta velocidad. Pueden ser añadidos o eliminados del clúster según sea necesario para ajustar la capacidad de procesamiento

¿Cómo sabe el proceso Driver qué Executors están disponibles para el procesamiento y a quién distribuir las tareas? gracias al Administrador de Clúster (Cluster Manager). Realiza un seguimiento del estado de los recursos del clúster (qué procesos Executor en qué nodos Worker están disponibles, etc.). El Driver está conectado con el Cluster Manager a través de una SparkSession o un SparkContext (SparkSession estaría por encima del SparkContext).

La arquitectura de Apache Spark cosiste principalmente en dos capas de abstracción:

Resilient Distributed Datasets (RDD):
Es la célula del ecosistema Spark, el elemento básico para trabajar con los datos. Se caracterizan porque son inmutables (los datos no pueden cambiarse una vez se crean), distribuidos (siguiendo el patrón de Spark, se dividen en particiones entre los nodos del clúster) y resilientes (automáticamente es capaz de regenerar una partición que se haya perdido). Hay dos operaciones que pueden realizarse sobre los RDDs: transformaciones y acciones.

Directed Acyclic Graph (DAG):
El driver convierte cada tarea en un job de tipo DAG (grafo acíclico dirigido) formado por vértices (RDD) y aristas (sus transformaciones). En lenguaje coloquial, cada tarea es un trabajo dividido en etapas (vértices) que siguen un secuencia lineal (acíclico). Las etapas están construidas con alguno de los componentes de Spark (API Core, Spark SQL, Streaming, real-time processing, MLlIB o GraphX).

Ecosistema Spark

El ecosistema de Spark lo forman los siguientes elementos:

• Spark Core: Es el componente principal de Spark y proporciona las funcionalidades básicas, como el procesamiento distribuido, la programación paralela y la tolerancia a fallos. Es la API para el procesamiento batch.
• Spark SQL: Proporciona una API para trabajar con datos estructurados o semiestructurados usando SQL. Nos ofrece tres vías para ello:
  • DataFrames: estructura de datos distribuida que se organiza en columnas con nombres y tipos de datos (similar a una tabla relacional). Se pueden crear a partir de archivos de datos estructurados como CSV o JSON, o mediante la lectura de datos de una base de datos relacional utilizando Spark SQL. Los DataFrames también se pueden transformar mediante operaciones de filtrado, agregación y unión para realizar tareas de análisis de datos.
  • Datasets: es una API más segura y fuertemente tipada que se encuentra en la parte superior de los DataFrames. Los Datasets permiten trabajar con datos estructurados de manera más fácil y natural, ya que se definen los esquemas de los datos de forma estática. Se generan a partir de ficheros CSV, JSON, bases de datos relacionales, etc. También se pueden transformar mediante operaciones de filtrado, agregación y unión.
  • Lenguaje SQL a través de una API SQL pata trabajar sobre DataFrames y Datasets. Admite una amplia gama de funciones SQL como SELECT, FROM, WHERE, JOIN, GROUP BY, ORDER BY, etc.
• Spark Streaming: Es un componente que permite procesar datos en tiempo real, como publicaciones de Twitter o Facebook. Procesa los datos en lotes y utiliza la misma API que Spark Core.
• Spark MLlib: Proporciona algoritmos de aprendizaje automático para realizar tareas como clasificación, regresión y agrupamiento de datos en modo distribuido.
• Spark GraphX: Proporciona herramientas para trabajar con datos gráficos y realizar análisis de redes y grafos.

PySpark (Python + Apache Spark)

PySpark es una librería Python para desarrollar aplicaciones que explotan todas las capacidades de Apache Spark (procesamiento distribuido paralelizando las cargas de trabajo entre nodos) ideal para proyectos de datos a gran escala y machine learning (ML). Necesitamos descargar la librería pyspark bien por pip o siguiendo las instrucciones del bootcamp.

import pyspark

SparkSession

SparkSession es una clase en PySpark que se utiliza para trabajar con Spark y que proporciona una interfaz única para interactuar con diferentes tipos de datos en Spark, como RDD, DataFrames y DataSet. SparkSession se utiliza para crear y configurar SparkContext, SQLContext y HiveContext en una sola sesión.

Para instanciar una SparkSession debemos invocar el constructor y pasarle varios parámetros, aunque en vamos a trabajar sólo con los dos primeros:

• appName: Nombre de la aplicación de Spark, por ejemplo «test«
• master: especifica la dirección del clúster de Spark en el que se ejecutará la aplicación. Puede ser una URL de un cluster de Spark independiente o ejecución local:
  • local: Especifica el modo de ejecución local, es decir, que se ejecutará en una única máquina como un proceso local.
  • local[N]: Especifica el modo de ejecución local con N hilos.
  • local[*]: Especifica el modo de ejecución local con tantos hilos como núcleos de CPU disponibles.
  • yarn: Especifica el modo de ejecución en un cluster de YARN.
  • mesos: Especifica el modo de ejecución en un cluster de Mesos.
  • spark://HOST:PORT: Especifica la URL de un cluster de Spark independiente.
  • k8s://https://HOST:PORT: Especifica la URL del API server de Kubernetes en el que se ejecutará la aplicación.
• config: Configuraciones adicionales de Spark.
  • spark.executor.memory: Cantidad de memoria asignada a cada ejecutor.
  • spark.driver.memory: Cantidad de memoria asignada al driver.
  • spark.sql.shuffle.partitions: Número de particiones utilizadas por las operaciones de shuffle en SQL.
  • spark.serializer: Serializador utilizado para serializar/deserializar objetos.
  • spark.ui.port: Puerto utilizado por la interfaz de usuario web de Spark.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("test") \
    .master("local[*]") \
    .getOrCreate()

Para acceder al Spark UI podemos consultar la URL http://localhost:4040 (si no hemos especificado otro puerto en la configuración). Si deseamos crear otra SparkSession para otro notebook podemos especificar un nuevo puerto distinto al de por defecto 4040 en el .config():

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .master("local[*]") \
    .appName('test') \
    .config("spark.ui.port", "4041") \
    .getOrCreate()

Spark DataFrames con CSV y Parquet

Como hemos mencionado previamente, un DataFrame es un set de datos con tipado y organizado en columnas. A continuación vamos a ver cómo cargar en un DataFrame de PySpark el contenido de un CSV o un Parquet. Como breve pincelada del formato Parquet indicar que es un tipo de archivo muy utilizado en el mundo de datos al tener un alto grado de compresión (1:10 respecto a CSV) gracias a que almacena los datos de forma columnar, no por filas, lo que permite ajustar el tipado y por tanto el espacio necesario. Otra de las ventajas es que el esquema de los datos está incluido en el propio fichero, por lo que no es necesario inferirlo o asignarlo.

CSV
Para cargar un CSV utilizaremos el método read.csv(). Podemos especificar algunos parámetros de configuración, por ejemplo header=True indica que la primera fila es la cabecera y inferSchema=True va a inferir esquema en base a los datos de cada columna.

df = spark.read.csv('taxi+_zone_lookup.csv', header=True, inferSchema=True)

Parquet
En el caso de de Parquet usaremos el método read.parquet(). La ventaja de cargar un archivo Parquet en lugar de otro formato es que estos archivos suelen tener un tamaño más pequeño y una mayor eficiencia de procesamiento debido a su estructura columnar y compresión de datos.

df = spark.read.parquet('taxi+_zone_lookup.parquet')

Partitions

En PySpark, las particiones se utilizan para dividir un conjunto de datos en fragmentos más pequeños y distribuirlos a través de un clúster de Spark. Cada partición es procesada en paralelo por un executor en el clúster, lo que permite procesar grandes conjuntos de datos de manera más eficiente y escalable. Por ejemplo, si deseamos procesar un fichero CSV o parquet de 1GB, podríamos segmentarlo en 10 particiones para que cada una sea trabajada de forma paralela en 10 nodos del clúster.

Las particiones son una de las unidades fundamentales de procesamiento en Spark y se utilizan en varios tipos de objetos de datos, como RDD, DataFrames y DataSets. En general, se recomienda tener un número adecuado de particiones para un conjunto de datos dado, ya que demasiadas particiones pueden provocar un exceso de sobrecarga en la comunicación entre los ejecutores y demasiado pocas particiones pueden resultar en una utilización ineficiente de los recursos del clúster.

Las particiones se pueden especificar al crear un RDD o al leer un conjunto de datos en un DataFrame o DataSet. Por ejemplo, al leer un archivo CSV en un DataFrame, se puede especificar el número de particiones mediante el parámetro numPartitions:

df = spark.read.csv("path/to/csv/file.csv" \
        , header=True \
        , inferSchema=True \
        , numPartitions=8)

También es posible ajustar el número de particiones de un RDD o DataFrame existente utilizando el método repartition o coalesce. El método repartition redistribuirá los datos aleatoriamente a través del clúster y creará el número especificado de particiones, mientras que coalesce fusionará particiones adyacentes para crear el número especificado de particiones.

df = df.repartition(24)

Si queremos persistir el DataFrame en un fichero parquet, por ejemplo, usaremos el método write.parquet pasándole como parámetro el directorio para la salida:

df = df.repartition(24)
df.write.parquet('fhvhv/2021/01/')

Si consultamos en la Spark UI (http://localhost:4040) el progreso del trabajo podemos ver que está dividido en dos etapas: en primer lugar Spark crea tantas particiones como cores tenga nuestra CPU (recuerda que hemos creado el SparkSession con Local[*]), por lo que si tenemos 4, va a dividir el DataFrame en 4 particiones. Cada partición genera una task. Cuando hayan finalizado las 4 tareas de esta etapa, se genera el particionado que hayamos especificado. En el ejemplo, se generarán 10 tasks, cada una para cada partición. Como nuestra CPU sólo tiene 4 cores las tasks se van a ir encolando y ejecutando según se vayan quedando disponibles. Cada una de las particiones se almacena en un fichero con el sufijo _snappy.parquet, que es el formato de compresión de alta velocidad por defecto de parquet.

Para ilustrar el ejemplo, mi equipo tiene 12 cores y vamos a crear 24 particiones del fichero parquet.

Al lanzar el comando .write.parquet() Spark va a generar dos etapas: la primera de 12 tasks que se corresponden con los 12 cores de mi CPU y la segunda con las 24 tasks correspondientes a las 24 particiones que he especificado.

Si consultamos el directorio podremos observar como se han generado 24 ficheros de tipo .snappy.parquet:

Transformaciones y Acciones de PySpark

En PySpark, las operaciones se dividen en dos categorías: Transformaciones (Transformations) y Acciones (Actions).

Las transformaciones son operaciones que toman un DataFrame como entrada, aplican una transformación y generan un nuevo DataFrame como resultado. Son operaciones «perezosas» (lazy), lo que significa que no se ejecutan inmediatamente sino que se almacenan en el grafo de transformación hasta que se requiere una acción.

• select(): selecciona columnas específicas.
• filter(): filtra filas que satisfacen una condición específica.
• groupBy(): agrupa filas por una o más columnas.
• join(): une dos DataFrames en función de una o más columnas comunes.
• distinct(): devuelve un nuevo DataFrame que contiene solo valores distintos.
• orderBy(): ordena las filas en función de una o más columnas.
• withColumn(): agrega una nueva columna o reemplaza una columna existente con una nueva.
• drop(): elimina una o más columnas.

Por otro lado, las acciones son operaciones que toman un DataFrame como entrada y producen un resultado que se almacena o se muestra. Las acciones son operaciones que «activan» el grafo de transformación y hacen que se ejecuten las transformaciones almacenadas.

• show(): muestra una vista previa de un número determinado de filas.
• count(): cuenta el número de filas.
• collect(): recopila todos los datos en la memoria del driver.
• write(): escribe en un archivo o fuente de datos externa.
• first(): devuelve la primera fila.
• max() y min(): devuelve el valor máximo o mínimo en una columna numérica.
• sum(): devuelve la suma de los valores en una columna numérica.
• mean(): devuelve la media de los valores en una columna numérica.
• pivot(): crea una tabla dinámica a partir.

En general, se recomienda minimizar el número de acciones en PySpark y maximizar el uso de transformaciones, ya que éstas son más eficientes y permiten una mejor optimización del flujo de procesamiento de datos.

Funciones de Spark

Dentro del ecosistema Spark podemos encontrar dos tipos de funciones: un grupo de funciones built-in incluidas por defecto para realizar distintas operaciones y las User-defined functions (UDF) que son funciones personalizadas que podemos desarrollar a nuestro gusto.

Funciones built-in
La forma de importarlas es la siguiente. Para ver todas las opciones disponibles basta con escribir en una celda de un cuaderno F. y pulsar el tabulador.

from pyspark.sql import functions as F

Siguiendo el ejemplo del curso, podemos convertir un DATETIME a DATE utilizando la función to_date():

df \
    .withColumn('pickup_date', F.to_date(df.pickup_datetime)) \
    .withColumn('dropoff_date', F.to_date(df.dropoff_datetime)) \
    .select('pickup_date', 'dropoff_date', 'PULocationID', 'DOLocationID') \
    .show()

User-defined functions (UDF)
Una UDF (User-Defined Function) en Spark es una función definida por el usuario que se puede usar para realizar transformaciones de datos personalizadas en un DataFrame o RDD. Las UDF se definen en Python, Java, Scala o R, y se pueden aplicar en PySpark, Spark SQL y Spark Streaming. Para definir una UDF en PySpark, podemos usar la función udf() de la biblioteca pyspark.sql.functions. A continuación se muestra un ejemplo de cómo definir una UDF en PySpark para calcular el cuadrado de un número:

from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType

def square(x):
    return x*x

square_udf = F.udf(square, DoubleType())

En este ejemplo, se define una función square() que calcula el cuadrado de un número, y luego se usa la función F.udf() para convertirla en una UDF. La UDF se define para que tome un argumento de entrada de tipo double y devuelva un valor de tipo double.

Una vez que se ha definido una UDF, se puede aplicar a una columna de un DataFrame mediante la función withColumn() de PySpark:

df \
    .withColumn('square_trip_miles', square_udf(df.trip_miles)) \
    .select("hvfhs_license_num","trip_miles","square_trip_miles") \
    .show(10)

Trabajando con DataFrames y Spark SQL

Podemos trabajar con un DataFrame utilizando los métodos de PySpark o bien aprovechar uno de los componentes clave de Spark: API SQL. Gracias al cual podemos consultar DataFrames como si fueran tablas relacionales y lanzar queries en SQL estándar. Vamos a verlo con dos ejemplos del mismo caso de uso. Vamos a cargar un parquet con las ventas de una tienda online. El fichero tiene tres columnas (producto, cantidad y fecha):

Consulta de DataFrame con PySpark
Para explotar un DataFrame con PySpark tenemos a nuestra los operadores de transformación que comentamos más arriba. Todos ellos se apilan sobre el mismo DataFrame por puntos » . «:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ConsultaVentas").getOrCreate()

df = spark.read.parquet("ruta/ventas.parquet")

fecha = "2022-02-28"
df_productos_vendidos= df.filter(df.fecha == fecha).groupBy("producto").sum("cantidad")

df_productos_vendidos.show()

Consulta de DataFrame con Spark SQL
Al utilizar el API de Spark SQL vamos a poder escribir SQL estándar y consultar el DataFrame como si se tratase de una tabla relacional utilizando el método spark.sql(). Como paso previo es necesario crear una tabla temporal a partir del DataFrame utilizando la función createOrReplaceTempView() y pasándole como parámetro el nombre de la tabla, en nuestro caso ventas.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ConsultaVentas").getOrCreate()

df = spark.read.parquet("ruta/ventas.parquet")

df.createOrReplaceTempView("ventas")

fecha = "2022-02-28"
df_productos_vendidos = spark.sql(f"SELECT producto, SUM(cantidad) FROM ventas WHERE fecha = '{fecha}' GROUP BY producto")

df_productos_vendidos.show()

DataFrames de NY Taxis
En los ejemplos del bootcamp vamos a utilizar los datos de las carreras de taxis de NY green y yellow de 2020 y 2021. Podemos descargarlos de forma manual, desde un notebook de Jupyter y con el shell script que han preparado en el curso (download_data.sh) y lanzando los comandos:

bash download_data.sh yellow 2020
bash download_data.sh yellow 2021
bash download_data.sh green 2020
bash download_data.sh green 2021

1. Vamos a generar un DataFrame agrupando todos los ficheros mensuales por cada tipo de taxi (green y yellow).

df_green = spark.read.csv('data/raw/green/*/*', header=True, inferSchema=True)
df_yellow = spark.read.csv('data/raw/yellow/*/*', header=True, inferSchema=True)

2. Como el objetivo es unirlos en un único DataFrame debemos asegurarnos de que tienen el mismo esquema. Podemos consultarlo con df_yellow.schema() y df_green.schema(). Vamos a realizar varias acciones para crear el nuevo df con las columnas que tienen en común ambos y además, añadir una nueva que identifique el origen:

• Añadir una nueva columna en cada df para identificar el origen usando el operador de transformación .withColumn(nombre_columna, valor) y la función F.lit() de las librería built-in de Spark que nos permite especificar un valor literal: .withColumn('service_type', F.lit('green'))
• Columnas que no están en ambos DataFrames. La forma rápida es convirtiendo la lista de columnas a SET de python y combinarlos, pero perdemos el orden de las columnas, por lo que tenemos que usar un bucle FOR para recorrer comparar ambas listas y generar una nueva con las que tienen en común.
• Columnas con las fechas de subida y bajada (xy_pickup _datetime y xy_dropoff_datetime) tienen distinto nombre. Vamos a renombrarlas en ambos df con el operador de transformación .withColumnRenamed(columna, nueva_columna).

from pyspark.sql import functions as F

# renombramos columnas de fecha:

df_green = df_green \
    .withColumnRenamed('lpep_pickup_datetime', 'pickup_datetime') \
    .withColumnRenamed('lpep_dropoff_datetime', 'dropoff_datetime')

df_yellow = df_yellow \
    .withColumnRenamed('tpep_pickup_datetime', 'pickup_datetime') \
    .withColumnRenamed('tpep_dropoff_datetime', 'dropoff_datetime')

# generamos lista con columnas en común entre ambos DataFrames
common_colums = []

yellow_columns = set(df_yellow.columns)

for col in df_green.columns:
    if col in yellow_columns:
        common_colums.append(col)

# generamos los nuevos df por cada tipo sólo con las columnas en común y añadiendo el service_type para identificar el tipo de taxi:
df_green_sel = df_green \
    .select(common_colums) \
    .withColumn('service_type', F.lit('green'))

df_yellow_sel = df_yellow \
    .select(common_colums) \
    .withColumn('service_type', F.lit('yellow'))

3. Por último vamos a combinar ambos DataFrames que ahora sí tienen el mismo esquema y una nueva columna para identificar el tipo de taxi.

df_trips_data = df_green_sel.unionAll(df_yellow_sel)

Podemos comprobar el número de registros por tipo de taxi usando PySpark:

df_trips_data.groupBy('service_type').count().show()

Si observamos el DAG generado por el job Spark vemos que se leen los dos CSV por separado y se unen en único RDD:

GroupBy en Spark

Vamos a analizar cómo gestiona internamente Spark el operador GroupBy. Podemos probar con Spark SQL o con PySpark. Para nuestro ejemplo vamos a calcular el beneficio y cantidad de viajes por hora y zona de los taxis.

Si lo queremos calcular con PySpark:

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .master("local[*]") \
    .appName('test_groupby') \
    .config("spark.ui.port", "4042") \
    .getOrCreate()

df_green = spark.read.csv('data/raw/green/*/*', header=True, inferSchema=True)

df_green_revenue = df_green.filter("lpep_pickup_datetime >= '2020-01-01 00:00:00'") \
    .withColumn("hour", F.date_trunc("hour", "lpep_pickup_datetime")) \
    .groupBy("hour", "PULocationID") \
    .agg({"total_amount": "sum", "*": "count"}) \
    .withColumnRenamed("sum(total_amount)", "amount") \
    .withColumnRenamed("count(1)", "number_records") \
    .orderBy("hour","PULocationID")

O bien con Spark SQL:

import pyspark
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .master("local[*]") \
    .appName('test_groupby') \
    .config("spark.ui.port", "4042") \
    .getOrCreate()

df_green = spark.read.csv('data/raw/green/*/*', header=True, inferSchema=True)

df_green.createOrReplaceTempView("green")

df_green_revenue = spark.sql("""
SELECT 
    date_trunc('hour', lpep_pickup_datetime) AS hour, 
    PULocationID AS zone,

    SUM(total_amount) AS amount,
    COUNT(1) AS number_records
FROM
    green
WHERE
    lpep_pickup_datetime >= '2020-01-01 00:00:00'
GROUP BY
    1, 2  
""")

df_green_revenue.show()

Vamos a persistir la salida en un parquet y analizar cómo realiza la tarea Spark:

df_green_revenue.write.parquet('data/report/revenue/green', mode="overwrite")

Al tratarse de procesamiento distribuido se dividen los datos en particiones que son enviadas a cada no de los executors. Para combinar los resultados de cada uno se realiza la operación de shuffle.

Si observamos en la Spark UI, vemos que la tarea se ha dividido en dos stages, el primero para preparar el groupBy (genera las agrupaciones intermedias de cada executor) y el segundo genera el resultado final combinándolas (shuffle). Spark no sólo ha generado dos stages para la tarea, si no que además ha ejecutado cada stage en un job independiente. El motivo es que la operación de shuffle es muy costosa. Para agilizar el procesamiento el primer job almacena los resultados intermedios en la caché que luego usará el shuffle, por eso en el segundo job aparece el primer stage como Skipped, que ya ha sido procesado en el primero. Más info en la documentación de Shuffle de Spark.

Si vemos el DAG de cada job:

Si analizamos cada uno de los stages en detalle, en el primero los datos se dividen en particiones y cada una es enviada a un executor, donde se realizan las agrupaciones y operaciones necesarias (filter y groupBy). Esto genera las agrupaciones o resultados intermedios, donde H es la hora, Z la zona y los dos KPIs con el beneficio y total de carreras.

En el segundo stage se realiza la operación de shuffle para combinar todos los resultados intermedios y agrupar los datos con la misma clave en la misma partición. Spark identifica como clave las columnas del GroupBy (en nuestro ejemplo hour y zone). Por último, se realiza un nuevo GroupBy sobre las nuevas particiones para realizar reducir los datos agrupándolos por la clave. Es posible que en alguna de las particiones generadas por la operación shuffle haya datos de distintas claves, pero en la última operación de agrupación se reparten en las particiones correspondientes.

Join en Spark

En Spark podemos combinar dos DataFrames como si se tratasen de dos tablas relacionales mediante el operador .join(df, on, how), donde:

1. df: El DataFrame que se unirá con el DataFrame principal. Debe especificarse como un objeto DataFrame de Pyspark.
2. on: Una o varias columnas comunes a ambas tablas que se utilizarán para unir los df. Puede especificarse como una cadena que contenga el nombre de la columna, o una lista de cadenas que contengan los nombres de las columnas (["id","nombre"]).
3. how: El tipo de unión que se realizará. Puede tomar uno de los siguientes valores:
   • 'inner': Realiza una unión interna, es decir, devuelve sólo los registros que tienen coincidencias en ambas tablas.
   • 'outer' o 'full': Realiza una unión externa completa, es decir, devuelve todos los registros de ambas tablas, incluso si no tienen una coincidencia en la otra tabla.
   • 'left' o 'left_outer': Realiza una unión externa izquierda, es decir, devuelve todos los registros del DataFrame izquierdo y los registros coincidentes del DataFrame derecho. Si no hay coincidencias en el DataFrame derecho, los valores para las columnas del DataFrame derecho serán null.
   • 'right' o 'right_outer': Realiza una unión externa derecha, es decir, devuelve todos los registros del DataFrame derecho y los registros coincidentes del DataFrame izquierdo. Si no hay coincidencias en el DataFrame izquierdo, los valores para las columnas del DataFrame izquierdo serán null.
   • 'left_semi': Realiza una unión semijoin izquierda, es decir, devuelve sólo los registros del DataFrame izquierdo que tienen una coincidencia en el DataFrame derecho.
   • 'left_anti': Realiza una unión anti-izquierda, es decir, devuelve sólo los registros del DataFrame izquierdo que no tienen una coincidencia en el DataFrame derecho.

from pyspark.sql.functions import *

df = df1.join(df2, on='id', how='left')

En el bootcamp profundizan en cómo se comporta Spark internamente cuando debe realizar operaciones de join entre tablas muy grandes o cuando una de las tablas es muy grande y la otra pequeña. En el primer caso, el comportamiento es muy similar al del GroupBy. Spark particiona los DataFrames, hace un shuffle para reorganizarlos por la clave y finalmente los agrupa en particiones por clave.

En el ejemplo observamos como hacemos un outer join entre los DataFrames de carreras de taxis yellow y green para quedarnos todos los registros de ambas tablas. Esto a a generar un nuevo DataFrame con las columnas de la clave (hour y zone) y las columnas de amount y number_records de yellow y green. Para facilitar el trabajo se renombran. Podéis consultar el notebook completo en mi repo.

df_green_revenue_tmp = df_green_revenue \
    .withColumnRenamed('amount', 'green_amount') \
    .withColumnRenamed('number_records', 'green_number_records')

df_yellow_revenue_tmp = df_yellow_revenue \
    .withColumnRenamed('amount', 'yellow_amount') \
    .withColumnRenamed('number_records', 'yellow_number_records')

df_join = df_green_revenue_tmp.join(df_yellow_revenue_tmp, on=['hour', 'PULocationID'], how='outer')

df_join.write.parquet('data/report/revenue/total', mode='overwrite')

Si comprobamos en el Spark UI observamos que la operación se ha dividido en tres jobs que procesan un stage cada uno. Los dos primeros se encargan de la lectura de cada uno de los DataFrames y el último (con 12 tasks) es el encargado de realizar el join.

Si entramos a ver el DAG del último job observamos que los dos primeros stages están skipped (se han ejecutado previamente). En el Stage 87 se encarga de realizar el shuffle y por último agrupar el resultado (reduces).

En el segundo caso, vamos a cruzar una tabla lookup pequeña (zones) con una muy grande (el DataFrame que hemos acabamos de generar combinando las carreras de ambos tipos de taxi).

En Spark cuando se realiza una operación de join entre dos DataFrames el proceso de unión puede ser costoso en términos de recursos computacionales, especialmente si uno de ellos es mucho más grande que el otro. Una forma de optimizar este proceso de unión es utilizar el broadcasting.

En lugar de enviar el DataFrame pequeño a través de la red y unirlo con el DataFrame grande en los nodos de trabajo (wokers), Spark envía el DataFrame pequeño a todos los nodos de trabajo que procesan el DataFrame grande. De esta manera, cada worker puede realizar la unión localmente sin necesidad de realizar una operación de red costosa.

El broadcasting de Spark se puede utilizar en operaciones de join cuando se cumplen las siguientes condiciones:

• Uno de los DataFrames es significativamente más pequeño que el otro, por lo general, cuando el tamaño del DataFrame pequeño es menor que el tamaño de la memoria disponible en cada nodo del clúster.
• El tamaño del DataFrame pequeño es menor que el parámetro spark.sql.autoBroadcastJoinThreshold. Este parámetro define el tamaño máximo que un DataFrame puede tener para que Spark utilice la técnica de broadcasting.

Para el ejemplo los nombres de columna no son iguales en ambos df en lugar de especificar el parámetro on vamos a indicar directamente las columnas clave de cada uno df_join.PULocationID == df_zones.LocationID. La operación de join se lleva todas las columnas al nuevo DataFrame. Como en este caso la clave tiene un nombre distinto en cada df, en el nuevo estarán las dos columnas de Location. Como buena práctica de limpieza es recomendable borrar una de ellas .drop('LocationID').

!wget https://s3.amazonaws.com/nyc-tlc/misc/taxi+_zone_lookup.csv

df_zones = spark.read \
    .option("header", "true") \
    .csv('taxi+_zone_lookup.csv')

df_result = df_join.join(df_zones, df_join.PULocationID == df_zones.LocationID)

df_result.drop('LocationID').write.parquet('tmp/revenue-zones')

Vamos a volver a Spark UI a ver cómo ha generado la orden de trabajo. Observamos que se ha dividido en 4 jobs con una stage cada uno. Si prestamos atención, vemos que el job con id 69 con sólo 3″ de duración habría cargado el DataFrame pequeño de zonas, a continuación el job 70 realiza la operación de broadcast, el 71 carga el segundo DataFrame y por último el job que realiza el join.

Resilient Distributed Datasets (RDDs)

Imagina que Spark es como una gran caja de herramientas para trabajar con datos en un grupo de ordenadores (clúster). En esa caja, una de las herramientas más importantes es el RDD, que es un conjunto de objetos distribuidos entre todos los nodos del clúster. Los RDD son muy útiles porque se pueden procesar en paralelo y procesar datos de manera más rápida.

Pero para trabajar con datos estructurados, como si fueran una tabla de una base de datos, se necesita una herramienta un poco diferente. Esa herramienta es el DataFrame, que es como una tabla que se puede manipular con facilidad. La ventaja de los DataFrames es que son más fáciles de usar que los RDDs y se pueden procesar de manera más eficiente.

Aunque los DataFrames son más fáciles de usar, siguen utilizando los RDDs en segundo plano para hacer todo el trabajo duro. Cuando se convierte un RDD en un DataFrame, se están organizando los datos en una tabla fácil de usar, y cuando se convierte un DataFrame en un RDD, se están sacando los datos de esa tabla para que se puedan procesar de manera más eficiente en paralelo. En general, se recomienda utilizar DataFrames siempre que sea posible debido a que son más fáciles de usar y optimizados para consultas SQL y de tipo tabla. Sin embargo, hay algunas situaciones donde puede ser útil utilizar RDD, como por ejemplo, si se necesita realizar operaciones más complejas o cálculos de bajo nivel que no son posibles con DataFrames. Además, si estamos trabajando con tipos de datos no estructurados, como archivos de registro, los RDD pueden ser una opción más adecuada.

Los datos en un RDD están organizados en un tipo de objeto llamado Row, que sería un concepto similar al de fila o tupla de una base de datos relacional o estructurada. Podemos acceder a los valores de cada columna en esa fila, ya sea por su índice (como un número de posición) o por su nombre (como un título).

Dentro de RDD, se pueden utilizar tres operaciones principales: Map(), Filter() y ReduceByKey()

Map
Se utiliza para transformar cada elemento de un RDD en otro elemento. Por ejemplo, se puede aplicar una función a cada elemento del RDD para convertirlo en una cadena o para realizar una operación. La función map crea un nuevo RDD que contiene los elementos resultantes de aplicar la transformación a cada elemento del RDD original.

# Creamos un RDD con una lista de números
numeros = sc.parallelize([1, 2, 3, 4, 5])

# Utilizamos la función map para multiplicar cada número por 2
numeros_por_dos = numeros.map(lambda x: x * 2)

# Imprimimos el nuevo RDD con los elementos resultantes
print(numeros_por_dos.collect())  # Resultado: [2, 4, 6, 8, 10]

Filter
se utiliza para seleccionar un subconjunto de elementos de un RDD que cumple una determinada condición. Crea un nuevo RDD que contiene los elementos seleccionados.

# Creamos un RDD con una lista de palabras
palabras = sc.parallelize(['hola', 'adios', 'gato', 'perro', 'casa'])

# Utilizamos la función filter para seleccionar las palabras con más de 5 letras
palabras_largas = palabras.filter(lambda x: len(x) > 5)

# Imprimimos el nuevo RDD con los elementos seleccionados
print(palabras_largas.collect())  # Resultado: ['adios']

ReduceByKey
sirve para reducir los valores de un RDD que tienen la misma clave. Por ejemplo, se puede aplicar una función a cada valor de un RDD para agregarlos y, a continuación, utilizar la función reduceByKey para agregar los valores que tienen la misma clave. Crea un nuevo RDD que contiene las claves y los valores resultantes de aplicar la función de agregación.

# Creamos un RDD con una lista de tuplas que representan ventas de productos
ventas = sc.parallelize(
[('norte', 'producto1', 100), 
('sur', 'producto1', 200), 
('norte', 'producto2', 50), 
('sur', 'producto2', 150)])

# Utilizamos la función reduceByKey para sumar las ventas de cada producto por región
ventas_por_region = ventas.reduceByKey(lambda x, y: x + y)

# Imprimimos el nuevo RDD con los resultados de las ventas por región
print(ventas_por_region.collect())  # Resultado: [('norte', 150), ('sur', 350)]

MapPartition
El método mapPartitions es útil cuando necesitas aplicar una función a un conjunto de datos que está dividido en particiones, especialmente en proyectos ML. Digamos que tienes una gran cantidad de datos que Spark divide en múltiples particiones para procesarlos más rápido. Esta función es recomendable usarla cuando deseas aplicar una transformación a cada partición de un conjunto de datos en lugar de aplicarla a cada registro individualmente (que para eso usaríamos .map()).

Por ejemplo, imagina que tienes un conjunto de datos que contiene información de ventas de diferentes tiendas en todo el país. Cada partición contiene información de ventas de una región geográfica específica. Si deseas agregar la cantidad de ventas de cada región, en lugar de aplicar una función a cada registro, puedes usar mapPartitions para aplicar la función a cada partición de datos y luego sumar los resultados.

# Crear un RDD de números
rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 3) # Dividir en 3 particiones

# Definir una función que devuelve el número máximo en cada partición
def max_partition(iterator):
    yield max(iterator)

# Aplicar la función a cada partición usando mapPartitions
max_per_partition = rdd.mapPartitions(max_partition).collect()

# Imprimir el número máximo en cada partición
print(max_per_partition)

En este ejemplo creamos un RDD con 10 números y lo dividimos en 3 particiones usando el argumento numSlices en la función parallelize. Luego, definimos una función llamada max_partition que toma un iterador de números y devuelve el número máximo en esa partición. Usamos yield en lugar de return en la función porque estamos generando un único valor para cada partición.

Finalmente, usamos mapPartitions para aplicar la función max_partition a cada partición y obtener el número máximo en cada una de ellas. El resultado es una lista con el número máximo de cada partición, que imprimimos usando la función print.

Trabajando con RDDs en Spark

Vamos a realizar el ejercicio de generar la consulta de Spark SQL que hemos visto previamente con RDD en lugar de DataFrames. Esta es la consulta original:

df_green_revenue = spark.sql("""
SELECT 
    date_trunc('hour', lpep_pickup_datetime) AS hour, 
    PULocationID AS zone,

    SUM(total_amount) AS amount,
    COUNT(1) AS number_records
FROM
    green
WHERE
    lpep_pickup_datetime >= '2020-01-01 00:00:00'
GROUP BY
    1, 2  
""")

Vamos a utilizar los datos de los taxis green que hemos descargado previamente generando el DataFrame df_green con todos los CSVs correspondientes. Para convertir un DataFrame a RDD simplemente utilizamos el operador .rdd. Vamos a comenzar convirtiendo el SELECT a RDD con el mismo operador de transformación .select() (líneas 18-20).

El WHERE lo vamos a generar utilizando el operador .filter(), para lo que se crea la función filter_outliers(row) que recibe como parámetro una Row que el propio RDD va iterando (como las funciones lambda cuando las usamos para transformar DataFrames). La función simplemente realiza la comprobación de que el la fecha de recogida (lpep_pickup_datetime ) sea a partir del 1.1.2022.

Para realizar las operaciones de agregación SUM y COUNT del GROUPBY necesitamos crear dos funciones para preparar los datos y agruparlos. En primer lugar vamos a generar un nuevo RDD con los resultados intermedios (con una Key y los valores) usando el operador .map(). Vamos a dividir cada Row en dos tuplas (clave(x,y) , valor(k,z)). La función prepare_for_grouping genera las tuplas key(hour, zone) y value (amount, count). El count es 1 por cada Row para que cuando se sumen todas nos de el número total de Rows, es decir, un count(1).

Por otro lado, vamos a crear otra función (calculate_revenue) para realizar el cálculo de agregación (sumatorio del amount y el conteo del nº de carreras) y vamos a invocarla desde reduceByKey(). Este operador infiere cuál es la clave del RDD original (columnas con los mismos valores), por lo que en cada iteración va a agrupar dos Rows con la misma Key sumando los valores amount y count.

El método unwrap lo vamos a usar para mostrar los resultados como si fueran una fila y no un conjunto de tuplas, para después generar el DataFrame de una forma más higiénica con el operador .toDF(). Si lo deseamos, podemos pasarle como parámetro el esquema de los datos.

from datetime import datetime
from collections import namedtuple
from pyspark.sql import types

import pyspark
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .master("local[*]") \
    .appName('test') \
    .getOrCreate()


start = datetime(year=2020, month=1, day=1)

df_green = spark.read.csv('data/raw/green/*/*', header=True, inferSchema=True)

rdd = df_green \
    .select('lpep_pickup_datetime', 'PULocationID', 'total_amount') \
    .rdd

def filter_outliers(row):
    return row.lpep_pickup_datetime >= start

def prepare_for_grouping(row): 
    hour = row.lpep_pickup_datetime.replace(minute=0, second=0, microsecond=0)
    zone = row.PULocationID
    key = (hour, zone)
    
    amount = row.total_amount
    count = 1
    value = (amount, count)

    return (key, value)

def calculate_revenue(left_value, right_value):
    left_amount, left_count = left_value
    right_amount, right_count = right_value
    
    output_amount = left_amount + right_amount
    output_count = left_count + right_count
    
    return (output_amount, output_count)

def unwrap(row):
    return RevenueRow(
        hour=row[0][0], 
        zone=row[0][1],
        revenue=row[1][0],
        count=row[1][1]
    )



result_schema = types.StructType([
    types.StructField('hour', types.TimestampType(), True),
    types.StructField('zone', types.IntegerType(), True),
    types.StructField('revenue', types.DoubleType(), True),
    types.StructField('count', types.IntegerType(), True)
])

df_result = rdd \
    .filter(filter_outliers) \
    .map(prepare_for_grouping) \
    .reduceByKey(calculate_revenue) \
    .map(unwrap) \
    .toDF(result_schema) 

df_result.write.parquet('tmp/green-revenue')

PicklingError: Could not serialize object: IndexError: tuple index out of range

Spark Standalone Mode

Spark Standalone Mode es un modo de ejecución de Apache Spark que permite ejecutar aplicaciones en un clúster dedicado. En este modo, Spark actúa como un administrador de clúster y controla la asignación de recursos y la planificación de tareas. El modo Standalone es una de las opciones de implementación de Spark, junto con YARN (Yet Another Resource Negotiator) y Mesos. Podemos consultar la documentación de Spark Standalone Mode para más información.

En el modo Standalone, se puede iniciar un clúster Spark simplemente ejecutando una serie de comandos para arrancar el nodo maestro (master) y varios nodos de trabajo (workers). El nodo maestro coordina el clúster y administra los recursos disponibles, mientras que los nodos de trabajo ejecutan tareas.

Para arrancar el Spark Standalone mode en Windows 10 por línea de comandos CMD debemos seguir los siguientes pasos. Antes de empezar, recuerda que tienes que tener definida la variable de entorno SPARK_HOME (si lo has hecho por Git Bash, no está en Windows). Para declararla desde línea de comandos:

setx SPARK_HOME "c:\tools\spark-3.3.2-bin-hadoop3"

1. Abrimos un prompt de CMD y nos dirigimos a la ruta donde hemos descargado Spark.

cd %SPARK_HOME%\bin

2. Ejecutamos el comando el siguiente comando para arrancar el nodo master. Nos va a generar una URL tipo spark://192.168.0.38:7077 que usaremos para arrancar el worker. Por otro lado, también arranca el Spark UI que por defecto se despliega en el puerto 8080 (http//localhost:8080).

spark-class org.apache.spark.deploy.master.Master 

3. Abrimos otro prompt de CMD y arrancamos un worker con el siguiente comando. Fíjate en que se le pasa como parámetro la URL del master que hemos iniciado en el paso preivo:

spark-class org.apache.spark.deploy.worker.Worker spark://192.168.0.38:7077

4. Creamos un nuevo cuaderno de jupyter y conectamos la SparkSession al nodo master que acabamos de levantar indicando la URL spark://192.168.0.38:7077:

spark = SparkSession.builder \
    .master("spark://192.168.0.38:7077") \
    .appName('test') \
    .getOrCreate()

5. Nos dirigimos a la Spark UI para ver cómo se han arrancado los nodos master y worker, y la app:

failed to launch: nice -n 0 /c/tools/spark-3.3.2-bin-hadoop3/bin/spark-class org.apache.spark.deploy.master.Master --host --port 7077 --webui-port 8080
  ps: unknown option -- o
  Try `ps --help' for more information.
  Error: Could not find or load main class org.apache.spark.launcher.Main
  Caused by: java.lang.ClassNotFoundException: org.apache.spark.launcher.Main
  /c/tools/spark-3.3.2-bin-hadoop3/bin/spark-class: line 96: CMD: bad array subscript
full log in /c/tools/spark-3.3.2-bin-hadoop3/logs/spark--org.apache.spark.deploy.master.Master-1-FORGEWORLD.out

¡Bien! ya tenemos arrancado nuestro propio clúster Spark en local, vamos a probar a crear scripts python que corran sobre el mismo. Para nuestra prueba vamos a aprovechar uno de los notebooks que hemos realizado previamente y vamos a convertirlo en un script python con jupyter nbconvert desde Git Bash. Una vez generado el fichero debemos limpiarlo de código autogenerado por Jupyter.

jupyter nbconvert --to=script 06_spark_sql.ipynb

Vamos a sustituir los nombres de los ficheros y los directorios que están puestos a fuego por parámetros, para lo que usaremos la librería argparse. Podéis consultar el fichero en la repo de Github:

import argparse

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

parser.add_argument('--input_green', required=True)
parser.add_argument('--input_yellow', required=True)
parser.add_argument('--output', required=True)

input_green = args.input_green
input_yellow = args.input_yellow
output = args.output

Para lanzar el script Python en el clúster local de Spark que hemos levantado, desde la terminal ejecutamos:

python my_script.py \
    --input_green=data/pq/green/2020/*/ \
    --input_yellow=data/pq/yellow/2020/*/ \
    --output=data/report-2020

Spark submit

Spark submit es una herramienta de línea de comandos en Apache Spark que permite enviar y ejecutar aplicaciones Spark en un cluster de Spark. Para enviar una aplicación Spark con spark-submit simplemente indicamos el script python y los parámetros necesarios. También proporciona opciones para especificar el número de cores y memoria que se asignarán a la aplicación en el clúster de Spark. Además, puede ser utilizado para enviar aplicaciones a diferentes modos de ejecución de Spark, incluyendo modo local, modo cluster y modo cliente.

Para utilizarlo con nuestros ejemplos, dentro del mismo script Python en lugar de conectarnos al SparkSession indicando el master, vamos a eliminar este parámetro dejándolo así:

spark = SparkSession.builder \
    .appName('test') \
    .getOrCreate()

Lo que vamos a hacer es ejecutarlo desde spark-submit y pasarle como parámetro la URL del master de Spark que hemos levantado:

spark-submit \
    --master="spark://192.168.0.38:7077" \
    my_script.py \
        --input_green=data/pq/green/2020/*/ \
        --input_yellow=data/pq/yellow/2020/*/ \
        --output=data/report-2020

Spark en Google Cloud Platform: Dataproc

Dataproc es un servicio de Google Cloud que te permite ejecutar tareas de procesamiento de datos a gran escala de manera fácil y rápida. En otras palabras, es una herramienta que te permite procesar grandes cantidades de datos en paralelo usando el poder de la nube.

Con Dataproc, puedes crear un clúster de computación en la nube y ejecutar tareas de procesamiento de datos como procesamiento por lotes, análisisy aprendizaje automático. El servicio está diseñado para ser fácil de usar, escalable y rentable, lo que lo hace ideal para empresas y organizaciones que necesitan procesar grandes cantidades de datos de manera eficiente. Además, Dataproc está integrado con otras herramientas de Google Cloud, como BigQuery, Cloud Storage y Cloud Dataprep, lo que te permite integrar y analizar datos de diferentes fuentes de manera eficiente.

Vamos a seguir los siguientes pasos para ejecutar nuestro script Python en Dataproc:

1. En primer lugar vamos a otorgar permisos sobre Dataproc al service account que estamos usando en el bootcamp. Desde el IAM lo seleccionamos y añadimos privilegios sobre el rol Dataproc Administrator.

2. Para crear una instancia de Dataproc simplemente buscamos desde en la barra de búsqueda por Dataproc. La primera vez que accedamos nos solicitará habilitar la API.

3. Creamos un nuevo cluster de Dataproc en modo Compute Engine:

3. Vamos a definir un nombre, región (debe ser la misma que nuestro Bucket (GCS)). En los siguientes pasos nos permite especificar el número de nodos (la versión Single Node es ideal para hacer pruebas, sólo incluye un nodo y sale más económico). Por último, podemos especificar que el clúster de Dataproc incluya ciertos componentes (Docker, Anaconda, Jupyter Notebook…).

Ejecutar job Spark en Dataproc

Hay varias formas de ejecutar un job de Spark en Dataproc. En el curso vemos desde el UI de la GCP y por línea de comandos con la SDK gcloud, pero también podríamos lanzarlo a través del API, node.js, etc.. Más info en la documentación de Google.

Ejecutar job Spark desde Google Cloud Patform UI

Ejecutamos un job desde el UI simplemente entrando en el Dataproc Cluster y pulsando en el botón Submit job en el menú superior. Se abrirá un blade donde podemos configurar las propiedades del mismo:

• Job ID: identificador del job
• Job Type: lenguaje de programación del script, en nuestro caso, PySpark.
• Main python file: ubicación física del script Python (porque hemos seleccionado PySpark como type). Vamos a utilizar nuestro Bucket para alojar el script, para subirlo podemos hacerlo a mano desde el UI o utilizando el SDK gcloud que instalamos en la primera semana del curso. Recuerda que no debemos especificar en el SparkSession el clúster master, debemos utilizar el script del ejemplo de spark-submit. Por tanto, introducimos la ruta física del script en nuestro bucket: gs://dtc_data_lake_digital-aloe-375022/code/06_spark_sql_big.py. Para copiarlo de local al Bucket:

gsutil cp 06_spark_sql.py gs://dtc_data_lake_digital-aloe-375022/code/06_spark_sql_big.py

• Jar files: para este ejemplo no es necesario, ya que sólo ejecutamos un script Python.
• Arguments: aquí especificamos los parámetros que espera recibir el script, recuerda que son tres. Se añaden uno por uno (como si fueran tags) incluyendo el prefijo de «–«. Revisa la ubicación de los ficheros en tu bucket.

--input_green=data/green/2020/*/
--input_yellow=data/yellow/2020/*/ 
--output=data/report-2020

3. Pulsamos sobre el botón submit y comienza a ejecutarse de inmediato.

Ejecutar job Spark desde gcloud SDK

Otra forma de lanzar un job de Spark en Dataproc es usando el gcloud SDK. Abrimos una terminal y lanzamos el siguiente comando. Como nuestro script es un Python con PySpark, indicamos pyspark en job-command. Especificamos el nombre del clúster Dataproc que hemos instanciado, la región y por último

gcloud dataproc jobs submit pyspark \
    --cluster=de-zoomcamp-cluster \
    --region=europe-west6 \
    gs://dtc_data_lake_de-zoomcamp-nytaxi/code/06_spark_sql.py \
    -- \
        --input_green=gs://dtc_data_lake_de-zoomcamp-nytaxi/pq/green/2020/*/ \
        --input_yellow=gs://dtc_data_lake_de-zoomcamp-nytaxi/pq/yellow/2020/*/ \
        --output=gs://dtc_data_lake_de-zoomcamp-nytaxi/report-2020

Ejemplo:

gcloud dataproc jobs submit pyspark \
    --cluster=<your-cluster-name> \
    --region=europe-west6 \
    gs://<url-of-your-script> \
    -- \
        --param1=<your-param-value> \
        --param2=<your-param-value>

Job Spark + Dataproc + BigQuery

Por último, vemos cómo ejecutar un job en Dataproc para que ingeste y transforme los datos para finalmente guardarlos en una tabla de BigQuery en lugar de ficheros parquet.

1. Vamos a crear otro script de Python copia de 06_spark_sql.py donde modificaremos dónde queremos persistir el output. En lugar de un fichero parquet, vamos a guardarlos en una tabla de BigQuery. Modificamos la última línea del .write(). En .option() indicamos que es una tabla y la variable output es uno de los parámetros que se recibe al ejecutar el job.

df_result.write.format('bigquery') \
    .option('table', output) \
    .save()

2. Subimos el nuevo script al Bucket de Google:

gsutil cp 06_spark_sql_big_query.py gs://dtc_bucket/code/06_spark_sql_big_query.py

3. Vamos a crear un nuevo job de Spark donde además de seguir la misma configuración del ejemplo anterior, vamos a añadir la librería gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar en el apartado de jar files. Para más información, consulta la documentación.

gcloud dataproc jobs submit pyspark \
    --cluster=de-zoomcamp-cluster \
    --region=europe-west6 \
    --jars=gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar \
    gs://dtc_data_lake_de-zoomcamp-nytaxi/code/06_spark_sql_big_query.py \
    -- \
        --input_green=gs://dtc_data_lake_de-zoomcamp-nytaxi/pq/green/2020/*/ \
        --input_yellow=gs://dtc_data_lake_de-zoomcamp-nytaxi/pq/yellow/2020/*/ \
        --output=trips_data_all.reports-2020

4. Lanzamos el job y observamos cómo se ha creado la tabla en nuestro esquema de BigQuery:

La entrada Data Engineering Zoomcamp – Semana 5 se publicó primero en Marcos Torregrosa.

Cada vez se pone más interesante el bootcamp para Data Engineers. La cuarta semana está dedicada al modelaje de datos y los Analytics Engineer como pieza intermedia entre los Data Engineers y los Data Scientists. Su labor es hacer de catalizador entre ambos roles participando en el modelado de datos para lo que hemos explorado dbt (Data build tool). El último capitulo está centrado en la visualización de datos con Google Data Studio, disponible dentro del ecosistema de GCP (Google Cloud Platform).

• Repo Bootcamp: https://github.com/DataTalksClub/data-engineering-zoomcamp/tree/main/week_4_analytics_engineering
• Repo personal con proyecto dbt: https://github.com/hegdehog/zoomcamp-2023-dbt
• Repo personal con ejercicios: https://github.com/hegdehog/data-engineering-zoomcamp-2023

Última actualización: 20/02/2023

Prerequisitos

Vamos a seguir utilizando los recursos que hemos desplegado en GCP (BigQuery y GCS Bucket). La novedad esta semana es la introducción de una nueva herramienta para el modelado de datos:

• dbt (Data Build Tool). Podemos utilizarla de dos formas, bien mediante el SaaS gratuito para usuarios individuales registrándonos en la web dbt Cloud o descargando el paquete para trabajar en local y con la consola CLI. Se puede descargar de varias formas, la más sencilla a mi parecer es con pip en nuestro entorno de Python. Es muy interesante el curso de introducción oficial (duración: 5 horas).
• Google Data Studio. Nuevo recurso que probamos en Google Cloud Platform, simplemente lo desplegamos. Podemos acceder desde aquí.

Mas allá de software y herramientas, vamos a necesitar cargar en nuestra base de datos de BigQuery los datasets de los ejercicios:

• Datasets de las carreras de taxis: he dejado en mi repositorio de github un script de Prefect para cargar todos los ficheros en GCS Bucket para luego trabajar con ellos en dbt y persistirlos en la base de datos BigQuery.
  • Yellow taxi data – Años 2019 y 2020
  • Green taxi data – Años 2019 y 2020
  • fhv data – Año 2019
• Dataset de zonas para lookup: puedes descargarlo desde el repositorio original aquí
• Schema, staging models y macros para generar los primeros modelos en dbt. Están publicados en la repo del bootcamp (macros y schema-models)

Analytics Engineer

Según el estudio What Are the Essential Roles for Data and Analytics? de Gartner publicado en el año 2022 los perfiles de D&A (Data and analytics) llevan varios años en continúo desarrollo y para este 2023 se espera que continúen surgiendo nuevos perfiles que respondan a necesidades más concretas:

Dentro de este extenso e imparable abanico de roles, los Analytics Engineers son un híbrido a medio camino entre los Data Engineers y los Data Scientists. Se encargan de diseñar, construir y mantener sistemas de análisis y visualización para facilitar la toma de decisiones basadas en datos. Similar a un ingeniero de datos, se encarga de crear canalizaciones y visualizaciones, pero principalmente para modelos de IA/ML y equipos de ciencia de datos.

En comparación, los data engineers se enfocan en la recopilación, integración y almacenamiento de datos, mientras que los data scientists se enfocan en la investigación y modelado de datos para el descubrimiento de patrones y tendencias. La tarea principal de un Analytics Engineer es proporcionar una plataforma de análisis de datos sólida y confiable para que los data scientists y otros perfiles puedan realizar análisis complejos.

Es importante que los Analytics Engineer estén familiarizados con una amplia gama de tecnologías, ya que la capacidad de trabajar con diferentes herramientas y plataformas es clave para la solución efectiva de problemas de análisis de datos. El stack tecnológico necesario puede variar dependiendo de los requisitos específicos del proyecto y la organización, pero algunas tecnologías comunes que se utilizan en este campo incluyen:

• Almacenamiento de datos: Data warehousing, sistemas de almacenamiento en la nube como Azure Blob Storage, Amazon S3 o Google Cloud Storage.
• Procesamiento de datos: Apache Hadoop, Apache Spark, Apache Flink.
• Lenguajes de programación: Python, SQL, Java.
  • Bibliotecas de análisis de datos: Pandas, Numpy, Matplotlib, etc.
• Visualización de datos: Tableau, Power BI, QlikView, Looker.
• Colaboración y gestión de proyectos: Azure DevOps, Github, JIRA, Confluence, Asana.
• Integración de datos: Apache NiFi, Apache Airflow, Prefect, Talend.

Modelización de Data Warehouse: Kimball, Inmon y Data Vault

Existen varias metodologías de Data Warehouse basadas en la arquitectura de la modelización, las más extendidas son las de Kimball, Inmon y más recientemente, Data Vault.

Data Build Tool (dbt)

dbt (Data Build Tool) es una librería Python open-source que agiliza la construcción de modelos de datos permitiendo a los desarrolladores definir, orquestar y ejecutar transformaciones en un almacén de datos moderno como BigQuery, Snowflake, Redshift, etc. Podríamos decir que se trata de una herramienta de gobierno focalizado en la «T» de un proceso ETL/ELT, nos permite centralizar y construir todas las transformaciones de datos en SQL organizándolas como módulos reutilizables (models). Por otro lado, al estar inspirado en las prácticas de la ingeniería software, podemos crear test de validación e implementar todo el ciclo de CI/CD en nuestras canalizaciones de datos. En paralelo al conocimiento que aporta el Bootcamp, es muy interesante el curso de introducción oficial (duración: 5 horas).

En mi periodo profesional anterior la lógica estaba repartida en procedimientos almacenados en el servidor de BBDD SQL Server, queries SQL en las ETL (Azure Data Factory, SSIS y Taled) e incluso en los modelos de carga de las herramientas de visualización (Qlikview y Power BI). No estaba documentada en ningún lado. Lo que aporta dbt en este paradigma es una capa de gobierno o control que facilita el mantenimiento y documentación de la lógica, el linaje, aumentar la resiliencia y colaboración gracias a un control de versiones y por último, facilitaría la integración continua / entrega o despliegue continuo CI/CD.

Algunas de las características principales de dbt.

1. Reutilización de código: Permite la definición de modelos de datos y la organización de transformaciones en paquetes.
2. Énfasis en los controles de calidad: fomenta el uso de pruebas automatizadas para garantizar la calidad de los datos y prevenir errores en las transformaciones.
3. Control de versiones y colaboración: está diseñado para trabajar con sistemas de control de versiones como Git, Bitbucket, etc., lo que facilita el seguimiento de los cambios y la colaboración en el desarrollo de pipelines de datos.
4. Escalabilidad: diseñado para trabajar con almacenes de datos modernos como BigQuery, Snowflake, Redshift, etc., permite escalar fácilmente el procesamiento de grandes volúmenes de datos.

¿Cómo empezar con dbt?

Hay dos formas de utilizar dbt de forma gratuita:

• dbt Core: versión de código abierto (open-source) que se instala localmente o en un servidor propio. La interacción es por consola CLI.
• dbt Cloud: plataforma alojada en la nube (SaaS) que ofrece funcionalidades adicionales a la versión Core (programación de ejecuciones, integraciones con servicios de BI, monitoreo y alertas). Es más fácil de utilizar al disponer de un GUI. Además de los planes de pago, ofrece una versión gratuita limitada para desarrolladores.

Instalar dbt Core con PIP

Tenemos varias opciones para instalar dbt Core en nuestro equipo o servidor local, la forma más sencilla es mediante pip en un entorno de python.

pip install dbt-core

Después instalaremos el adapter o conector del motor de base de datos que vayamos a utilizar. Tenemos a nuestra disposición un catálogo oficial y otro complementario de la comunidad, puedes consultar todos los conectores disponibles desde aquí. En nuestro caso vamos a hacer la instalación del adapter de BigQuery.

pip install dbt-bigquery

error: Microsoft Visual C++ 14.0 or greater is required. Get it with "Microsoft C++ Build Tools":
https://visualstudio.microsoft.com/visual-cpp-build-tools/
      [end of output]

  note: This error originates from a subprocess, and is likely not a problem with pip.
  ERROR: Failed building wheel for pandas
Failed to build pandas
ERROR: Could not build wheels for pandas, which is required to install pyproject.toml-based projects

Al intentar instalar el adapter de BigQuery me arrojó un error por una dependencia con la librería Pandas que necesita como mínimo la versión 14.0 del runtime de Microsoft Visual C++. Descargamos Build Tools desde aquí o el enlace facilitado en el mensaje de error. Lo ejecutamos y pulsamos en Modificar para añadir el paquete que nos falta:

Volvemos a instalar el adapter de BigQuery y debería terminar con éxito.

ERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.
prefect-gcp 0.2.3 requires google-api-python-client>=2.20.0, but you have google-api-python-client 1.7.11 which is incompatible.

pip install google-api-python-client --upgrade 

python -m venv PythonDbt
PythonDbt\Scripts\activate

Creamos un nuevo proyecto ejecutando el comando dbt init. Lo que hace este comando es clonar en la carpeta desde donde lo hayamos ejecutado el starter project de dbt que contiene todos los ficheros y directorios necesarios para arrancar nuestro proyecto

• dbt_project.yml: fichero de configuración del proyecto dbt (nombre, profile que identifica el motor de base de datos que vamos a utilizar, como PostgreSQL o BigQuery y variables globales). Si vas a usar dbt de forma local es necesario que asegurar que el profile indicado en este fichero de configuración coincida con el de la instalación (~/.dbt/profiles.yml).
• Readme.MD: fichero para literatura en la repo
• Directorios analysis, data, macros, models, snapshots y tests

dbt init

$ dbt init
21:47:00  Running with dbt=1.4.1
No adapters available. Go to https://docs.getdbt.com/docs/available-adapters

Si todo ha ido bien, podremos configurar nuestro proyecto a través de la consola (CLI) para generar el fichero profiles.yml:

Comprobamos que se han generado todos los ficheros y directorios del proyecto dbt en la ruta en la que hemos ejecutado el comando:

Instalar dbt Core con una imagen de Docker

Os recomiendo esta lectura para profundizar más en este paso. Las imágenes disponibles para montar un contenedor de dbt con Docker son:

• dbt-core (no tiene soporte de base de datos)
• dbt-postgres
• dbt-redshift
• dbt-bigquery
• dbt-snowflake
• dbt-spark
• dbt-third-party 
• dbt-all (instala todas las imágenes en una sola)

docker build --tag my-dbt  --target dbt-bigquery .

Una vez creada la imagen arrancamos el contenedor:

docker run \
--network=host
--mount type=bind,source=path/to/project,target=/usr/app \
--mount type=bind,source=path/to/profiles.yml,target=/root/.dbt/profiles.yml \
my-dbt \
ls

Crear proyecto en DTB Cloud

Previo a crear nuestro primer proyecto en dbt Cloud debemos reunir los ingredientes necesarios: crear una service account, generar la JSON Key para otorgar acceso a dbt a nuestra instancia de BigQuery y crear una repo vacía en Github donde se almacenarán los ficheros y directorios del proyecto:

Crear service account + JSON Key para Big Query

Como en nuestro caso vamos a utilizar BigQuery, la autenticación se realiza por BigQuery OAuth. Debemos crear un service account desde GCP y descargar la JSON key para otorgar acceso a dbt.

1. Accedemos a la consola de Google Cloud Platform y nos dirigimos a IAM and admin > Service accounts para crear un nuevo service account con permisos BigQuery Admin y Storage Object Viewer.
2. Descargamos la JSON Key a nuestro equipo (más tarde la subiremos en dbt en el proceso de configuración del proyecto).

Crear repositorio en Github

Simplemente creamos un repo vacía en Github y pulsamos en Git Clone para copiar la clave SSH con la que vincularemos el proyecto de dbt. La conexión con Github se hace en dos partes, por ahora sólo nos interesa la key, más adelante veremos cómo configurar la deploy key generada desde dbt.

¡Ahora sí tenemos los ingredientes! Nos registramos en dbt cloud con la opción gratuita para un usuario desde este enlace. Una vez verificado el email podremos crear nuestro proyecto eligiendo en primer lugar el almacenamiento de datos.

En el siguiente paso cargamos la JSON Key que hemos generado con el service account de BigQuery y automáticamente se cargan todos los parámetros. Es recomendable crear el dataset en BigQuery de forma manual para evitar problemas de incompatibilidad entre regiones. En mi caso, he usado el nombre que ha generado por defecto y creado el dataset directamente:

Hacemos el test de conexión y si todo ha ido bien ¡al paso siguiente!

Ahora es el turno de configurar el repositorio Github que hemos creado previamente y realizar el segundo paso que teníamos pendiente. Seleccionamos Git Clone y pegamos la Key SSH que hemos copiado antes. Pulsamos el botón de Import.

Generará una clave de despliegue que debemos copiar en la configuración del repositorio de Github:

Volviendo a nuestro repositorio de Github, pulsamos en Settings y en la sección de Security pulsamos en Deploy Keys para añadirla. Es necesario marcar la opción Allow write access:

¡Listo!

Si damos al botón Next en la configuración del proyecto de dbt ya habremos terminado:

Accedemos a Develop y debemos inicializar nuestro proyecto en dbt cloud (similar al comando dbt init que ejecutaríamos en dbt core):

Tras esperar unos segundos, se crean todos los ficheros de yml, SQL y los directorios del proyecto dbt. Debemos pulsar sobre Commit and sync para hacer un push a nuestra repo de Github.

Recuerda que al estar vinculado a una repo en github, si quieres trabajar sobre el GUI de dbt cloud es necesario que crees primero una rama. Para ejecutar cualquier comando desde GUI, podemos usar la consola que tenemos en el pie:

dbt model

Un modelo en dbt es la unidad mínima con la que vamos a trabajar, se trata simplemente de una fichero SQL con una sentencia SELECT. En el siguiente ejemplo se va a crear una tabla en la BBDD de destino con los datos del SQL:

{{      config(materialized='table')    }}

SELECT *
FROM staging.source_table
WHERE record_state = 'ACTIVE'

Además de la query, un modelo dbt incluye al principio un bloque de código bajo la notación Jinja que reconoceremos por los doble brackets {{ }}. Dentro de este bloque se llama a la función config() de dbt que habitualmente se usa para especificar la estrategia de persistencia del modelo dbt en la base de datos de destino. Por defecto hay cuatro formas de materializar las consultas, aunque es posible crear las nuestras propias:

• table: los datos del modelo se persisten en una tabla sobre el warehouse
• view: ídem al anterior, pero en lugar de una tabla se trata de una vista
• incremental: estos modelos permiten a dbt insertar y/o actualizar registros en una table si han cambiado desde la última vez que se ejecutó.
• ephemeral: no generan un objeto directamente en la base de datos, crea un CTE (Common Table Expression), que viene a ser una subquery temporal para utilizar en otras consultas (como el WITH de SQL Server).

Para construir el modelo que persiste la query en una tabla de nuestra base de datos, debemos ejecutar el comando dbt build. Si no añadimos ningún parámetro, compilará y construirá todos los models. Para especificar que sólo queremos construir uno en concreto añadimos el parámetro --select; veámoslo con dos ejemplos:

Ejecutamos y construimos todos los modelos:

dbt run

Para construir sólo el model myModel.sql corremos este comando:

dbt run --select myModel.sql

Cuando ejecutamos el modelo, dbt lo compila en la siguiente query:

CREATE TABLE my_schema.my_model AS (
    SELECT *
    FROM staging.source_table
    WHERE record_state = 'ACTIVE'
)

La cláusula FROM en un modelo dbt

Además de indicar directamente esquema y nombre de la tabla, en un modelo dbt hay disponibles dos formas de configurar el origen de los datos en la cláusula FROM: sources y seeds.

{{ config(materialized='table') }}

SELECT *
FROM {{ source('staging','green_tripdata') }}

version: 1

sources:
    - name: staging
      database: dbt_mtorregrosa
      schema: trips_data_all

      tables:
          - name: green_tripdata
          - name: yellow_tripdata

Por tanto, en nuestra carpeta de /models/ de dbt tendremos un fichero .sql para cada modelo y un schema.yml con la configuración de los sources.

Seeds

Similar a las External Tables de BigQuery o Azure Synapse, podemos referenciar cualquier fichero CSV almacenado en el repositorio dentro del directorio /seeds, Al estar almacenado en una repo podemos aprovechar su control de versiones. Se recomiendo usar seeds para datos que no cambian con frecuencia (tablas de dimensiones paramétricas, como provincias).

En lugar de usar la macro source() como hemos visto antes para una tabla de BBDD, para los seeds se usa la macro ref() que recibe como parámetro el nombre del fichero y dbt automáticamente detecta las dependencias y su ubicación. La macro ref() también se puede usar para referenciar tablas o vistas de BBDD del mismo modo, sólo pasando por parámetro el nombre de la tabla. Para crear un seed simplemente subimos un fichero CSV al directorio /seeds de nuestra repo y ejecutamos el comando dbt seed taxi_zone_lookup.csv. Si ejecutasemos dbt seed se cargarían en la base de datos todos los CSVs del directorio. Este comando genera una tabla en nuestra bbdd, en BigQuery:

dbt seed taxi_zone_lookup.csv

Si accedemos a BigQuery observamos que se ha creado una tabla con nombre taxi_zone_lookup que podremos consultar con ref() en cualquier dbt model.

Ahora podemos crear un dbt model que referencie al seed recién creado:

{{ config(materialized='table') }}

SELECT *
FROM {{ ref('taxi_zone_lookup') }}

Si quisiéramos cambiar los tipos de datos del seed que por defecto a inferido del CSV, debemos modificar el fichero de configuración del proyecto dbt_project.yml y añadir el bloque de seeds:

seeds:
    taxi_riders_ny:
       taxi_zone_lookup:
          +column_types:
              locationid: numeric

Si queremos recrear el seed debemos utilizar el comando dbt seed micsv.csv --full-refresh.

Macros

Una macro en dbt es similar a una función en cualquier otro lenguaje redactadas en notación jinja. Se generan en ficheros independientes .sql en el directorio /macros del proyecto dbt. Como ya hemos comentado, por defecto dbt dispone de varias macros como source(), ref() y config(), pero podemos crear una nueva que atienda a nuestras necesidades. Algunas características de las macros:

• Permite añadir lógica con bucles FOR y sentencias IF
• Pueden utilizar variables de entorno definidas a nivel de proyecto en dbt
• Permite reutilizar código entre distintos models
• Utiliza en una consulta el resultado de otra subconsulta

Podemos utilizar tres tipos de bloques jinja distintos dentro de una macro:

• Expresiones {{ }}: cuando queremos devolver una cadena. Se pueden utilizar expresiones para referenciar variables o llamar a otras macros.
• Sentencias {% %}: se utilizan para el control del flujo, por ejemplo, para loops FOR o sentencias IF.
• Comentarios {# #}: el texto del comentario no se compila, nos permite indicar notas.

Por ejemplo, en el siguiente código vamos a crear la macro get_payment_type_description que recibe como parámetro un valor en payment_type y resuelve el CASE WHEN devolviendo el valor correspondiente:

{# This macro returns the description of the payment_type #}

{% macro get_payment_type_description(payment_type) %}

    case {{ payment_type }}
        when 1 then 'Credit card'
        when 2 then 'Cash'
        when 3 then 'No charge'
        when 4 then 'Dispute'
        when 5 then 'Unknown'
        when 6 then 'Voided trip'
    end

{% endmacro %}

Usamos la macro en nuestro modelo dbt:

{{ config(materialized='table') }}

SELECT 
    get_payment_type_description(payment_type)
FROM{{ source('staging','green_tripdata') }}
WHERE vendorid is not null

Cuando ejecutamos el modelo, dbt compila el código de forma que la tabla generada ha traducido la macro a un CASE WHEN en SQL:

SELECT 
    case payment_type
        when 1 then 'Credit card'
        when 2 then 'Cash'
        when 3 then 'No charge'
        when 4 then 'Dispute'
        when 5 then 'Unknown'
        when 6 then 'Voided trip'
    end as payment_type_description
FROM {{ source('staging','green_tripdata') }}
WHERE vendorid is not null

Packages

Nos permite reutilizar macros entre distintos proyectos, similar a las librerías o módulos en otros lenguajes de programación. Para utilizar un package en nuestro proyecto debemos crear un fichero de configuración packages.yml en el directorio raíz de nuestro proyecto dbt.

packages:
  - package: dbt-labs/dbt_utils
    version: 0.8.0

Después debemos instalarlos ejecutando el comando dbt deps que se encarga de descargar todas las dependencias y ficheros del paquete dentro de nuestro proyecto. Una vez haya terminado se creará el directorio dbt_packages/dbt_utils en nuestro proyecto.

Podemos utilizar las macros del package recién instalado en cualquier model de nuestro proyecto. Por ejemplo, el package dbt-utils incluye la macro surrogate_key para crear una clave subrogada (id secuencial interno a la tabla).

{{ config(materialized='table') }}

SELECT
    {{ dbt_utils.surrogate_key(['vendorid', 'lpep_pickup_datetime']) }} as tripid,
    vendorid
FROM {{ source('staging','green_tripdata') }}
WHERE vendorid is not null

Dbt compila este modelo creando una clave subrogada con una función hash:

SELECT
    to_hex(md5(cast(coalesce(cast(vendorid as string), '') || '-' 
        || coalesce(cast(lpep_pickup_datetime as string), '') as string))) as tripid,
    vendorid
FROM {{ source('staging','green_tripdata') }}
WHERE vendorid is not null

Variables

Similar a las variables de cualquier lenguaje de programación. Con la misma finalidad, nos permite guardar un valor y reutilizarlo en cualquier parte del proyecto. Se pueden crear de dos formas:

• En el fichero de configuración del proyecto dbt (dbt_project.yml) ubicado en el directorio raíz añadiendo:

vars:
    payment_type_values: [1, 2, 3, 4, 5, 6]

• Por línea de comandos cuando construimos los modelos:

dbt build --var 'is_test_run: false'

Para utilizar una variable debemos usar la macro var() dentro de un modelo:

{{ config(materialized='table') }}

SELECT *
FROM {{ source('staging','green_tripdata') }}
{% if var('is_test_run', default=true) %}

    limit 100

{% endif %}

Caso de uso: crear dbt models para proyecto tripdata

Vamos a cargar en nuestro proyecto dbt los datasets del bootcamp y crear todos los artefactos (models, macros y schema). Repito los prerequesitos de este post:

1. Cargamos en nuestro GCS Bucket todos los datasets comprimidos en formato csv.gz:

• Datasets de las carreras de taxis: he dejado en mi repositorio de github un script de Prefect para cargar todos los ficheros en GCS Bucket para luego trabajar con ellos en dbt y persistirlos en la base de datos BigQuery.
  • Yellow taxi data – Años 2019 y 2020
  • Green taxi data – Años 2019 y 2020
  • fhv data – Año 2019
• Dataset de zonas para lookup: puedes descargarlo desde el repositorio original aquí

3. Una vez hemos cargado en nuestro GCS Bucket todos los ficheros csv.gz de tripdata, vamos a crear las external tables para consumirlos como tablas de BigQuery:

CREATE OR REPLACE EXTERNAL TABLE trips_data_all.fhv_tripdata
OPTIONS (
  format = 'CSV',
  uris = ['gs://tripdata_n4gash/data/fhv/*.csv.gz']
);

CREATE OR REPLACE EXTERNAL TABLE trips_data_all.green_tripdata
OPTIONS (
  format = 'CSV',
  uris = ['gs://tripdata_n4gash/data/green/*.csv.gz']
);

CREATE OR REPLACE EXTERNAL TABLE trips_data_all.yellow_tripdata
OPTIONS (
  format = 'CSV',
  uris = ['gs://tripdata_n4gash/data/yellow/*.csv.gz']
);

CREATE OR REPLACE EXTERNAL TABLE trips_data_all.zones_tripdata
OPTIONS (
  format = 'CSV',
  uris = ['gs://tripdata_n4gash/data/taxi+_zone_lookup.csv']
);

3. Creamos en dbt el schema, staging models y macros. Están publicados en la repo del bootcamp (macros y schema-models), simplemente copiamos el código en nuestra repo de github o directamente en dbt clout y compilamos:

dbt build

Compilation Error in test accepted_values_stg_green_tripdata_Payment_type__False___var_payment_type_values_ (models/staging/schema.yml)
  'NoneType' object is not iterable
  
  > in macro test_accepted_values (tests/generic/builtin.sql)
  > called by test accepted_values_stg_green_tripdata_Payment_type__False___var_payment_type_values_ (models/staging/schema.yml)

vars:
  payment_type_values: [1, 2, 3, 4, 5, 6]

BigQuery adapter: 404 Not found: Dataset was not found in location EU

No hemos configurado correctamente la location del conector a BigQuery en dbt. Para revisar la configuración en dbt nos dirigimos a Account settings > Project Analytics > BigQuery > bajamos hasta abajo del todo en Location y escribimos exactamente la location que vemos en GCP (EU):

Si todo ha ido bien, vemos que al ejecutar el comando dbt build va a compilar y generar todos los modelos:

Y si comprobamos en BigQuery, podemos ver que se han creado dos vistas (green y yellow):

4. Creamos un seed de la tabla lookup de zonas. Podemos subir el CSV directamente al directorio /seeds en dbt o crear un fichero y copiar el contenido. Después ejecutamos el comando dbt seed para crear la external table en BigQuery y observamos cómo se ha creado en nuestro dataset la tabla taxi_zones_lookup.

5. Creamos un dbt model del seed de zonas en la carpeta models/core con nombre dim_zones.sql:

{{ config(materialized='table') }}

select 
    locationid, 
    borough, 
    zone, 
    replace(service_zone,'Boro','Green') as service_zone
from {{ ref('taxi_zone_lookup') }}

6. Vamos a unir las tablas de hechos (stg_green_tripdata y stg_yellow_tripdata) y la el seed lookup (dim_zones) en una nueva tabla usando el script fact_trips.sql que encontramos en la repo del bootcamp. Creamos un nuevo dbt model en la carpeta models/core con nombre fact_table.sql con el contenido. Si lo copiáis de la repo, recordad que debéis revisar los nombres de las tablas en vuestro propio proyecto. Lanzamos el comando dbt build. Si todo ha ido bien vemos que cada uno de los pasos se ha ejecutado correctamente:

7. Comprobamos el linaje del dbt model fact_trips y observamos cómo se identifican todas las dependencias desde la capa de origen, a staging y por último la tabla final.

8. Por último, vemos cómo se ha creado la tabla fact_trips en BigQuery:

Testing en dbt

Una de las herramientas de las que disponemos los desarrolladores para certificar que nuestro código es correcto, hace lo que se espera y no afecta a lo que hubiera consolidado antes, es pasar varios tests automáticos. Pueden realizar distintas operaciones, por ejemplo, si nuestro proyecto está compuesto por varios métodos podríamos crear un test por cada una de las posibles salidas de cada método. Esto nos permite comprobar si todo sigue funcionando con normalidad y la nueva funcionalidad aporta el valor esperado. Si alguno de los test automáticos no pasa satisfactoriamente, el código no se despliega en el siguiente entorno.

En el mundo de datos el testing es más ortopédico en el sentido de que las validaciones que podemos construir son suposiciones en base a los datos: conteos, sumatorios, cálculo de KPIs, etc. Es decir, se trataría de sentencias SELECT cuyo resultado vamos a comparar con un valor esperado. En función del proyecto, podemos encontrarnos con frameworks de data quality o simplemente un puñado de queries que se ejecutan antes de desplegar los cambios en produccion.

En dbt podemos crear tests automáticos de forma sencilla usando sentencias SELECT. Cuando lanzamos un test en dbt, éste compila el SQL y nos devuelve el número de registros que no lo han cumplido. Podemos usar tres tipos de tests en dbt:

• Singular test: es la forma más sencilla de crear un test en dbt. Simplemente es una query donde definimos exactamente lo que deseamos comprobar para un caso de uso concreto. Esto implica que no se puede reutilizar, sólo nos sirve para un modelo. Se crea como un fichero .sql y se almacena en la carpeta de tests /tests o en la que hayamos definido en dbt-project.yml. Podemos asignarle el nombre que deseemos, ya que se trata de un test único para un sólo model.
• Generic test: como su nombre indica, se trata de tests genéricos que podemos reutilizar para cualquier model. Se trata de queries parametrizadas que aceptan parámetros de forma que podamos usarla para cualquier caso de uso.
• Out-of-the-box test: de forma predeterminada dbt dispone de 4 tests genéricos:
  • unique: la columna tiene todos los valores únicos
  • Not null: comprueba si tiene valor nulo (sin informar)
  • accepted_values: similar a los check constrings en una BBDD, se trata de una lista de valores que puede contener la columna. Se construye como un array de strings: values: ['placed', 'shipped', 'completed', 'returned']
  • relationships: comprueba la IR del modelo. Por ejemplo, cada valor de la columna debe existir como PK en otra tabla.

Los tests se definen en un fichero yml en la carpeta models, por ejemplo, schema.yml, donde definimos las reglas de validación para cada columna. Por ejemplo, usando los tests out-of-the-box. Para utilizar un test que hayamos creado nosotros, debemos usar el nombre del .sql del mismo. Por cada test podemos indicar la severidad: error o warn:

columns:
          - name: tripid
            description: Primary key for this table, generated with a concatenation of vendorid+pickup_datetime
            tests:
                - unique:
                    severity: warn
                - not_null:
                    severity: warn

Podemos ejecutar todos los tests del proyecto con el comando dbt_test. Al igual que para construirlos, si indicamos en el parámetro --select el nombre del model, sólo se ejecutarán los correspondientes.

dbt test --select stg_green_tripdata

Documentación automática

Podemos configurar nuestro proyecto dbt para que genere de forma automática la documentación en formato web y publicada en dbt cloud. Incluye información del proyecto:

• Models: script de los ficheros .sql y el compilado. Incluye sus dependencias con otros objetos en dbt (models, seeds…)
• Sources: tablas de base de datos origen
• DAG (Directed acyclic graph) de las macros ref() y source() son diagramas directos acíclicos dirigidos, una forma de representar el flujo de datos en una ETL donde la información viaja desde el punto A hasta el punto B.
• Descripciones ubicadas en los ficheros .yml
• Tests
• Información sobre los datos de las BBDD origen (nombre de columnas y tipo de datos, estadísticas de tablas, etc.)

Despliegue de un model dbt

Con dbt podemos configurar CI/CD (integración continua y despliegue/entrega continua) para desplegar nuestros models en producción. Separar los entornos de desarrollo y despliegue nos permite construir modelos y probarlos sin afectar a producción. En función del proyecto podemos encontrar que el entorno de despliegue o producción está ubicado en otra máquina o simplemente en otro esquema y usuario. El flujo habitual de desarrollo-despliegue sería:

1. Creamos una nueva rama en la repo
2. Abrimos una pull request para unir nuestra rama a la principal
3. Unimos (merge) nuestra rama con la principal
4. Ejecutamos los nuevos models en el entorno de producción
5. Programamos la ejecución de los models usando jobs.

Jobs en dbt

Dbt incluye un programador donde configurar jobs para que ejecuten los modelos en producción. Los jobs se pueden programar o ejecutar manualmente y cada uno puede tener uno o varios comandos (construir varios modelos, lanzar tests, etc).

Previo a crear el job, lo primero que tendríamos que hacer es configurar un entorno de producción para así tener diferenciados el de desarrollo del de despliegue. Para crearlo nos dirigimos a Deploy > Environments y pulsamos sobre el botón Create Environment. En Deployment credentials debemos indicar el dataset de BigQuery de destino. Podemos crearlo previamente desde GCP o escribir un nombre y dbt automáticamente lo creará:

Ahora sí podríamos crear un job para que publique los cambios de desarrollo en producción. Para ello, vamos a pulsar en Deploy > Jobs y en el botón Create Job. Vamos a especificar un nombre:

Marcamos que genere la documentación y añadimos los comandos: dbt seed (crea el seed del CSV de zonas), dbt run (ejecuta y compila los models) y dbt build (para construirlos).

Por último, configuramos la programación en el apartado de triggers:

Una vez ejecutado el job, comprobamos que todos los pasos han terminado correctamente y se ha generado la documentación:

Y que en BigQuery se han creado las tablas y vistas que teníamos en el dataset de staging:

Integración continua (CI)

La integración continua (CI) es una práctica de desarrollo de software que se utiliza para garantizar que el código se integra y se prueba automáticamente de manera regular y frecuente.

En CI, los desarrolladores envían su código a un repositorio compartido varias veces al día, lo que desencadena una serie de procesos automatizados que incluyen compilación, prueba y análisis estático de código. Estos procesos se ejecutan automáticamente en un entorno de prueba aislado, que se crea y destruye para cada ciclo de integración, lo que garantiza que el código se pruebe en un entorno limpio y repetible.

El objetivo de la integración continua es detectar y solucionar problemas en el código de manera temprana, lo que ayuda a reducir el tiempo y el costo de corrección de errores más adelante en el ciclo de desarrollo. Además, al integrar y probar regularmente el código, se mejora la calidad del mismo y se facilita el proceso de entrega de software.

Podemos emplear la integración continua (CI) en un proyecto dbt en las pull request (cuando solicitamos unir nuestra rama a la principal) usando webhooks de Github, Azure DevOps o Gitlab. Cuando una PR está aprobada se envía un webhook a dbt cloud que encola una nueva ejecución del job correspondiente. La ejecución del job se realiza sobre un esquema temporal que se crea y autodestruye. La PR no realizará el merge hasta que finalice la ejecución del job. Vamos a hacer una prueba (puedes consultar aquí toda la documentación):

1. Creamos un nuevo un job cuyo trigger sea Continious Integration (CI) y activamos la opción de Run on Pull Requests?

Si no nos aparece el check Run on Pull Requests? debemos reconfigurar la conexión a Github y usar la conexión nativa desde dbt. Es necesario seguir los siguientes pasos:

1. Conecta tu cuenta de dbt con Github y otorga permisos de lectura/escritura sobre el repositorio que vayas a utilizar. Desde Profile Settings > Linked Accounts seleccionamos Github y en el botón Configure integration with Github. Más info en esta nota de dbt.
2. Desconecta en el proyecto la configuración actual de Github por SSH desde Account Settings > Projects (analytics) > Github connection pulsamos en editar y abajo a la izquierda aparece el botón Disconnect.
3. SI volvemos a la pantalla de configuración del proyecto y pulsamos de nuevo en Repository Details podemos seleccionar de nuevo el proveedor de repositorios. Esta vez en lugar de clonar, vamos a conectarnos directamente a Github y seleccionar un repositorio:

2. Actualizamos nuestro proyecto con la última foto de la repo si es necesario (pull) y creamos una nueva rama (create new branch):

3. Hacemos cualquier cambio en alguno de los models. Hacemos commit del cambio y finalmente una PR:

4. Desde Github aprobamos la PR.

Si nos dirigimos a dbt podemos ver como se ha encolado una nueva ejecución del job:

Si revisamos los pasos del job podemos ver que se ha disparado desde una PR y que se crea un esquema temporal en nuestro dataset de BigQuery con nombre dbt_cloud_pr_219037_6. Este esquema se autodestruye cuando finaliza el job.

Google Data Studio

Google Data Studio es una herramienta gratuita basada en Looker Studio de visualización de datos y creación de informes que permite a los usuarios conectarse a múltiples fuentes de datos, como Google Analytics, Google Ads, Google Sheets, bases de datos y más, para crear informes personalizados y visualizaciones interactivas.

La plataforma ofrece una interfaz gráfica fácil de usar que permite a los usuarios diseñar y personalizar informes con diferentes tipos de gráficos, tablas, indicadores clave de rendimiento (KPIs) y otras visualizaciones, y compartirlos con otros usuarios de forma segura y fácilmente.

Básicamente tenemos dos tipos de elementos: informes y fuentes de datos. Los primeros son los cuadros de mando con las visualizaciones y los segundos son los conectores con las tablas de los sistemas fuente. El primer paso para generar un dashboard es configurar las fuentes de datos.

Fuente de datos (Data Source)

Una fuente de datos es una conexión a una tabla de nuestro sistema origen (BigQuery en el caso del ejemplo). Por defecto infiere la estructura de datos de la base de datos, pero podemos modificarla o crear nuevos elementos como campos y parámetros:

• Campos: podemos crear nuevos KPI o campos derivados a partir de otros de la tabla utilizando el catálogo de funciones disponible en la documentación de Google Data Studio.
• Parámetros: los parámetros permiten interactuar de forma dinámica al usuario con los datos del informe. Mediante entrada de datos, podemos por ejemplo realizar estimaciones de algún cálculo en base al valor que haya introducido el usuario.

Seguimos los siguientes pasos para configurar una fuente de datos:

1. Pulsamos en Crear y seleccionamos fuente de datos. En el nuevo panel buscamos por el conector de BigQuery:

2. Autorizamos el acceso de Google Data Studio a nuestro BigQuery:

3. Seleccionamos la tabla que queremos usar como origen de datos:

4. El último paso para la fuente de datos es revisar y confirmar la estructura de datos que ha inferido del sistema origen. En este punto también podemos realizar tareas de transformación y crear nuevos campos y parámetros. Por ejemplo, aunque podemos hacerlo al vuelo mientras diseñamos un informe, desde este punto podríamos crear el campo month_pickup que segmente los datos por mes con la fórmula month(pickup_datetime):

O en tiempo de diseño, desde el blade de datos:

Informes en Google Data Studio (reports)

Crear un informe en Google Data Studio es muy sencillo, siguiendo la tendencia actual de herramientas como Power BI de Microsoft, Qlik Sense o MicroStrategy Visual Insights. Tenemos a nuestra disposición un lienzo en blanco sobre el que vamos a construir visualizaciones en base al set de datos que hemos configurado: seleccionamos el control, el gráfico sobre el que configuramos sus dimensiones y métricas y ¡listo! Veamos qué son estos dos tipos de elementos:

• Controles: objetos que nos permiten interactuar con los datos de las visualizaciones, por ejemplo, filtros de selección, cajas para introducir texto, listas desplegables con todos los valores de una dimensión, etc.
• Gráficos: o visualizaciones, son todo tipo de gráficos estadísticos que podemos utilizar para analizar y presentar la información: gráficos de tartas, barras, línea, burbujas, etc. En función del gráfico seleccionado debemos elegir una o varias dimensiones y una o varias métricas.

Tras unos minutos de esfuerzo podemos disfrutar de nuestro primer report en Google DAta Studio:

La entrada Data Engineering Zoomcamp – Semana 4 se publicó primero en Marcos Torregrosa.

import pandas as pd
data = [
        {'name' : 'John Connor', 'world' : 'Earth', 'age' : 16, 'survive' : 1},
        {'name' : 'Max Rockatansky', 'world' : 'Earth', 'age' : 31, 'survive' : 1},
        {'name' : 'Ender', 'world' : 'Albion' , 'age' : 6, 'survive' : 1},
        {'name' : 'Anakin Skywalker', 'world' : 'Tatooine', 'age' : 8, 'survive' : 0},
        {'name' : 'Ellen Ripley', 'world' : 'Earth', 'age' : 37, 'survive' : 0},
        {'name' : 'Willow Ufgood', 'world' : 'Earth', 'age' : 25, 'survive' : 1}
]
df = pd.DataFrame(data)
df.dtypes

En primer lugar, para analizar el consumo de memoria de cada columna de un dataframe usamos el método de Pandas memory_usage. Vamos a usar dos parámetros con este método:

• deep = True : estima el uso de memoria más preciso a nivel de fila y tipo de dato.
• index = False : si no indicamos nada, por defecto nos indica el consumo de RAM del índice del dataframe además de cada columna.

Como el resultado lo representa en bytes, podemos dividir dos veces por 1024 para tener el dato en MB.

df.memory_usage(deep = True, index = False)

Nos devuelve el peso en bytes del índice y cada una de las columnas.

Index      128
name       414
world      376
age         48
survive     48
dtype: int64

Si comprobamos el tipo de datos inferido del dataset:

df.dtypes

Vemos que no es lo más óptimo. Las columnas de tipo string las ha tipado como OBJECT y las numéricas INT64, veamos qué podemos hacer:

name       object
world      object
age         int64
survive     int64
dtype: object

Filtro de datos categóricos (CATEGORY)

Las columnas de datos categóricos de tipo STRING o DATE podemos convertirlos en CATEGORY. Es importante tener en cuenta la cardinalidad de los datos (cantidad de valores distintos). Vamos a conseguir reducir el consumo de memoria siempre que tengan baja o media cardinalidad, si convertimos a category una columna con una cardinalidad muy alta probablemente necesite más memoria que si no lo hiciéramos. Lo que hace el tipo CATEGORY es crear un diccionario de todos los valores distintos de una columna, sustituyéndolos por punteros al diccionario. Vamos a probar con nuestro ejemplo, primero vamos a observar el consumo de memoria de las columnas name y world según están definidas como tipo OBJECT con df.memory_usage(deep = True, index = False):

name       414
world      376
dtype: int64

Aplicamos la optimización cambiando el tipo a CATEGORY:

df["name"] = df["name"].astype("category")
df["world"] = df["world"].astype("category")

Y volvemos a observar el uso de memoria:

name       592
world      304
dtype: int64

La columna world logra reducir el peso un 19% (de 375 bytes a 304), sin embargo ¿qué ha pasado con name? ¡ha aumentado el consumo de memoria! este comportamiento se debe a que la cardinalidad de la columna es muy alta (un dataset con 5 registros y 5 valores distintos), por lo que al generar el diccionario de la categoría necesita más memoria.

Optimizar el tipado de los datos

En este paso vamos a cambiar el tipo de cada columna para intentar ahorrar costes. Las columnas de texto a STRING y las numéricas que por defecto asigna como INT64, podríamos convertirlas a INT8, INT16 o INT32. Siguiendo el ejemplo del post:

name       object
world      object
age         int64
survive     int64
dtype: object

Vamos a analizar cómo cambia el consumo de memoria al convertir las columnas OBJECT a STRING y las numéricas a INT8 (nos vale un entero de 8 bits porque abarca valores de -128 a 127). Primero observamos el consumo de memoria antes de hacer la conversión:

name       414
world      376
age         48
survive     48
dtype: int64

Y aplicamos las conversiones:

df["name"] = df["name"].astype("string")
df["world"] = df["world"].astype("string")
df["age"] = df["age"].astype("int8")
df["survive"] = df["survive"].astype("int8")

¿Cómo cambia?

name       414
world      376
age          6
survive      6
dtype: int64

Observamos que las columnas numéricas age y survive han reducido su peso 87,5% (de 48 bytes a 8). En las columnas de texto no apreciamos cambios. Lo ideal es asignar el tipado al crear el dataframe, pero depende de lo que estemos usando como origen. Por ejemplo, al cargar un fichero Parquet se carga el esquema definido en el propio fichero. Si se trata de un CSV o XML que cargamos con los métodos de Pandas read_xml y read_csv podemos especificar un tipo común a todas las columnas (dtype="string") o bien pasarle un diccionario con el tipo de cada columna (dtype={"col1" : "string"}) :

# Asignamos el tipo String a todas las columnas
df = pd.DataFrame(data, dtype = "string")

# Especificamos el tipo por columna gracias a un diccionario
dictTypes = {"name":"string", "world":"string", "age":"int8", "survive":"int8"}
df = pd.DataFrame(data, dtype = dictTypes )

Y observamos los tipos con dtypes:

name       string
world      string
age          int8
survive      int8
dtype: object

Reducir número de columnas que cargamos en un dataframe

Es la optimización más sencilla y lógica, la forma más fácil de reducir el consumo de recursos es utilizar sólo los datos que necesitamos. A la hora de cargar un dataframe seleccionamos sólo las columnas necesarias. Para este caso vamos a imaginar que cargamos el dataset desde un CSV con READ_CSV, podemos pasarle en el parámetro USECOLS un listado de las columna que deseamos cargar.

import pandas as pd

fields = ['name','world']
df = pd.read_csv('dataset.csv', usecols = fields)
print(df)

La entrada Python: Optimizar uso de memoria con Pandas se publicó primero en Marcos Torregrosa.

• Repo Bootcamp: https://github.com/DataTalksClub/data-engineering-zoomcamp/tree/main/week_3_data_warehouse
• Repo personal con ejercicios: https://github.com/hegdehog/data-engineering-zoomcamp-2023

Última actualización 06/02/2023

Prerequisitos

Esta semana no hemos dado grandes pasos en la suite tecnológica. Para seguir el temario y realizar los ejercicios continuaremos con la cuenta de Google Cloud Platform y la instancia de BigQuery que creamos a inicio del bootcamp.

Google BigQuery

Dentro de la suite de servicios cloud que ofrece la GCP (Google Cloud Platform) encontramos BigQuery como solución de almacenamiento y análisis de datos. Como todo servicio cloud, destaca por la escalabilidad, alta disponibilidad y pago por uso. Entre sus características, destacan:

1. Serverless: al ser un PaaS el usuario no debe preocuparse por el mantenimiento software/hardware, sólo de su configración y explotación.
2. Escalabilidad: BigQuery es altamente escalable, lo que significa que puede manejar un aumento en el volumen de datos sin interrupciones.
3. Almacenamiento en la nube: BigQuery permite almacenar y procesar grandes volúmenes de datos en la nube de Google, lo que significa que los usuarios no tienen que preocuparse por la infraestructura de hardware.
4. Análisis en tiempo real: BigQuery permite realizar consultas en tiempo real sobre grandes cantidades de datos.
5. Integración con herramientas de Google: BigQuery se integra con otras herramientas de Google, como Google Data Studio, Google Cloud Storage y Google Cloud Dataproc.
6. Consultas SQL: BigQuery permite realizar consultas SQL en tiempo real sobre grandes cantidades de datos. El SQL es compatible con la semántica de transacción de la base de datos (ACID)
7. Pago por uso (pay-as-you-go): BigQuery es un servicio de pago por uso, lo que significa que los usuarios solo pagan por el almacenamiento y el análisis de datos que utilizan.
8. Seguridad y cumplimiento normativo: BigQuery ofrece una amplia gama de características de seguridad y cumplimiento normativo para ayudar a proteger los datos de los usuarios.
9. Compatibilidad con diferentes formatos de datos: BigQuery es compatible con diferentes formatos de datos, como CSV, JSON y Avro, y permite a los usuarios importar y exportar datos de y hacia otras plataformas.

Precios de BigQuery

La facturación de BigQuery consta de dos partes: el precio del análisis y el del almacenamiento. Más información en el pricing de GCP BigQuery.

• El precio de análisis: coste de procesar consultas (SQL, funciones definidas por el usuario UDF, scripts DML y DDL). Sin entrar en mucho detalle, hay dos modelos de precios disponibles para el análisis: bajo demanda (5$ por TB) o tarifa plana (el precio varía en función del tipo ranura, mensual o anual)
• El precio de almacenamiento coste de almacenar físicamente los datos. El almacenamiento es más barato y encontramos dos tipos (similar a los tiers del los Storage Account de Azure):
  • Activo: 0,020$/GB (los primeros 10 GB/MES son gratis). Automáticamente para las tablas o particiones con cambios en los últimos 90 días.
  • Largo plazo: 50% de descuento: 0,010$/GB para aquellas tablas y particiones que no se hayan tocado en los últimos 90 días.

External tables

Al igual que en otras soluciones cloud como en Azure Synapse, desde Google BigQuery podemos crear tablas externas (external tables) para consultar conjuntos de datos como si fueran una tabla SQL que no estén almacenados en BigQuery (archivos en un Cloud Storage, otra base de datos fuera de GCP, etc). Este tipo de objeto almacena en BQ únicamente el esquema, que infiere del fichero a la hora de crear el objeto, pero no los datos. Las tablas externas tienen las mismas características que una tabla estándar en BigQuery, con sus propiedades, administración de acceso, metadatos, etc. La única diferencia es que son una cáscara, los datos están en otra ubicación.

Por ejemplo, en lugar de ingestar un CSV en una tabla de la BBDD BigQuery, vamos a crear una external table para acceder directamente a los datos sin persistir:

CREATE OR REPLACE EXTERNAL TABLE 'my_database.yellow_tripdata_external_table'
OPTIONS (
  format = 'CSV',
  uris = ['gs://data/trip data/yellow_tripdata_2020-01.csv']
);

Optimización de Google BigQuery

A diferencia de una base de datos relacional, BigQuery no soporta índices para agilizar las consultas SQL. En su lugar, ofrece dos alternativas: particionado (partitioning) y agrupación (clustering). Estas opciones no son recomendables cuando nuestro volumen de datos es < 1GB.

Particionado (Partitioning)

Una tabla particionada es una tabla dividida en segmentos aka particiones basados en los valores de una columna. Dividir una tabla en porciones agiliza enormemente las consultas ya que los datos que necesita recuperar son mucho más reducidos que si tuviera que leer toda la tabla. BigQuery ofrece tres tipos de particiones:

• Partición por rangos de números enteros: se crean particiones en base al valor numérico de una columna de tipo INTEGER. Por ejemplo, por el código de país.
• Partición de columnas por unidad de tiempo: la partición más habitual, se particiona la tabla por una columna de tipo DATE, TIMESTAMP o DATETIME.
• Partición por tiempo de ingesta: BigQuery asigna de forma automática filas a las particiones en función del momento en que BigQuery transfiere los datos. Puedes elegir el nivel de detalle por hora, día, mes o año para las particiones. Tiene un límite de 4k particiones. Se añade la columna _PARTITIONTIME y en cada tupla se asigna el valor del momento en el que se ha almacenado el dato.

En el siguiente ejemplo creamos una nueva tabla a partir de una consulta SELECT y añadimos el particionado por la columna tpep_pickup_datetime

CREATE OR REPLACE TABLE my_database.yellow_tripdata_partitoned
PARTITION BY
  DATE(tpep_pickup_datetime) AS
SELECT * FROM my_database.yellow_tripdata;

Cuando lanzas una query en BigQuery sobre una tabla particionada no es necesario que lo especifiques, internamente lo identifica y realiza la consulta sobre la partición que cumpla el criterio del predicado (o filtro). Diferencia importante respecto a SQL Server u Oracle, donde sí se debe especificar la partición en la sentencia SELECT.

SELECT DISTINCT(VendorID)
FROM my_database.yellow_tripdata_non_partitoned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2021-01-01' AND '2021-02-28';

Agrupación (Clustering)

El clustering reordena los datos de la tabla en base a una o más columnas (hasta 4). Características de la agrupación de columnas en BigQuery:

• El orden de las columnas agrupadas es relevante para determinar la prioridad de las columnas
• Mejora el rendimiento en consultas que utilicen predicados o funciones de agregación
• Funciona mucho mejor con columnas con una alta cardinalidad (email, categorías, nombres)
• Se pueden hacer agrupaciones en columnas de tipo:
  • DATE
  • BOOL
  • GEOGRAPHY
  • INT64
  • NUMERIC
  • BIGNUMERIC
  • STRING
  • TIMESTAMP
  • DATETIME
• Límite de como máximo 4 columnas agrupadas (clusters) por tabla.

Podemos crear clusters al mismo tiempo que las particiones. Tomando como ejemplo la query anterior, vamos a añadir una columna grupada o cluster por el campo VendorID:

CREATE OR REPLACE TABLE my_database.yellow_tripdata_partitoned_clustered
PARTITION BY DATE(tpep_pickup_datetime)
CLUSTER BY VendorIDKey AS
SELECT * FROM my_database.external_yellow_tripdata;

BigQuery Partitioning vs Clustering

Podemos utilizar ambas opciones de optimización juntas y aprovechar las sinergias que generan, o en función de la necesidad elegir una estrategia u otra. Normalmente usaremos particionado cuando nuestras consultas filtran o agregan datos sobre un único campo con baja cardinalidad (año, mes, ciudad, departamento…), mientras que si es por varios podemos particionar por el que tenga menor cardinalidad (fecha, por ejemplo) y aplicar clustering por el resto (hasta un máximo de 4 columnas).

El propio motor de BigQuery es el encargado de actualizar automáticamente los clusters cuando se insertan nuevos datos en la tabla de forma transparente para el usuario. En el caso de tablas particionadas, los clusters se actualizan de forma independiente en cada partición.

Buenas prácticas SQL para optimizar consultas

La mayoría no sólo aplican a Google BigQuery, se trata de recomendaciones para queries ejecutadas sobre cualquier motor de BBDD:

• Evitar usar SELECT *, lo ideal es recuperar sólo las columnas que necesitemos o vayamos a utilizar.
• Evalúa el coste de ejecutar la query antes de lanzarla. Esto es especialmente útil en entornos cloud donde la facturación seleccionada sea por ejecución (se paga por cada ejecución), que normalmente es más caro que si se selecciona una capacidad o paquete.
• Aplicar la optimización por partitioning y/o clustering
• En los casos de real-time debemos prestar atención y ser cuidadosos con los INSERT de datos (insertAll)
• Crear vistas materializadas como pasos intermedios cuando la query debe manejar un gran volumen de datos. Debemos tener en cuenta que BigQuery también cachea los resultados de las columnas.
• Aplicar filtros por columnas de partición o agrupación (clusters)
• Desnormalizar los datos bajando las formas normales a lo mínimo, en otras palabras, destruir la integridad referencial manteniendo todos los datos en una única tabla para evitar joins entre varias. Se recomienda usar campos anidados o repetidos (nested fields) con STRUCT o ARRAY. Aunque tiene ciertas desventajas (mayor almacenamiento por repetir datos y se pierde la integridad de los datos), es la forma más óptima de explotar grandes volúmenes de datos.
• Intenta utilizar funciones de agregación aproximada HyperLogLog++ o HLL++. Necesita menos memoria que las funciones de agregación exacta, como COUNT(DISTINCT), pero generan incertidumbre estadística. Son muy útiles para grandes volúmenes de datos en los que el uso de la memoria lineal es poco práctico teniendo en cuenta que el dato que nos devuelve es una aproximación estadística, no el valor exacto.
• Evitar usar funciones propias SQL o JavaScript UDF
• Cuando cruces varias tablas organiza el JOIN ubicando la más voluminosa primero. Será la que BigQuery use en primer lugar para distribuirla por los nodos y las siguientes tablas las irá repartiendo por cada uno. Además, intenta reducir el tamaño de las subqueries o vistas materializadas antes de hacer los cruces.

Arquitectura Google BigQuery

La arquitectura de BigQuery desvincula el almacenamiento de la computación (motor de análisis), lo que permite escalar cada recurso de forma independiente. Esta flexibilidad favorece un control de costes mucho más granularizado ¿Qué piezas encontramos dentro de BigQuery? Dremel, Colossus, Jupiter y Borg:

Borg: orquestador de contenedores

Orquestador de contenedores propio de Google que se encarga de proveer del hardware necesario para que operen los slots y mixers del motor Dremel.

Jupyter: red

Como la estructura de BigQuery es desacoplada (separa físicamente el almacenamiento del motor de cómputo) necesita un artefacto que conecte ambas entidades: Jupyter. Ofrece suficiente ancho de banda como para permitir la comunicación entre 100K máquinas a una velocidad ultra rápida de 10Gbs/s.

Dremel: motor de ejecución

Se trata del motor de consultas de alta velocidad de BigQuery que usa Google en su propio buscador. Orquesta las consultas segmentándolas en pequeñas porciones que se distribuyen por nodos y cuando finalizan se agrupan para devolver el resultado; la definición de procesamiento distribuido. Dremel convierte una consulta SQL en un árbol de ejecución donde encontramos slots y mixers, todos ejecutados sobre Borg (ver más adelante). El propio motor asigna de forma dinámica slots a las consultas que van entrando:

• Slots: serían las hojas del árbol y se encargan de la parte más pesada: leer datos en Colossus y realizar operaciones de cómputo.
• Mixers: las ramas. Se encargan de las operaciones de agregación

Colossus: almacenamiento distribuido

Sistema de almacenamiento distribuido de última generación de Google. Gestiona las réplicas, recuperación (cuando los discos fallan) y administración distribuida (lo que mitiga el impacto en caso de caída). Colossus utiliza el formato columnar y de compresión ColumnIO capaz de manejar fácilmente petabytes de datos.

Almacenamiento

Column-oriented vs record-oriented

Tradicionalmente el almacenamiento de datos tabular está orientado a filas (record-oriented o row-oriented). De esta forma, los datos se almacenan y consultan fila por fila y después se accede a cada una de las columnas. BigQuery en su lugar, utiliza almacenamiento orientado a columnas (column-oriented). Este tipo de almacenamiento es ideal para grandes volúmenes de datos ya que permite reducir la consulta sólo a las columnas deseadas, sin necesidad de recuperar todos los datos de una fila para luego quedarnos con lo que nos interesa.

El motor de ejecución Dremel guarda los datos en Colossus en formato columnar, es decir, divide un registro o fila por los valores de sus columnas y almacena cada valor en un volumen distinto. En las base de datos tradiciones normalmente se almacena el registro completo en el mismo volumen. Cuando se lanza una consulta sobre BQ, Dremel crea un árbol de ejecución distribuyendo la query en porciones a través de los mixers que a su vez se distribuyen en paquetes más pequeños sobre la malla de slots que acceden a la capa de almacenamiento Colossus (la siguiente sección incluye un diagrama al final donde se visualiza esta explicación).

Campos anidados y repetidos (repeated & nested fields)

Una de las buenas prácticas para modelar una base de datos analítica es desnormalizar con el fin de reducir los cruces entre tablas (hechos + dimensiones). Se almacenan todos los datos en una única tabla, de forma que la información (como la ciudad) se repite y almacena en la misma tabla, no es necesario cruzar con una tabla de dimensión para recuperar los valores. BigQuery soporta el uso de estructuras repetidas o anidadas mediante las operadores STRUCT y ARRAY. Esto nos permite representar la información de forma natural además de agilizar las consultas. Podemos combinar ambas estrategias para el almacenamiento de datos

• STRUCT: subconjunto de datos (columna) formado por varios campos ordenados con nombre y tipo. Se usa notación de puntos para consultar los campos anidados de una columna, por ejemplo: cliente.nombre donde nombre es el campo y cliente la columna.

• ARRAY: lista ordenada con 0 o más elementos del mismo tipo. Por ejemplo, producto es un ARRAY de STRUCTs. Podemos usar UNNEST() para aplanar los elementos de la lista por cada fila o usar OFFSET/ORDINAL para acceder a cada elemento de forma individual (más adelante hay ejemplos de estas consultas).

El motor Dremel debe su velocidad, entre otras cosas, al almacenamiento columnar de los campos anidados (nested fields). Todos los valores de un campo anidado (por ejemplo A.B.C) se almacenan de forma contigua, por lo que permite que se recupere sólamente A.B.C sin necesidad de leer A.E, A.B.D, etc.

En la siguiente query vamos a crear la tabla my_database.tabla_array_struct con cuatro elementos: dos columnas estándar de tipo string y date, y dos campos anidados: cliente de tipo ARRAY y producto que sería un ARRAY o colección de STRUCT.

CREATE TABLE `my_dataset.tabla_array_struct`
(
  ID_PEDIDO int64,
  FECHA date,
  CLIENTE  ARRAY<string>,  
  PRODUCTOS ARRAY<STRUCT<SKU int64, CANTIDAD int64, PRECIO numeric>>,
);

Poblamos la tabla de la siguiente forma:

INSERT INTO `my_dataset.tabla_array_struct` (ID_PEDIDO, FECHA, CLIENTE, PRODUCTOS)
VALUES (
    1,
    CAST('2021-01-01' AS date), 
    ['231', 'Marcos Torregrosa', 'Madrid, Spain'],
    [
      STRUCT(6633 as SKU, 2 as CANTIDAD, cast(3.99 as numeric) as PRECIO),
      STRUCT(1667, 1, 1.99),
      STRUCT(8871, 1, 16.99)
    ] 
    );

Si observamos el esquema vemos que BQ trata estas columnas como RECORD y modo REPEATED.

Echando un ojo a los datos:

Con OFFSET u ORDINAL podemos acceder directamente por el índice a cualquier elemento de una columna de tipo ARRAY:

SELECT cliente, cliente[OFFSET(1)] as offset_1, cliente[ORDINAL(1)] as ordinal_1
FROM my_dataset.tabla_array_struct

Con el operador UNNEST podemos aplanar un ARRAY mostrar una tabla con una fila para cada elemento en ese array. 

select * from my_dataset.tabla_array_struct, UNNEST(cliente) as clientes
where cliente[OFFSET(1)] like "Marcos%"

BigQuery y Machine Learning

Desde BigQuery podemos ejecutar modelos de machine learning de forma sencilla y ágil usando simplemente SQL estándar sin necesidad de plataformas específicas, movimiento de datos o conocimientos de programación (python, scala, etc). Los algoritmos de ML disponibles nativamente dentro de BigQuery pueden consultarse en la documentación oficial.

Por ejemplo, para crear un modelo de regresión lineal con el objetivo de predecir la propina (tip_amount) de una carrera en taxi usamos la sentencia CREATE MODEL (recomendable ver la documentación con todas las opciones):

CREATE OR REPLACE MODEL `my_database.nytaxi.tip_model`
OPTIONS (
  model_type='linear_reg',
  input_label_cols=['tip_amount'],
  DATA_SPLIT_METHOD='AUTO_SPLIT'
) AS
SELECT
  *
FROM
  `my_database.nytaxi.yellow_tripdata`
WHERE
  tip_amount IS NOT NULL;

• CREATE OR REPLACE MODEL es la sentencia para crear nuestro modelo
• Dentro de las OPTIONS() vamos a indicar los parámetros y configuración del modelo:
  • MODELE_TYPE='linear_reg' en nuestro ejemplo vamos a crear un modelo de regresión lineal. Podríamos usar cualquiera de los disponibles en BQ (como KMEANS para crear clústers de datos o RANDOM_FOREST_CLASSIFIER para crear un modelo de clasificación)
  • INPUT_LABEL_COLS=['tip_amount'] Array de columnas separadas por comas que vamos a utilizar para entrenar y usar el modelo.
  • DATA_SPLIT_METHOD='AUTO_SPLIT' especificamos que queremos dividir automáticamente el conjunto de datos en dos partes, una de entrenamiento y otra de comprobación (training/test).
• El SELECT especifica la fuente de datos, así como el predicado si lo hubiera (filtro).

El modelo que hemos creado se almacena dentro del conjunto de datos en BigQuery como un nuevo objeto:

BQ nos ofrece una serie de sentencias para analizar y explotar el modelo. Más información en la documentación oficial.

• ML.FEATURE_INFO: muestra estadísticas de cada columna del conjunto de datos (valores mínimo y máximo, medias, etc). Similar a ejecutar el comando describe() de en Pandas (python).
• ML.EVALUATE: muestra las métricas de un modelo, ideal para comprobar con un nuevo conjunto de datos cómo respondería el modelo. las métricas que ofrece son las mismas que podemos consultar viendo el detalle del model creado desde el GUI de GCP.
• ML.PREDICT: nos permite ejecutar el modelo sobre un conjunto de datos y generar las predicciones para las que se haya configurado.
• ML.EXPLAIN_PREDICT: añade información a la sentencia anterior sobre cuáles de las columnas o features son las que más colaboran para el cálculo de la predicción.

La entrada Data Engineering Zoomcamp – Semana 3 se publicó primero en Marcos Torregrosa.

Que una organización se defina data-driven está estrechamente relacionado con el nivel de data literacy implementado en toda su estructura. Si las decisiones empresariales deben estar basadas en los datos, se hace imprescindible desarrollar los niveles de comprensión y análisis de la información, esto es, de data literacy. En los informes de Gartner se puede observar como uno de los principales bloqueos identificados en la obtención de resultados de los datos es una pobre alfabetización de datos.

Compañero en este camino surge la necesidad de desarrollar un lenguaje común entre todos los individuos. Según el informe Getting Started With Data Literacy publicado por Gartner en 2018, este lenguaje común favorece:

• Los líderes de negocio internalizan la importancia de crear una cultura de compartir datos y recursos
• Comprender el valor fundamental de mantener la calidad de los datos
• Crear soluciones basadas en datos gracias a una alta colaboración y creatividad
• Todos los empleados hablan el mismo lenguaje lo que permite abrir debates en búsqueda de nuevas métricas y análisis

Cómo medir la data literacy

Según Gartner, para medir el nivel de data literacy en una organización podemos responder a las siguientes preguntas:

• ¿Cuántas personas de la organización pueden interpretar gráficos o KPI?
• ¿Cuántos miembros de dirección serían capaces de construir un caso de negocio basado en cifras concretas, precisas y relevantes?
• ¿Cuántos miembros de dirección pueden explicar los resultados de los procesos de sus áreas?
• ¿Cuántos data scientists pueden explicar los resultados de los algoritmos de ML que ejecutan?
• ¿Cuántos clientes pueden realmente apreciar e internacionalizar la esencia de los datos que se comparten con ellos?

Si las simplificamos y traducimos a las necesidades del día a día a nivel de individuo, podríamos encontrar:

• Cuando te enfrentas a los datos ¿cuál es tu sensación? ¿aburrimiento, miedo o interés?
• ¿Utilizas los datos para realizar o mejorar tu trabajo?
• ¿Crees que la comprensión de los datos es tan importante como leer o escribir?
• ¿Tomas decisiones basadas en tu instinto?
• ¿Son accesibles foros donde se fomente la discusión de soluciones en torno a los datos?
• ¿Has consultado o eres consciente de los datos de tu área?

Las respuestas a estas preguntas nos ofrecen un marcador para identificar el nivel de data literacy de la organización. Según ese nivel, podremos diseñar palancas para mejorar o incrementar la sensibilidad de los datos en todas los estratos.

Desarrollar habilidades de data literacy

Un buen nivel de data literacy en una organización se traduce una buena evangelización de los datos por todas las capas (desde dirección hasta los empleados del último nivel). Desarrollar las habilidades de comprensión, interpretación y análisis crítico de los datos en cada una de las personas allana el camino mejorando y facilitando la toma de decisiones en la dirección correcta (o al menos, basada en datos empíricos y no por intuición). Algunas ideas para mejorar la alfabetización de los datos:

• Educar a todos los individuos a comprender y diseñar sus propias visualizaciones de datos, cuadros de mando, etc. Con la irrupción desde hace unos años de herramientas self-service de visualización de datos que no precisan de conocimientos técnicos se ha democratizado el acceso (Power BI, Qlik Sense, MicroStrategy Visual Insight)
• Organizar talleres impartidos por los expertos en data (IT o dominios) al resto de áreas.
• Mejorar y desarrollar documentación que orbita a entender los datos que utiliza o expone cada área de la compañía
• Mantener, distribuir y apostar por un gobierno de datos que favorezca el descubrimiento de metadatos para todos los individuos (linaje de datos, explotación, definiciones, seguridad, etc). Esta herramienta debe estar accesible para toda la compañía, de forma que cualquiera sea capaz de resolver cualquier duda sobre los datos que necesita.
• Incentivar y facilitar las conversaciones o debates en torno a los datos para encaminar la toma de decisiones basadas en éstos y no en la intuición. Para ello es imprescindible

El equipo de Qlik ha desarrollado la plataforma Data Literacy Project para ayudar a organizaciones e individuos a entender y desarrollar las habilidades necesarias para extender la cultura del dato. Incluso ofrecen un plan de formación y certificación. A modo de pequeño juego, podemos hacer un pequeño examen o encuesta para comprobar nuestro nivel de data literacy.

La entrada Data literacy o alfabetización de los datos se publicó primero en Marcos Torregrosa.

Data-Driven Design

En Data-Driven Design (DDD) se utiliza los datos para tomar decisiones y guiar el diseño y desarrollo de un sistema. Los datos se ven como el principal insumo para la construcción de ese sistema. Se utilizan para la definición de funcionalidades y características, es decir, para la toma de decisiones técnicas y arquitectónicas. El objetivo es construir sistemas que sean altamente escalables, flexibles y adaptables, y que sean capaces de responder a los cambios en los requisitos y las necesidades del negocio apoyándose en una fuerte capacidad de recopilación, almacenamiento, procesamiento y análisis de datos.

Domain-Driven Design

En Domain-driven Design (DDD) busca resolver un problema de negocio (dominio) convirtiendo el software en un reflejo del sistema en el mundo real. La toma de decisiones se basa en el conocimiento y comprensión del dominio. Éste se puede modularizar en subdominios que respondan a necesidades más específicas e independientes, mejorando la adaptabilidad, escalabilidad y resiliencia del sistema. El objetivo es crear un entorno de colaboración entre los expertos de negocio (saben el qué) y el equipo técnico (saben el cómo) para un desarrollo que sea coherente y pueda adaptarse a los cambios constantes en los requisitos.

Conceptos de Domain-Driven Design (DDD)

Los siguientes términos definen los elementos clave que componen un enfoque DDD:

• Core Domain y Subdomains se refieren al ámbito del problema o necesidad del mundo real que se busca resolver. Los dominios pueden dividirse en subdominios para centrarse en dar soluciones más acotadas modularizando el dominio en porciones más pequeñas independientes.
• Domain model representación abstracta del problema del mundo real en entidades, relaciones y reglas de negocio.
• Bounded Context sería una frontera clara y explícita alrededor de una parte del dominio que puede ser comprendida y utilizada de manera autónoma. Esta frontera ayuda a evitar ambigüedades, malentendidos y acoplamientos. Permite que los equipos de desarrollo se enfoquen en una parte específica del dominio.
• Context Mapping es un proceso en el que los equipos de desarrollo trabajan con los expertos del negocio para mapear los conceptos y procesos de negocios y establecer las fronteras de los diferentes Bounded Contexts. Un mismo concepto no tiene por qué tener el mismo significado en dos dominios, por lo que el contexto identifica el significado de ese concepto dentro de un dominio. Este proceso ayuda a identificar las relaciones entre los diferentes Bounded Contexts.
• Ubiquitous Language es un lenguaje común y coherente utilizado por todas las partes interesadas, incluyendo los expertos del negocio y los desarrolladores, para describir los conceptos y procesos de negocios. Este lenguaje ayuda a reducir malentendidos y a asegurarse de que todas las partes comprendan de manera consistente.

Definición de un modelo

Uno de los elementos clave de DDD es el modelo de dominio formado por las entidades, relaciones y reglas de negocio de un problema o necesidad del mundo real. Para la definición del modelo es importante la interacción entre los expertos de negocio y el equipo técnico siguiendo la regla de ubiquitous language comentaba en el punto anterior. Algunos de los elementos clave del modelo de dominio incluyen:

1. Entities: Las entidades son objetos únicos y distintos que representan conceptos significativos en el dominio, como clientes, productos o pedidos. Cada entidad tiene atributos y relaciones con otras entidades.
2. Values: Los valores son objetos que representan conceptos sin identidad en el dominio, como una dirección o una fecha. Los valores son inmutables y se usan para representar información que no cambia.
3. Services: Los servicios son componentes que representan procesos o tareas que no son parte de una entidad o un valor en particular, pero que son importantes en el dominio. Los servicios se utilizan para implementar lógica que es demasiado compleja o no pertenece a una entidad o un valor en particular.
4. Aggregates: Los agregados son colecciones de entidades y valores que representan una unidad coherente en el dominio. Cada agregado tiene una entidad de raíz que es la entidad principal.

Caso de uso

Veámoslo con un ejemplo, el diseño de un sistema de gestión de viajes para una aerolínea.

El equipo de desarrollo comenzaría identificando los distintos dominios o áreas de negocio dentro del sistema, como reservas, facturación, asignación de asientos, etc. Cada uno de estos dominios tendrá sus propias reglas y requisitos específicos. A continuación, se utilizaría el Ubiquitous Language para asegurarse de que todas las partes involucradas (expertos de negocio y equipo técnico) comprenden claramente los términos y conceptos específicos de cada dominio o subdominio. Por ejemplo, en el dominio de reservas, se podrían definir entidades como «pasajero», «reserva», «vuelo», etc. Una vez que se han definido las entidades, se puede comenzar a modelar el comportamiento y las interacciones entre ellas. Por ejemplo, se puede especificar cómo se realiza una reserva, cómo se asigna un asiento, cómo se factura un vuelo, etc. A partir de aquí, se podría comenzar a implementar el sistema, utilizando patrones de diseño y técnicas específicas de DDD para asegurarse de que el sistema se ajuste claramente a los requisitos y reglas de cada dominio.

Data Mesh y Domain-Driven Design (DDD)

Data Mesh es un enfoque para la gestión de datos que se basa en el Domain-Driven Design (DDD). La idea detrás de Data Mesh es que los datos se traten como un producto de software independiente, con sus propios equipos de desarrollo, su propia arquitectura y su propio ciclo de vida.

Un ejemplo para ilustrar esto podría ser una inmobiliaria. En este caso, siguiendo el enfoque DDD se identificarían los distintos dominios de negocio, como propiedades, compradores, vendedores, etc. Cada uno de estos dominios tendrá sus propios datos específicos, como descripciones de propiedades, información de contacto de compradores y vendedores, etc. En lugar de gestionar todos los datos en una arquitectura monolítica como un data warehouse o data lake, cada Dominio se encarga de crear y mantener sus propios conjuntos de datos conocidos como Data as a Product. Cada dominio estaría formado por los expertos de negocio y personal técnico. De esta manera, los equipos pueden centrarse en el desarrollo de los datos para su dominio específico, y el sistema en su conjunto se vuelve más escalable y fácil de mantener.

En resumen, en este ejemplo, Data Mesh utiliza DDD para dividir los datos en pequeñas piezas autónomas, permitiendo que los equipos se centren en el desarrollo de los datos para sus respectivos dominios de negocio.

La entrada Domain Driven Design (DDD) en arquitectura de datos se publicó primero en Marcos Torregrosa.

¿Qué es un Data Swamp?

Un Data Swamp es un sistema de almacenamiento de datos ineficiente y sin estructura que se encuentra sobrecargado con grandes cantidades de información sin clasificar y sin gestionar adecuadamente. La falta de un plan o procesos efectivos de gestión de datos puede resultar en un acúmulo de información sin valor que es difícil de recuperar y analizar.

Data Lake vs Data Swamp

Cómo surgen los Data Swamps

Los Data Swamps surgen cuando una organización recopila y almacena grandes cantidades de datos sin un plan o procesos efectivos de gestión y clasificación. La falta de estructura y organización puede hacer que los datos se acumulen de manera descontrolada y se conviertan en un problema, ya que no sólo no responden a su cometido: dar respuestas, si no que además, dificulta su explotación y comprensión.

Las organizaciones a menudo se enfocan en recopilar y almacenar la mayor cantidad de datos posible sin preocuparse por su calidad y valor (¡hay que historificar todo! ¡debemos ingestar cualquier dato que se genere!). Esto puede resultar en que se guarden de datos duplicados, incompletos o sin valor que contribuyen al crecimiento del Data Swamp y que además de que nunca vayan a ser utilizados, dificultan encontrar y trabajar con los datos que realmente sí son valiosos.

Cómo evitar los Data Swamps

Para evitar el surgimiento de un Data Swamp, es importante implementar una estructura y procesos efectivos de gestión de datos. Algunos de los pasos que puede tomar una organización para evitarlos incluyen:

1. Implementar una estrategia de gestión de datos: Definir una estrategia clara para la recopilación, almacenamiento y análisis de datos, así como los objetivos y metas a alcanzar ¿es necesario guardar todos los datos de un origen? ¿debemos historificar toda la información? Priorizar qué datos son los realmente valiosos y por tanto, deben almacenarse. Es cierto que la respuesta a esta pregunta es el ahora y no sabemos si en el futuro necesitaremos datos que a día de hoy no se explotan. El ejercicio por tanto sería analizar y priorizar los datos que estamos seguros necesitamos o podremos necesitar en un futuro cercano.
2. Catálogo de metadatos: los metadatos es información sobre los propios datos. Es un mecanismo para clasificar e identificar la información, clave para mantener un Data Lake. Si no sabemos qué tenemos, qué significa, de dónde viene o para que se usa, lo que hemos construido es un data swamp.
3. Fomentar una cultura de datos: Asegurarse de que los empleados comprendan la importancia de los datos y cómo pueden ser utilizados para mejorar la toma de decisiones. En nuestra organización se cuida la cultura del dato mediante un programa denominado Data Fluency cuyo objetivo es evangelizar alrededor del concepto de qué son los datos desde el punto de vista de IT.
4. Adoptar tecnologías efectivas de gestión de datos: Invertir en tecnologías que permitan una gestión eficiente de los datos, incluyendo herramientas de integración, limpieza y análisis de datos. Aunque esto entraría más en procesos ETL o ELT posteriores, en realidad deberían implementarse controles en la ingesta de datos en el data lake.
5. Definir políticas y procedimientos claros: Establecer políticas y procedimientos claros para la gestión de datos, incluyendo la seguridad, privacidad y cumplimiento de regulaciones. Especialmente en compliance y controles RGPD con la conservación de datos.
6. Evaluar regularmente la eficacia de la gestión de datos: Realizar periódicamente auditorías y evaluaciones para determinar si se están cumpliendo los objetivos y si se requiere mejorar la estrategia de gestión de datos.
7. Involucrar a todos los departamentos: Asegurarse de que todos los departamentos de la organización estén involucrados en la gestión de datos y que comprendan su importancia. En línea con el punto nº2.

La entrada Data Swamp: ¿Qué es y cómo evitarlo? se publicó primero en Marcos Torregrosa.

En comparación con los Data Warehouse, los Data Fabrics son más flexibles y escalables. Los Data Warehouse están diseñados para almacenar grandes cantidades de datos estructurados y proporcionar informes y análisis a partir de ellos. Sin embargo, esto puede limitar la capacidad de un Data Warehouse para adaptarse a nuevos tipos de datos o fuentes, y puede requerir una reconfiguración costosa y técnica para hacerlo.

Los Data Lake son soluciones de almacenamiento de datos no estructurados, diseñados para recopilar grandes cantidades de datos raw sin una estructura predefinida y a un coste moderado. Mientras que los Data Lakes pueden ser útiles para almacenar grandes cantidades de datos, pueden ser difíciles de acceder y analizar, y pueden requerir una gran cantidad de tiempo y recursos para asegurarse de que los datos son precisos y coherentes (riesgo de convertirse en data swamps).

Los Data Mesh es una arquitectura o filosofía para la gestión de datos que busca desarrollar una estructura descentralizada y de autogestión, donde cada equipo o departamento es responsable de sus propios datos y puede compartirlos de forma segura con otros equipos. Los datos se tratan como Data as a product (DaaP). A diferencia de los Data Fabrics, los Data Mesh no tienen un punto central de control o gestión, y en su lugar, se basan en la colaboración y el intercambio de datos entre dominios en un sistema federado de propiedad de los datos (DaaPs). Los data fabric intentan resolver los mismos problemas (organizaciones con multitud de orígenes de datos en distintas ubicaciones y tipologías) que el Data Mesh pero desde un punto de vista técnico y no organizativo.

En resumen, el Data Fabric es una arquitectura de datos que combina las fortalezas de los Data Warehouse y Data Lakes, con el mismo objetivo que los Data Mesh pero desde la tecnología y de forma centralizada, que ofrece una solución completa y flexible para la gestión de datos. Supone una vista unificada de los datos, permite conectar diferentes fuentes de datos y ofrece una gestión eficiente y segura de los mismos.

Desde mi punto de vista, Data Fabric se trata de una arquitectura complementaria, no excluyente. Tanto si la arquitectura de datos es centralizada (data lake, data lakehouse) o descentralizada (data mesh), puede ser una pieza acompañante que haga de facilitador para el gobierno y disponibilidad en la ingesta desde los distintos orígenes, así como en la capa de consumo.

Ventajas y desventajas de Data Fabric

Componentes de un Data Fabric

• Virtualización de datos: permite el acceso a los datos desde diferentes fuentes a través de una única capa de abstracción.
• Catalogación y metadatos: proporciona información sobre los datos y su ubicación para facilitar su acceso y gestión.
• Seguridad y privacidad: proporciona medidas para garantizar la seguridad y privacidad de los datos.
• Integración de datos: permite la integración de diferentes fuentes de datos para un acceso más fácil y una gestión más eficiente.
• Procesamiento de datos: permite la ejecución de tareas de procesamiento de datos, incluyendo la limpieza, transformación y análisis de datos.
• Colaboración: permite la colaboración entre los usuarios y equipos para una gestión más eficiente de los datos.
• Monitorización y análisis: proporciona información sobre el uso y rendimiento de los datos para una mejor toma de decisiones y optimización del sistema.

Virtualización de datos

La virtualización de datos es una tecnología que permite crear una capa abstracta sobre los datos reales para ocultar su complejidad y mejorar su accesibilidad. Esa nueva capa se encarga de agrupar los datos de distintos orígenes y complejidad, exponiéndolos al usuario de una forma homogénea e integrada. La virtualización de datos permite a los usuarios acceder a los datos de una manera más sencilla y rápida, sin tener que preocuparse por la dificultad subyacente de la infraestructura de datos. Facilitaría el cruce de información entre distintos orígenes (SoR) almacenados en software diferentes. Por ejemplo, seríamos capaces de cruzar datos de una tabla de Oracle con otra de MongoDB ubicadas en distintas máquinas sin necesidad de construir ETLs o replicar datos en capas intermedias.

La virtualización de datos es un componente clave del Data Fabric, ya que permite crear una vista unificada de los datos que se encuentran en diferentes formatos, estructuras y localizaciones. Permite a los usuarios acceder a los datos de una manera sencilla, al tiempo que mantiene la seguridad, la privacidad y la calidad de los mismos.

Además, permite modificar los datos virtuales sin afectar a los datos reales, lo que significa que los datos pueden ser integrados, migrados y gestionados de forma más eficiente. La gestión de los datos se vuelve más flexible y escalable, lo que significa que los usuarios pueden adaptarse rápidamente a los cambios en las necesidades de negocios y a las nuevas fuentes de datos.

Una de las desventajas de la virtualización de datos es el impacto en los sistemas fuente, ya que cuando se realiza una consulta en realidad se está atacando directamente el origen. Para mitigar problemas de rendimiento, se pueden cachear las consultas en una capa intermedia (una BD) de forma que cuando se consulte una vista virtualizada en primer lugar lo haga sobre la caché. El ciclo de vida de esta caché se puede configurar y gestionar desde el software de virtualización. Otra forma de mejorar el rendimiento es mapeando datos entre los orígenes, si consultamos datos de dos fuentes distintas el sistema copiaría los datos menos voluminosos en el otro origen para realizar la consulta sobre la misma ubicación.

La virtualización de datos está construida por un catálogo de vistas que se alimentan de los distintos orígenes de datos cuya comunicación se configura a través de conectores. Por tanto, podríamos decir que para construir una capa de virtualización de datos necesitaríamos en primer lugar crear los conectores a los distintos orígenes (Oracle, Teradata, Azure PaaS SQL Server, Google BigQuery, PostgreSQL, mongoDB, Profile, SQL Server…) y después generar las vistas que extraen información de estos orígenes. El cemento que une todo el sistema es un catálogo de metadata que se encarga de almacenar la ubicación, conexión y tipología de cada dato.

Conectores

Como en cualquier herramienta de integración de datos, los conectores se encargan de abrir la conexión del origen de datos con nuestro sistema. El propio software de virtualización de datos (Denodo, CP4D) dispone de conectores ya prefabricados que solo debemos configurar para abrir la comunicación.

Vistas

Una vista es una representación de datos de un origen específico, el concepto sería similar a una vista tradicional en una BBDD. Podemos encontrar base views que sería una vista generada directamente de una tabla o query del mismo origen de datos, o derived views, que serían la combinación de varias base views donde ya podrían cruzarse datos de distintos orígenes. Al utilizar los conectores, la vista permite el acceso a los datos sin tener que conocer su ubicación o formato real.

Catálogo de metadatos

El catálogo de metadatos sería donde se almacena toda la información de los datos: ubicación, formato, estructura y relación con otros datos. Puede utilizarse para ayudar a la toma de decisiones sobre la gestión, incluyendo la seguridad, la calidad y la integridad de los datos. También puede ser utilizado para optimizar el rendimiento del sistema y para mejorar la eficiencia.

La entrada ¿Qué es una arquitectura Data Fabric? se publicó primero en Marcos Torregrosa.