ETL en Databricks SQL

Al tratar con grandes cantidades de datos, necesita una canalización que pueda procesar solo los registros nuevos y modificados en lugar de volver a procesar todo el conjunto de datos. Esto se denomina ETL incremental. En Databricks SQL, puede crear canalizaciones ETL incrementales mediante tablas de streaming y vistas materializadas, sin escribir código de procedimientos ni programar actualizaciones manuales.

Este tutorial le guía a través de un patrón común: realizar un seguimiento de los cambios de producto a lo largo del tiempo. Se crea una tabla de origen, se capturan los eventos de cambio, se construye una tabla de dimensiones que preserve el historial completo de cada producto, se mantiene una tabla de hechos que sobrescribe de forma incremental los registros recientes y se añade una capa agregada de informes encima. Usas AUTO CDC para la tabla de dimensiones y un flujo REPLACE WHERE para la tabla de hechos.

La característica clave de este tutorial es AUTO CDC. En un almacenamiento tradicional, escribiría instrucciones complejas MERGE INTO para conciliar los eventos de inserción, actualización y eliminación en una tabla de destino. Este enfoque es propenso a errores, especialmente cuando los eventos llegan fuera de orden. AUTO CDC gestiona esto para usted. Declara la clave de negocio, la columna de secuenciación y si desea SCD Type 1 (solo valor más reciente) o SCD Type 2 (historial completo) y Azure Databricks aplica automáticamente la lógica de combinación correcta. Para obtener información general sobre CDC, consulte Las AUTO CDC API: Simplificación de la captura de cambios en datos con canalizaciones.

Al final de este tutorial, tendrá:

  1. Se ha creado una tabla de origen que realiza un seguimiento de los cambios con el Change Data Feed.
  2. Inspeccionó los datos de cambio sin procesar para comprender la secuencia de eventos CDC.
  3. Se usa AUTO CDC para crear una tabla de dimensiones de tipo 2 de SCD a partir de esos eventos.
  4. Los eventos de eliminación se procesan de forma incremental a través del canal.
  5. Se ha creado una tabla de datos que sobrescribe de forma incremental las filas recientes con un flujo de REPLACE WHERE.
  6. Ha creado una vista materializada que mantiene incrementalmente un reporte agregado.
  7. Configurado SCHEDULE REFRESH EVERY 1 DAY para que los cambios se propaguen automáticamente a través de la tubería.

Requisitos

Para completar este tutorial, debe cumplir los siguientes requisitos:

Paso 1: Configurar el catálogo y el esquema

Abra el editor de SQL de Databricks y establezca el catálogo de trabajo y el esquema. Debe tener permiso para USE el catálogo y el esquema que seleccione:

USE CATALOG <your-catalog>;
USE SCHEMA <your-schema>;

Paso 2: Creación de una tabla de origen y carga de datos

