En las dos partes anteriores, presentamos el modelo de base de datos en vivo para un negocio basado en suscripción y un almacén de datos (DWH) que podríamos usar para generar informes. Si bien es obvio que deberían funcionar juntos, no hubo conexión entre estos dos modelos. Hoy, daremos el siguiente paso y escribiremos el código para transferir datos de la base de datos en vivo a nuestro DWH.
Los modelos de datos
Antes de sumergirnos en el código, recordemos los dos modelos con los que trabajaremos. Primero está el modelo de datos transaccionales que usaremos para almacenar nuestros datos en tiempo real. Teniendo en cuenta que operamos un negocio basado en suscripciones, necesitaremos almacenar los detalles de los clientes y las suscripciones, los pedidos de los clientes y los estados de los pedidos.
Realmente hay mucho que podríamos agregar a este modelo, como el seguimiento de pagos y el almacenamiento de datos históricos (especialmente cambios en los datos de clientes y suscripciones). Sin embargo, para enfatizar el proceso ETL (extracción, transformación y carga), quiero mantener este modelo lo más simple posible.
El uso de un modelo de datos transaccionales como base de datos de informes podría funcionar en algunos casos, pero no funcionará en todos los casos. Ya lo hemos mencionado, pero vale la pena repetirlo. Si queremos separar nuestras tareas de informes de nuestros procesos en tiempo real, debemos crear algún tipo de base de datos de informes. Un almacén de datos es una solución.
Nuestro DWH se centra en cuatro tablas de hechos. Los dos primeros rastrean el número de clientes y suscripciones a nivel diario. Los dos restantes rastrean el número de entregas y los productos incluidos en estas entregas.
Mi suposición es que ejecutaremos nuestro proceso ETL una vez al día. Primero, completaremos las tablas de dimensiones con nuevos valores (cuando sea necesario). Después de eso, completaremos las tablas de hechos.
Para evitar repeticiones innecesarias, demostraré solo el código que llenará las dos primeras tablas de dimensiones y las dos primeras tablas de hechos. Las tablas restantes se pueden completar con un código muy similar. Te animo a que escribas el código tú mismo. No hay mejor manera de aprender algo nuevo que probándolo.
La idea:tablas de dimensiones
La idea general es crear procedimientos almacenados que podamos usar regularmente para llenar el DWH:tablas de dimensiones y tablas de hechos. Estos procedimientos transferirán datos entre dos bases de datos en el mismo servidor. Esto significa que algunas consultas dentro de estos procedimientos usarán tablas de ambas bases de datos. Esto se espera; necesitamos comparar el estado del DWH con la base de datos en vivo y hacer cambios en el DWH de acuerdo con lo que sucede en la base de datos en vivo.
Tenemos cuatro tablas de dimensiones en nuestro DWH:dim_time , dim_city , dim_product y dim_delivery_status .
La dimensión de tiempo se rellena añadiendo la fecha anterior. La suposición principal es que ejecutaremos este procedimiento diariamente, después del cierre de operaciones.
Las dimensiones de la ciudad y del producto dependerán de los valores actuales almacenados en la city y product diccionarios en la base de datos en vivo. Si agregamos algo a estos diccionarios, se agregarán nuevos valores a las tablas de dimensiones en la próxima actualización de DWH.
La última tabla de dimensiones es dim_delivery_status mesa. No se actualizará porque solo contiene tres valores predeterminados. Una entrega está en tránsito, cancelada o entregada.
La idea:tablas de hechos
Llenar tablas de hechos es en realidad el verdadero trabajo. Si bien los diccionarios en la base de datos en vivo no contienen un atributo de marca de tiempo, las tablas con datos insertados como resultado de nuestras operaciones sí lo contienen. Notará dos atributos de marca de tiempo, time_inserted y time_updated , en el modelo de datos.
Nuevamente, asumo que ejecutaremos con éxito la importación de DWH una vez al día. Esto nos permite agregar los datos a nivel diario. Contaremos la cantidad de clientes y suscripciones activos y cancelados, así como las entregas y los productos entregados para esa fecha.
Nuestro modelo en vivo funciona bien si ejecutamos un procedimiento de inserción después del COB (cierre comercial). Aún así, si queremos más flexibilidad, deberíamos hacer algunos cambios en el modelo. Uno de esos cambios podría ser tener una tabla de historial separada para rastrear el momento exacto en que cambiaron los datos relacionados con los clientes o las suscripciones. Con nuestra organización actual, sabremos que se produjo el cambio, pero no sabremos si hubo cambios antes de este (por ejemplo, un cliente canceló ayer, reactivó su cuenta después de la medianoche y luego canceló nuevamente hoy) .
Rellenar tablas de dimensiones
Como se mencionó anteriormente, supondré que ejecutaremos la importación de DWH exactamente una vez al día. Si ese no es el caso, necesitaríamos un código adicional para eliminar los datos recién insertados de las tablas de hechos y dimensiones. Para las tablas de dimensiones, esto se limitaría a eliminar la fecha dada.
Primero, verificaremos si la fecha dada existe en el dim_time mesa. Si no, agregaremos una nueva fila a la tabla; si lo hace, no tenemos que hacer nada. En la mayoría de los casos, todas las fechas se insertan durante la implementación de producción inicial. Pero usaré este ejemplo con fines educativos.
Para la dim_city y dim_product dimensiones, agregaré solo cualquier valor nuevo que detecte en la city y product mesas. No haré ninguna eliminación porque se podría hacer referencia a cualquier valor insertado previamente en alguna tabla de hechos. Podríamos ir con una eliminación suave, p. tener una bandera "activa" que podríamos activar y desactivar.
Para la última tabla, dim_delivery_status , no haré nada porque siempre contendrá los mismos tres valores.
El siguiente código crea un procedimiento que llenará las tablas de dimensiones dim_time y dim_city .
Para la dimensión de tiempo, agregaré la fecha de ayer. Voy a suponer que el proceso ETL comienza justo después de la medianoche. Verificaré si esa dimensión ya existe y, si no, agregaré la nueva fecha en la tabla.
Para la dimensión de la ciudad, usaré LEFT JOIN para unir datos de la base de datos en vivo y la base de datos DWH para determinar qué filas faltan. Luego agregaré solo los datos faltantes a la tabla de dimensiones. Vale la pena mencionar que hay algunas formas de comprobar si se han cambiado los datos. Este proceso se denomina captura de datos modificados o CDC. Un método común es buscar marcas de tiempo o versiones actualizadas. Hay algunas formas adicionales, pero están fuera del alcance de este artículo.
Echemos un vistazo al código ahora, que está escrito usando la sintaxis de MySQL .
DROP PROCEDURE IF EXISTS p_update_dimensions//
CREATE PROCEDURE p_update_dimensions ()
BEGIN
SET @time_exists = 0;
SET @time_date = DATE_ADD(DATE(NOW()), INTERVAL -1 DAY);
-- procedure populates dimension tables with new values
-- dim_time
SET @time_exists = (SELECT COUNT(*) FROM subscription_dwh.dim_time dim_time WHERE dim_time.time_date = @time_date);
IF (@time_exists = 0) THEN
INSERT INTO subscription_dwh.`dim_time`(`time_date`, `time_year`, `time_month`, `time_week`, `time_weekday`, `ts`)
SELECT
@time_date AS time_date,
YEAR(@time_date) AS time_year,
MONTH(@time_date) AS time_month,
WEEK(@time_date) AS time_week,
WEEKDAY(@time_date) AS time_weekday,
NOW() AS ts;
END IF;
-- dim_city
INSERT INTO subscription_dwh.`dim_city`(`city_name`, `postal_code`, `country_name`, `ts`)
SELECT
city_live.city_name,
city_live.postal_code,
country_live.country_name,
Now()
FROM subscription_live.city city_live
INNER JOIN subscription_live.country country_live
ON city_live.country_id = country_live.id
LEFT JOIN subscription_dwh.dim_city city_dwh
ON city_live.city_name = city_dwh.city_name
AND city_live.postal_code = city_dwh.postal_code
AND country_live.country_name = city_dwh.country_name
WHERE city_dwh.id IS NULL;
END//
-- CALL p_update_dimensions ()
Ejecutar este procedimiento, que hacemos usando el procedimiento comentado CALL -- inserta una nueva fecha y todas las ciudades faltantes en las tablas de dimensiones. Intente agregar su propio código para completar las dos tablas de dimensiones restantes con nuevos valores.
El proceso ETL en un almacén de datos
La idea principal detrás del almacenamiento de datos es contener datos agregados en el formato deseado. Por supuesto, debemos conocer ese formato incluso antes de comenzar a construir el almacén. Si hemos hecho todo según lo planeado, podemos obtener todos los beneficios que nos ofrece un DWH. El principal beneficio es un rendimiento mejorado al ejecutar consultas. Nuestras consultas funcionan con menos registros (porque se agregan) y se ejecutan en la base de datos de informes (en lugar de la activa).
Pero antes de que podamos consultar, necesitamos almacenar hechos en nuestra base de datos. La forma en que lo haremos depende de lo que necesitemos hacer con nuestros datos más adelante. Si no tenemos una buena imagen general antes de comenzar a construir nuestro DWH, ¡pronto podríamos encontrarnos en problemas! pronto.
El nombre de este proceso es ETL:E =Extraer, T =Transformar, L =Cargar. Toma los datos, los transforma para adaptarse a la estructura DWH y los carga en el DWH. Para ser precisos, el proceso real que usaremos es ELT:Extraer, Cargar, Transformar. Dado que estamos utilizando procedimientos almacenados, extraeremos datos, los cargaremos y luego los transformaremos para satisfacer nuestras necesidades. Es bueno saber que, si bien ETL y ELT son ligeramente diferentes, los términos a veces se usan indistintamente.
Rellenar las tablas de hechos
Rellenar las tablas de hechos es la razón por la que realmente estamos aquí. Hoy, llenaré dos tablas de hechos, la fact_customer_subscribed tabla y el fact_subscription_status mesa. Las dos tablas de hechos restantes son suyas para que las pruebe como tarea.
Antes de pasar a llenar la tabla de hechos, debemos suponer que las tablas de dimensiones se llenan con nuevos valores. Rellenar las tablas de hechos sigue el mismo patrón. Como tienen la misma estructura, los explicaré a ambos juntos.
Estamos agrupando datos por dos dimensiones:tiempo y ciudad. La dimensión de tiempo se establecerá en ayer y encontraremos el ID del registro relacionado en el dim_time tabla comparando fechas (el último INNER JOIN en ambas consultas).
El ID de dim_city se extrae uniendo todos los atributos que forman una combinación ÚNICA en la tabla de dimensiones (nombre de la ciudad, código postal y nombre del país).
En esta consulta, probaremos los valores con CASE y luego los SUMAREMOS. Para clientes activos e inactivos, no he probado la fecha. Sin embargo, he seleccionado valores tal cual para estos campos. Para cuentas nuevas y canceladas, probé la hora actualizada.
DROP PROCEDURE IF EXISTS p_update_facts//
CREATE PROCEDURE p_update_facts ()
BEGIN
SET @time_date = DATE_ADD(DATE(NOW()), INTERVAL -1 DAY);
-- procedure populates fact tables with new values
-- fact_customer_subscribed
INSERT INTO `fact_customer_subscribed`(`dim_city_id`, `dim_time_id`, `total_active`, `total_inactive`, `daily_new`, `daily_canceled`, `ts`)
SELECT
city_dwh.id AS dim_ctiy_id,
time_dwh.id AS dim_time_id,
SUM(CASE WHEN customer_live.active = 1 THEN 1 ELSE 0 END) AS total_active,
SUM(CASE WHEN customer_live.active = 0 THEN 1 ELSE 0 END) AS total_inactive,
SUM(CASE WHEN customer_live.active = 1 AND DATE(customer_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_new,
SUM(CASE WHEN customer_live.active = 0 AND DATE(customer_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_canceled,
MIN(NOW()) AS ts
FROM subscription_live.`customer` customer_live
INNER JOIN subscription_live.`city` city_live ON customer_live.city_id = city_live.id
INNER JOIN subscription_live.`country` country_live ON city_live.country_id = country_live.id
INNER JOIN subscription_dwh.dim_city city_dwh
ON city_live.city_name = city_dwh.city_name
AND city_live.postal_code = city_dwh.postal_code
AND country_live.country_name = city_dwh.country_name
INNER JOIN subscription_dwh.dim_time time_dwh ON time_dwh.time_date = @time_date
GROUP BY
city_dwh.id,
time_dwh.id;
-- fact_subscription_status
INSERT INTO `fact_subscription_status`(`dim_city_id`, `dim_time_id`, `total_active`, `total_inactive`, `daily_new`, `daily_canceled`, `ts`)
SELECT
city_dwh.id AS dim_ctiy_id,
time_dwh.id AS dim_time_id,
SUM(CASE WHEN subscription_live.active = 1 THEN 1 ELSE 0 END) AS total_active,
SUM(CASE WHEN subscription_live.active = 0 THEN 1 ELSE 0 END) AS total_inactive,
SUM(CASE WHEN subscription_live.active = 1 AND DATE(subscription_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_new,
SUM(CASE WHEN subscription_live.active = 0 AND DATE(subscription_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_canceled,
MIN(NOW()) AS ts
FROM subscription_live.`customer` customer_live
INNER JOIN subscription_live.`subscription` subscription_live ON subscription_live.customer_id = customer_live.id
INNER JOIN subscription_live.`city` city_live ON customer_live.city_id = city_live.id
INNER JOIN subscription_live.`country` country_live ON city_live.country_id = country_live.id
INNER JOIN subscription_dwh.dim_city city_dwh
ON city_live.city_name = city_dwh.city_name
AND city_live.postal_code = city_dwh.postal_code
AND country_live.country_name = city_dwh.country_name
INNER JOIN subscription_dwh.dim_time time_dwh ON time_dwh.time_date = @time_date
GROUP BY
city_dwh.id,
time_dwh.id;
END//
-- CALL p_update_facts ()
Una vez más, comenté la última línea. Elimine el comentario y puede usar esta línea para llamar al procedimiento e insertar nuevos valores. Tenga en cuenta que no he eliminado ningún valor anterior existente, por lo que este procedimiento no funcionará si ya tenemos valores para esa fecha y ciudad. Esto se puede solucionar realizando eliminaciones antes de las inserciones.
Recuerde, necesitamos completar las tablas de hechos restantes en nuestro DWH. ¡Te animo a que lo pruebes tú mismo!
Otra cosa que definitivamente recomendaría es colocar todo el proceso dentro de una transacción. Eso aseguraría que todas las inserciones tengan éxito o que no se realice ninguna. Esto es muy importante cuando queremos evitar que los datos se inserten parcialmente, p. si tenemos múltiples procedimientos para insertar dimensiones y hechos y algunos de ellos hacen su trabajo mientras que otros fallan.
¿Qué piensas?
Hoy hemos visto cómo podemos realizar el proceso ELT/ETL y cargar datos desde una base de datos activa en un almacén de datos. Si bien el proceso que demostramos está bastante simplificado, contiene todos los elementos necesarios para E (extraer) los datos, T (transformarlos) en un formato adecuado y, finalmente, L (cargarlos) en el DWH. ¿Qué piensas? Cuéntenos sus experiencias en los comentarios a continuación.
