Carga incremental de Dimensiones Tipo 2 en tablas Delta con PySpark

Las cargas incrementales son una de las herramientas más útiles de cara a desarrollar nuestras soluciones, al reducir el tiempo y los recursos empleados durante la sincronización de datos. No obstante, ante el nuevo paradigma de delta nos encontramos con algunas diferencias frente a las tradicionales tablas de SQL Server que debemos suplir si queremos seguir cargando dimensiones de tipo 2 de forma similar.

¡Comenzamos!

La carga incremental es el proceso de cargar únicamente los datos que han entrado nuevos o se han actualizado en nuestro origen de datos. Esta característica nos permite ahorrar tiempo y recursos en comparación con las cargas completas o diferenciales, al partir desde lo último que se cargó.

Para poder quedarnos siempre con el dato más reciente se suele utilizar como referencia un campo numérico autoincremental o uno de tipo fecha. Para evitar conflictos de duplicados, las cargas incrementales tienen que estar preparadas para identificar qué registros son nuevos y cuáles ya existen, para insertar los primeros y actualizar los segundos.

Las dimensiones lentamente cambiantes de tipo 2 son las tablas que presentan la capacidad de guardar un histórico de sus elementos, necesarios en algunos escenarios empresariales donde se realizan análisis a lo largo del tiempo de forma precisa. Esta historificación se lleva a cabo insertando una nueva fila por cada una de las versiones del atributo, que presentan su vigencia acotada por las fechas de inicio y fin, y una clave subrogada que actúa como la clave primaria de la tabla.

Esta peculiaridad provoca que a la hora de cargar las dimensiones debamos tener procesos que insertan los nuevos elementos de las dimensiones a la vez que se dan por finalizadas las versiones actuales, en el caso de estar ya presentes. De forma que, a estas nuevas inserciones o versiones se les asigna una nueva clave subrogada completamente independiente de su clave de negocio.

Las dimensiones lentamente cambiantes de tipo 2 son las tablas que presentan la capacidad de guardar un histórico de sus elementos, necesarios en algunos escenarios empresariales donde se realizan análisis a lo largo del tiempo de forma precisa. Esta historificación se lleva a cabo insertando una nueva fila por cada una de las versiones del atributo, que presentan su vigencia acotada por las fechas de inicio y fin, y una clave subrogada que actúa como la clave primaria de la tabla.

Esta peculiaridad provoca que a la hora de cargar las dimensiones debamos tener procesos que insertan los nuevos elementos de las dimensiones a la vez que se dan por finalizadas las versiones actuales, en el caso de estar ya presentes. De forma que, a estas nuevas inserciones o versiones se les asigna una nueva clave subrogada completamente independiente de su clave de negocio.

Las Tablas Delta son una evolución de las tablas basadas en ficheros Parquet, utilizando el formato Delta Lake, diseñado para entornos con grandes volúmenes de datos. Destacan por su consistencia, integridad y capacidad de auditoría. Entre sus características se encuentran:

Las tablas Delta están preparadas para los cambios en el esquema, ya sean añadiendo o modificando columnas, manteniendo la compatibilidad con los datos ya existentes.

Están diseñadas para manejar grandes volúmenes de datos distribuidos a través del Data Lake en Microsoft Fabric.

En Fabric podemos desarrollar nuestras canalizaciones con notebooks que permiten a los usuarios escribir y ejecutar código en distintos lenguajes como Python, SQL y Scala, lo que facilita el trabajo con grandes volúmenes de datos en entornos escalables.

Gracias a su integración con el ecosistema de Microsoft, los notebooks pueden conectarse de manera nativa con OneLake, Power BI y Azure Synapse, permitiendo una experiencia unificada para la ingeniería y análisis de datos. Además, ofrecen capacidades de ejecución distribuida gracias a Apache Spark, lo que mejora el rendimiento en tareas intensivas de procesamiento de datos.

PySpark es la API de Python para Apache Spark, un entorno de computación distribuida diseñado para procesar grandes volúmenes de datos de manera eficiente. Gracias a su integración con Spark, permite ejecutar operaciones con RDDs (Resilient Distributed Datasets) y DataFrames, facilitando la manipulación de datos a gran escala con una sintaxis similar a Pandas.

En el contexto de Microsoft Fabric, PySpark se convierte en una herramienta fundamental dentro de los notebooks, ya que permite ejecutar consultas y transformaciones de datos aprovechando la potencia de Spark en la nube. Esto lo hace ideal para casos de uso como procesamiento ETL, Machine Learning y análisis avanzado de datos.

En los escenarios de BI comunes donde disponemos de un Data Warehouses en SQL Server On-Premises o en la nube, nos encontrábamos con la característica IDENTITY que se asignaba a las claves subrogadas de las dimensiones. Ahora, con las tablas Delta nos encontramos con la problemática de que las IDENTITY desaparecen, por lo que el versionado de las SCD2 tiene que ser rediseñado.

Para ello, proponemos una solución para emular el atributo IDENTITY almacenando en una variable el último valor de la clave subrogada cargada, para después con código escrito en Spark SQL y la función de ventana ROW_NUMBER(), gestionar y mantener los índices autoincreméntales en los elementos que entren nuevos o hayan cambiado de versión.

1) Partiendo de la siguiente dimensión que presenta el nombre “tabla_delta”:

Con estas primeras líneas de código importamos las funciones del módulo pyspark.sql.functions, almacenamos la pk en la variable key_column, guardamos los campos en una lista, la reorganizamos para dejar fromDate al final y convertimos esa lista en un string separado por comas.

2) Suponiendo que en la variable sdf tenemos almacenado el DataFrame con los datos que han entrado en la última ejecución:

Con el siguiente código le añadimos el campo toDate y isCurrent, además de añadir un hash con la función xxhash64. Este hash puede estar compuesto por todos los elementos de la dimensión o únicamente por los que nos interesen para versionar.

3) Después, almacenamos en la variable next_surrogate_key el valor de la última sk cargada en la tabla, y guardamos en una vista temporal el DataFrame sdf.

4) Por último, añadimos las líneas de Spark SQL encargadas de finalizar las versiones actuales de la dimensión y de insertar las nuevas.

Tras ejecutar todo el código el resultado de la dimensión sería el siguiente:

En el cual disponemos de dos versiones para los id 2 y 6 y la nueva inserción para el id 7, cada una de ellas con su propia clave subrogada (sk).

Las cargas incrementales en el contexto de las SCD2 nos permiten mantener un histórico en aquellas dimensiones que lo precisen en función de los requisitos de negocio. Con este método podemos mantener de forma efectiva las claves subrogadas en el nuevo paradigma de Delta en Microsoft Fabric sin las IDENTITY propias de SQL Server.