Cree una products tabla con Use change data feed on Azure Databricks (CDF) habilitado. CDF es una característica de Delta Lake que registra cada inserción, actualización y eliminación como registro de cambios consultable. Esto es similar a un flujo CDC desde un sistema de origen transaccional, excepto que los cambios se capturan directamente dentro de la tabla Delta en lugar de desde un registro externo. Aquí se utiliza CDF para generar los eventos de cambio que consumirá el pipeline posterior.

  1. Cree la tabla y cargue los registros iniciales:

    CREATE OR REPLACE TABLE products (
  product_id INT,
  product_name STRING,
  category STRING,
  warehouse STRING
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

INSERT INTO products VALUES
  (1, 'Spoon', 'Cutlery', 'Seattle'),
  (2, 'Fork', 'Cutlery', 'Portland'),
  (3, 'Knife', 'Cutlery', 'Denver'),
  (4, 'Chair', 'Furniture', 'Austin'),
  (5, 'Table', 'Furniture', 'Chicago'),
  (6, 'Lamp', 'Lighting', 'Boston'),
  (7, 'Mug', 'Kitchenware', 'Seattle'),
  (8, 'Plate', 'Kitchenware', 'Atlanta'),
  (9, 'Bowl', 'Kitchenware', 'Dallas'),
  (10, 'Glass', 'Kitchenware', 'Phoenix');

  2. Simulación de cambios ascendentes, incluidos los nuevos productos, un traslado de almacenamiento y una reasignación de categorías:

    INSERT INTO products VALUES
  (11, 'Napkin', 'Dining', 'San Francisco'),
  (12, 'Coaster', 'Dining', 'New York');

UPDATE products SET warehouse = 'Los Angeles' WHERE product_id = 1;
UPDATE products SET category = 'Dining' WHERE product_id = 2;

Paso 3: Consulta del flujo de datos de cambios

Antes de crear el pipeline posterior, resulta útil examinar los eventos de cambio sin procesar para comprender qué AUTO CDC se va a procesar. La table_changes() función lee el registro de CDF y devuelve cada operación capturada junto con columnas de metadatos:

SELECT
  product_id, product_name, warehouse,
  _change_type, _commit_version
FROM table_changes('products', 1)
ORDER BY _commit_version, product_id;

Por ejemplo, Spoon tiene tres eventos: un insert (Seattle), un update_preimage (Seattle) y un update_postimage (Los Ángeles).

Observe que un único cambio lógico (por ejemplo, mover la cuchara a un almacén diferente) produce varios eventos: una imagen previa y una postimage. En un almacenamiento tradicional, escribiría una MERGE instrucción para conciliar todos estos eventos en una tabla de destino, controlar inserciones, actualizaciones y eliminaciones con lógica independiente y asegurarse de que los eventos se aplican en el orden correcto. Esta es exactamente la complejidad que AUTO CDC elimina en el paso siguiente.

Paso 4: Crear una dimensión de tipo 2 de SCD con AUTO CDC

Una tabla de streaming procesa los datos de forma incremental. En cada actualización, solo lee las nuevas filas desde la última ejecución, por lo que no es necesario volver a procesar el conjunto de datos completo. Esto hace que sea adecuado para orígenes de gran volumen o con frecuencia cambiantes.

AUTO CDC añade el procesamiento de captura de datos de cambio sobre una tabla de streaming. En lugar de escribir una instrucción MERGE INTO que controla manualmente las inserciones, las actualizaciones y las eliminaciones, declara la clave de negocio y la columna de secuenciación y permite que Azure Databricks apliquen la lógica correcta. AUTO CDC también controla los eventos desordenados automáticamente, que es un problema común al usar MERGE INTO para controlar eventos que llegan desde sistemas distribuidos o cargas por lotes con marcas de tiempo superpuestas.

La siguiente instrucción crea una tabla SCD Type 2 que conserva el historial de versiones completo de cada producto. Cada versión obtiene __START_AT y __END_AT marcas de tiempo. Un NULL en __END_AT marca la versión actual.

CREATE OR REFRESH STREAMING TABLE products_history
SCHEDULE REFRESH EVERY 1 DAY
FLOW AUTO CDC
FROM (
  SELECT * FROM STREAM products WITH (readChangeFeed = true)
  WHERE _change_type != 'update_preimage'
)
KEYS (product_id)
APPLY AS DELETE WHEN _change_type = 'delete'
SEQUENCE BY _commit_timestamp
COLUMNS * EXCEPT (_change_type, _commit_version, _commit_timestamp)
STORED AS SCD TYPE 2;
  • SCHEDULE REFRESH EVERY 1 DAY: actualiza la tabla según una programación diaria.
  • FLOW AUTO CDC: declara esto como un flujo CDC. Azure Databricks aplica automáticamente la semántica de inserción, actualización y eliminación.
  • WHERE _change_type != 'update_preimage': omite las imágenes previas. La postimage ya tiene la fila actualizada.
  • KEYS (product_id): la clave de negocio. Los eventos con la misma clave se combinan en filas versionadas.
  • APPLY AS DELETE WHEN _change_type = 'delete': cierra la versión actual cuando llega un evento de eliminación. Esto le permite definir la condición que identifica un evento de eliminación.
  • SEQUENCE BY _commit_timestamp: establece el orden de eventos. Controla las llegadas desordenadas correctamente.
  • STORED AS SCD TYPE 2: conserva el historial completo. AUTO CDC admite el tipo 1 de SCD y el tipo 2 de SCD.

Consulte la tabla de dimensiones:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
  • Cuchara: dos versiones. Seattle (cerrado, __END_AT establecido) y Los Ángeles (actual, __END_AT = NULL).
  • Bifurcación: dos versiones. Categoría Cubiertos (cerrada) y categoría Comedor (actual).
  • Servilletas y posavasos: una versión de cada uno (recién insertados, __END_AT = NULL).
  • Todos los demás productos: una versión cada una (__END_AT = NULL).

Paso 5: Procesar las eliminaciones a través del canal

Ahora, simulación de dos productos descontinuados eliminandolos de la tabla de origen:

DELETE FROM products WHERE product_id = 9;
DELETE FROM products WHERE product_id = 10;

Estos eventos de eliminación se registran en el registro de CDF, pero la tabla de streaming aún no los ha visto. Actualice la tabla de streaming para procesar los nuevos eventos:

REFRESH STREAMING TABLE products_history;

Consulte la tabla de dimensiones para comprobar que se aplicaron las eliminaciones:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;

Los cuencos y los vasos ahora están cerrados con el conjunto __END_AT, lo que los marca como descatalogados. Todos los demás productos actuales permanecen sin cambios. La tabla de streaming solo procesó los nuevos eventos de eliminación sin volver a procesar las inserciones y actualizaciones de la actualización anterior.

Paso 6: Crea una tabla de hechos con un flujo REPLACE WHERE

La tabla de dimensiones en los pasos anteriores se usa AUTO CDC porque registra el estado actual e histórico de cada producto. Las tablas de hechos tienen necesidades diferentes. Una tabla de hechos registra eventos, como ventas, y normalmente solo se reprocesa una ventana reciente, por ejemplo, para corregir registros que llegan tarde. Para este patrón, se utiliza un flujo incremental de REPLACEWHERE, que sobrescribe solo las filas que coinciden con un predicado y deja todas las demás filas intactas. Para una comparación completa de este constructo con vistas materializadas, véase Cuándo usar REEMPLAZAR WHERE vs. vistas materializadas.

  1. Crea una product_sales tabla y carga algunos eventos de ventas, algunos recientes y otros antiguos:

    CREATE OR REPLACE TABLE product_sales (
  sale_id INT,
  product_id INT,
  quantity INT,
  sale_date DATE
);

INSERT INTO product_sales VALUES
  (101, 1, 3, current_date()),
  (102, 2, 1, current_date()),
  (103, 4, 5, date_add(current_date(), -2)),
  (104, 6, 2, date_add(current_date(), -10));

  2. Crea una tabla de hechos que recalcule solo los últimos 7 días de ventas en cada actualización:

    CREATE OR REFRESH STREAMING TABLE product_sales_recent
SCHEDULE REFRESH EVERY 1 DAY
FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7) BY NAME
SELECT sale_id, product_id, quantity, sale_date
FROM product_sales;
    • FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7): en cada actualización, elimina las filas del destino que coinciden con el predicado, recalcula la consulta de origen para ese mismo rango e inserta los resultados. Las filas fuera de la ventana de 7 días quedan intactas.
    • SCHEDULE REFRESH EVERY 1 DAY: actualiza la tabla con el mismo horario diario que la tabla de dimensiones.
    • BY NAME: obligatorio. Empareja las columnas por nombre en lugar de por posición.

    La venta de hace 10 días queda fuera del predicado, así que el flujo no la carga. Solo se mantienen las ventas de los últimos 7 días.

  3. Consulta la tabla de hechos:

    SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

    El resultado contiene las tres ventas de los últimos 7 días. Los sale_date valores dependen del día en que ejecutes el tutorial:

    sale_id product_id cantidad fecha_de_venta
    101 1 3 Hoy
    102 2 1 Hoy
    103 4 5 Hace 2 días

    La venta 104 no aparece en el resultado porque su fecha está fuera del plazo de 7 días.

