ETL dans Databricks SQL

Lorsque vous traitez de grandes quantités de données, vous avez besoin d’un pipeline qui ne peut traiter que les enregistrements nouveaux et modifiés au lieu de retraiter l’ensemble du jeu de données. C’est ce qu’on appelle l’ETL incrémentiel. Dans Databricks SQL, vous pouvez créer des pipelines ETL incrémentiels à l’aide de tables de streaming et de vues matérialisées, sans écrire de code procédural ou planifier des actualisations manuelles.

Ce tutoriel vous guide dans un modèle courant : suivi des modifications de produit au fil du temps. Vous créez une table source, capturez les événements de changement, construisez une table de dimension qui préserve l’historique complet de chaque produit, maintenez une table de faits qui écrase progressivement les enregistrements récents, et ajoutez une couche de reporting agrégé par-dessus. Vous utilisez AUTO CDC pour la table des dimensions et un flux REPLACE WHERE pour la table des faits.

La fonctionnalité clé de ce didacticiel est AUTO CDC. Dans un entrepôt traditionnel, vous écrivez des instructions complexes MERGE INTO pour rapprocher les événements d’insertion, de mise à jour et de suppression dans une table cible. Cette approche est sujette aux erreurs, en particulier lorsque les événements arrivent dans le désordre. AUTO CDC gère cela pour vous. Vous déclarez la clé métier, la colonne de séquencement et si vous souhaitez que SCD Type 1 (valeur la plus récente uniquement) ou SCD Type 2 (historique complet) et Azure Databricks applique automatiquement la logique de fusion correcte. Pour obtenir une vue d’ensemble de la capture de données modifiées, consultez la section API AUTO CDC : simplifier la capture de données modifiées avec des pipelines.

À la fin de ce tutoriel, vous aurez :

  1. Création d'une table source qui suit les modifications avec le flux de données de changement.
  2. Inspectez les données modifiées brutes pour comprendre le flux d’événements CDC.
  3. Utilisé AUTO CDC pour générer une table de dimension SCD Type 2 à partir de ces événements.
  4. Traitement des événements de suppression de manière incrémentielle via le pipeline.
  5. J’ai construit une table de faits qui écrase de façon incrémentielle les lignes récentes avec un flux REPLACE WHERE.
  6. Une vue matérialisée a été créée qui gère de façon incrémentielle un rapport d’agrégation.
  7. Configuré SCHEDULE REFRESH EVERY 1 DAY de sorte que les modifications se propagent automatiquement via le pipeline.

Exigences

Pour suivre ce tutoriel, vous devez répondre aux exigences suivantes :

Étape 1 : Configurer votre catalogue et votre schéma

Ouvrez l’éditeur Databricks SQL et définissez votre catalogue de travail et votre schéma. Vous devez disposer de l’autorisation de USE le catalogue et le schéma que vous sélectionnez :

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

Étape 2 : Créer une table source et charger des données

Créez une products table avec Utiliser le flux de données de modification sur Azure Databricks (CDF) activé. CDF est une fonctionnalité Delta Lake qui enregistre chaque insertion, mise à jour et suppression en tant que journal des modifications interrogeable. Ceci est similaire à un flux CDC à partir du système source transactionnel, à la différence que les modifications sont capturées directement dans la table Delta plutôt qu’à partir d’un journal externe. Vous utilisez CDF ici pour générer les événements de modification que le pipeline en aval consommera.

  1. Créez la table et chargez les enregistrements initiaux :

    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. Simuler des modifications en amont, notamment de nouveaux produits, un déplacement d’entrepôt et une réaffectation de catégorie :

    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;

Étape 3 : Interroger le flux de données modifiées

Avant de construire le pipeline en aval, il est utile d’examiner les événements de changement bruts afin de comprendre ce que AUTO CDC va traiter. La table_changes() fonction lit le journal CDF et retourne chaque opération capturée avec les colonnes de métadonnées :

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

Par exemple, la Cuillère a trois événements : un insert (Seattle), un update_preimage (Seattle) et un update_postimage (Los Angeles).