Paso 7: Crear una vista materializada agregada

Ahora que tiene una tabla de dimensiones que permanece actualizada con los cambios de origen, puede agregar una capa de informes encima.

Una vista materializada almacena los resultados de las consultas calculadas previamente como una tabla física. A diferencia de una vista normal, que vuelve a ejecutar la consulta cada vez que la lee, una vista materializada conserva los resultados y solo vuelve a calcular las filas afectadas por los cambios ascendentes en cada actualización. Esto hace que sea adecuado para paneles e informes en los que el rendimiento de las consultas es importante.

CREATE OR REPLACE MATERIALIZED VIEW products_by_category
SCHEDULE REFRESH EVERY 1 DAY
AS
SELECT
  category,
  COUNT(*) AS active_products
FROM products_history
WHERE __END_AT IS NULL
GROUP BY category;

SCHEDULE REFRESH EVERY 1 DAY significa que esta vista se actualiza según una programación diaria. En combinación con la misma programación en la tabla de streaming, ahora dispones de un canal de tres etapas en el que los cambios en la tabla de origen se propagan a través de la dimensión y llegan al agregado en cada ciclo de actualización. No necesitas actualizar manualmente la dimensión ni el agregado.

SELECT * FROM products_by_category ORDER BY active_products DESC;

Paso 8: Verifica la cascada de extremo a extremo

Para verificar la cascada completa del pipeline, haz un cambio en la tabla de origen y registra una venta atrasada de ayer:

UPDATE products SET warehouse = 'Seattle' WHERE product_id = 3;

INSERT INTO product_sales VALUES (105, 3, 4, date_add(current_date(), -1));

El cuchillo se mueve de Denver a Seattle. Esto UPDATE activa la cascada completa de la canalización, demostrando cómo funcionan juntas las tres etapas:

  1. products registra el evento de cambio en el feed de datos de cambios.
  2. products_history procesa el evento y añade una nueva versión para el cuchillo.
  3. products_by_category vuelve a calcular únicamente la fila de Cubiertos afectada.

La tabla de datos se actualiza según su propio calendario. Para incluir la venta que llegó con retraso ahora en lugar de esperar la siguiente actualización programada, actualiza manualmente la tabla de datos:

REFRESH STREAMING TABLE product_sales_recent;

El flujo recalcula los últimos 7 días desde product_sales, por lo que la venta 105 ahora aparece en product_sales_recent.

Compruebe lo siguiente:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
WHERE product_id = 3
ORDER BY __START_AT;

SELECT * FROM products_by_category ORDER BY active_products DESC;

SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

Limpieza

Para limpiar los recursos creados por este tutorial, use el siguiente código SQL:

DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP TABLE IF EXISTS product_sales_recent;
DROP TABLE IF EXISTS product_sales;
DROP TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;

Recursos adicionales

  • Última actualización el