Notez qu’une seule modification logique (par exemple, le déplacement de la cuillère vers un autre entrepôt) produit plusieurs événements : une préimage et une postimage. Dans un entrepôt traditionnel, vous devez écrire une MERGE instruction pour rapprocher tous ces événements dans une table cible, gérer les insertions, les mises à jour et les suppressions avec une logique distincte et vérifier que les événements sont appliqués dans l’ordre correct. Il s’agit exactement de la complexité qui AUTO CDC élimine à l’étape suivante.

Étape 4 : Générer une dimension SCD Type 2 avec AUTO CDC

Une table de diffusion en continu traite les données de manière incrémentielle. À chaque actualisation, elle lit uniquement les nouvelles lignes depuis la dernière exécution. Il n’est donc pas nécessaire de retraiter le jeu de données complet. Cela permet de s’adapter parfaitement à un volume élevé ou à des sources fréquemment changeantes.

AUTO CDC ajoute le traitement de capture de données modifiées au-dessus d’une table de diffusion en continu. Au lieu d’écrire une instruction MERGE INTO qui gère manuellement les insertions, les mises à jour et les suppressions, vous déclarez la clé métier et la colonne de séquencement et laissez Azure Databricks appliquer la logique appropriée. AUTO CDC gère également automatiquement les événements hors commande, ce qui est un problème courant lors de l’utilisation MERGE INTO d’événements arrivant à partir de systèmes distribués ou de chargements par lots avec des horodatages qui se chevauchent.

L’instruction suivante crée une table SCD Type 2 qui conserve l’historique de version complet de chaque produit. Chaque version reçoit __START_AT et __END_AT horodatages. C'est une NULL dans __END_AT qui marque la version actuelle.

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: actualise la table selon une planification quotidienne.
  • FLOW AUTO CDC : déclare ceci en tant que flux CDC. Azure Databricks applique automatiquement la sémantique d’insertion, de mise à jour et de suppression.
  • WHERE _change_type != 'update_preimage': ignore les préimages. postimage contient déjà la ligne mise à jour.
  • KEYS (product_id): clé d’entreprise. Les événements avec la même clé sont fusionnés dans des lignes versionnées.
  • APPLY AS DELETE WHEN _change_type = 'delete': ferme la version actuelle lorsqu’un événement delete arrive. Cela vous permet de définir la condition qui identifie un événement delete.
  • SEQUENCE BY _commit_timestamp: établit l’ordre des événements. Gère correctement les arrivées hors commande.
  • STORED AS SCD TYPE 2: conserve l’historique complet. AUTO CDC prend en charge les types SCD 1 et SCD Type 2.

Interrogez la table de dimension :

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
  • Cuillère : deux versions. Seattle (fermé, __END_AT défini) et Los Angeles (actuel, __END_AT = NULL).
  • Fourche : deux versions. Catégorie Cutlery (fermée) et catégorie Dining (actuelle).
  • Napkin et Coaster : une version chacune (nouvellement insérée, __END_AT = NULL).
  • Tous les autres produits : une version chacune (__END_AT = NULL).

Étape 5 : Traiter les opérations de suppression via le pipeline

À présent, simulez deux produits supprimés en les supprimant de la table source :

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

Ces événements de suppression sont enregistrés dans le journal CDF, mais la table de diffusion en continu ne les a pas encore vus. Actualisez la table de diffusion en continu pour traiter les nouveaux événements :

REFRESH STREAMING TABLE products_history;

Interrogez la table de dimension pour vérifier que les suppressions ont été appliquées :

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

Bowl et Glass sont maintenant fermés avec __END_AT défini, les marquant comme abandonnés. Tous les autres produits actuels restent inchangés. La table de diffusion en continu a traité uniquement les nouveaux événements de suppression sans retraiter les insertions et les mises à jour de l’actualisation précédente.

Étape 6 : Construisez une table de faits avec un flux REPLACE WHERE

La table des dimensions dans les étapes précédentes utilise AUTO CDC car elle suit l’état actuel et historique de chaque produit. Les tables de faits ont des besoins différents. Une table de faits enregistre les événements, comme les ventes, et vous ne retraitez généralement qu’une fenêtre récente, par exemple, pour corriger les enregistrements arrivant en retard. Pour ce motif, utilisez un flux REPLACE WHERE incrémental, qui écrase uniquement les lignes correspondant à un prédicat et laisse toutes les autres lignes intactes. Pour une comparaison complète de cette construction avec les vues matérialisées, voir Quand utiliser REPLACE WHERE vs. vues matérialisées.

  1. Créez un product_sales tableau et chargez quelques événements de vente, certains récents et d’autres plus anciens :

    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. Créez un tableau de faits qui ne recalcule que les 7 derniers jours de ventes à chaque rafraîchissement :

    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): à chaque rafraîchissement, supprime les lignes de la cible correspondant au prédicat, recalcule la requête source pour cette même plage, et insère les résultats. Les rangées en dehors de la fenêtre de 7 jours restent intactes.
    • SCHEDULE REFRESH EVERY 1 DAY: actualise la table sur le même horaire quotidien que la table des dimensions.
    • BY NAME: obligatoire. Fait correspondre les colonnes par nom plutôt que par position.

    La vente datée d’il y a 10 jours sort du filtre, donc le flux ne la charge pas. Seules les ventes des 7 derniers jours sont maintenues.

  3. Interrogez la table de faits :

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

    Le résultat contient les trois ventes des 7 derniers jours. Les sale_date valeurs dépendent du jour où vous lancez le tutoriel :

    sale_id product_id quantité date_de_vente
    101 1 3 Aujourd’hui
    102 2 1 Aujourd’hui
    103 4 5 il y a 2 jours

    La vente 104 n’apparaît pas dans le résultat car sa date est en dehors de la fenêtre de 7 jours.

Étape 7 : Créer une vue agrégée matérialisée

Maintenant que vous disposez d’une table de dimension qui reste à jour avec les modifications source, vous pouvez ajouter une couche de création de rapports par-dessus.

Une vue matérialisée stocke les résultats de requête précalcalisés sous forme de table physique. Contrairement à une vue régulière, qui réexécute la requête chaque fois que vous y lisez, une vue matérialisée conserve les résultats et recompute uniquement les lignes affectées par les modifications en amont sur chaque actualisation. Cela permet de s’adapter parfaitement aux tableaux de bord et aux rapports où les performances des requêtes sont importantes.

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 signifie que cette vue est actualisée selon une planification quotidienne. Combiné avec la même planification sur la table de diffusion en continu, vous disposez maintenant d’un pipeline en trois étapes où les modifications apportées à la table source se produisent en cascade dans la dimension et dans l’agrégat sur chaque cycle d’actualisation. Vous n’avez pas besoin de rafraîchir manuellement la dimension ou l’agrégat.

SELECT * FROM products_by_category ORDER BY active_products DESC;

Étape 8 : Vérifier la cascade de bout en bout

Pour vérifier la cascade complète du pipeline, modifiez la table source et enregistrez une vente arrivée tardive d’hier :

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

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

Le couteau passe de Denver à Seattle. Cela UPDATE déclenche la cascade complète du pipeline, démontrant comment les trois étapes fonctionnent ensemble :

  1. products enregistre l’événement de changement dans le flux de données de changement.
  2. products_history traite l’événement et ajoute une nouvelle version pour Knife.
  3. products_by_category recalcule uniquement la ligne Cutlery affectée.

La table de faits se rafraîchit selon son propre calendrier. Pour récupérer la vente arrivée en retard dès maintenant au lieu d’attendre la prochaine mise à jour programmée, actualisez manuellement le tableau des faits :

REFRESH STREAMING TABLE product_sales_recent;

Le flux recalcule les 7 derniers jours à partir de product_sales, donc la vente 105 apparaît maintenant dans product_sales_recent.

Vérifier:

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;

Nettoyage

Pour nettoyer les ressources créées par ce didacticiel, utilisez le code SQL suivant :

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;

Ressources additionnelles

  • Dernière mise à jour :