Aller au contenu principal
Dernière mise à jour le

ETL dans Databricks SQL

Lorsque vous traitez de grandes quantités de données, vous avez besoin d'un pipeline capable de traiter uniquement les enregistrements nouveaux et modifiés au lieu de retraiter l'ensemble du dataset. Ceci est appelé 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 ni planifier de refresh manuels.

Ce tutoriel vous guide à travers un modèle courant : le suivi des changements de produits au fil du temps. Vous allez créer une table source, capturer les événements de changement, construire une table de dimension qui conserve l'historique complet de chaque produit, maintenir une table de faits qui écrase de manière incrémentale les enregistrements récents, et ajouter une couche de rapport agrégée par-dessus. Vous utilisez AUTO CDC pour la table de dimension et un flux REPLACE WHERE pour la table de faits.

La fonctionnalité clé de ce tutoriel est AUTO CDC. Dans un warehouse traditionnel, vous écririez des instructions MERGE INTO complexes 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équençage et si vous souhaitez un SCD de type 1 (dernière valeur uniquement) ou un SCD de type 2 (historique complet), et Databricks applique automatiquement la logique de Merge correcte. Pour une vue d'ensemble de la CDC, consultez Les APIs AUTO CDC : Simplifiez la capture des modifications de données avec des pipelines.

À la fin de ce didacticiel, vous aurez :

  1. Table source créée qui suit les modifications à l'aide du flux de données de modification.
  2. J'ai inspecté les données de modification brutes pour comprendre le Stream d'événements CDC.
  3. Utilisé AUTO CDC pour construire une table de dimensions SCD de type 2 à partir de ces événements.
  4. Événements de suppression traités de manière incrémentielle via le pipeline.
  5. A construit une table de faits qui écrase progressivement les lignes récentes avec un flux REPLACE WHERE.
  6. Création d'une vue matérialisée qui maintient de manière incrémentielle un rapport d'agrégation.
  7. Configuré SCHEDULE REFRESH EVERY 1 DAY afin que les modifications se propagent automatiquement dans le pipeline.

Exigences​

Pour suivre ce didacticiel, vous devez satisfaire aux exigences suivantes :

Étape 1 : Configurez votre catalogue et votre schéma​

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

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

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

Créez une table products avec Utiliser le flux de données de modification sur Databricks (CDF) activé. La CDF est une fonctionnalité de Delta Lake qui enregistre chaque insertion, mise à jour et suppression sous la forme d'un journal de modification interrogeable. Ceci est similaire à un Stream CDC provenant d'un système source transactionnel, à l'exception que les modifications sont capturées directement au sein de la table Delta plutôt que depuis un Log externe. Vous utilisez la CDF ici pour générer les événements de modification que le pipeline en aval consommera.

  1. Créer la table et charger les enregistrements initiaux :

    SQL
    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 les changements en amont, y compris les nouveaux produits, un déplacement de warehouse et une réaffectation de catégorie :

    SQL
    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 : query le flux de données de modification​

Avant de construire le pipeline en aval, il est utile d'examiner les événements de modification bruts afin de comprendre ce que AUTO CDC traitera. La fonction table_changes() lit le log CDF et renvoie chaque opération capturée ainsi que les colonnes de métadonnées :

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

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

Notez qu'un seul changement logique (par exemple, le déplacement de Spoon vers un autre warehouse) produit plusieurs événements : une pré-image et une post-image. Dans un warehouse traditionnel, vous écririez une instruction MERGE pour rapprocher tous ces événements dans une table cible, en gérant les insertions, les mises à jour et les suppressions avec une logique distincte, et en vous assurant que les événements sont appliqués dans le bon ordre. C’est exactement la complexité que AUTO CDC élimine à l’étape suivante.

Étape 4 : Créez une dimension SCD de type 2 avec AUTO CDC​

Une table de streaming traite les données de manière incrémentale. À chaque refresh, il ne lit que les nouvelles lignes depuis la dernière exécution, il n'a donc pas besoin de retraiter l'ensemble du dataset. Cela le rend bien adapté aux sources à fort volume ou fréquemment modifiées.

AUTO CDC ajoute le traitement de la capture des données modifiées sur une table de streaming. Au lieu de rédiger 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équençage et laissez Databricks appliquer la logique correcte. AUTO CDC gère également les événements désordonnés automatiquement, ce qui est un problème courant lors de l'utilisation de MERGE INTO pour gérer les événements provenant de systèmes distribués ou les charges batch avec des Timestamp qui se chevauchent.

L'instruction suivante crée une table SCD de type 2 qui préserve l'historique de version complet de chaque produit. Chaque version reçoit __START_AT et __END_AT timestamps. Un NULL dans __END_AT marque la version actuelle.

SQL
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: refresh la table selon un programme quotidien.
  • FLOW AUTO CDC: déclare ceci comme un flux de CDC. Databricks applique automatiquement les sémantiques d'insertion, de mise à jour et de suppression.
  • WHERE _change_type != 'update_preimage': ignore les préimages. La postimage possède déjà la ligne mise à jour.
  • KEYS (product_id): la clé métier. Les événements avec la même clé sont fusionnés dans des lignes versionnées.
  • APPLY AS DELETE WHEN _change_type = 'delete': clôture la version actuelle lorsqu'un événement de suppression se produit. Cela vous permet de définir la condition qui identifie un événement de suppression.
  • SEQUENCE BY _commit_timestamp: établit l'ordre des événements. Gère correctement les arrivées désordonnées.
  • STORED AS SCD TYPE 2: conserve l'historique complet. AUTO CDC prend en charge les types SCD 1 et SCD 2.

Interrogez la table de dimensions :

SQL
SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
  • Spoon : deux versions. Seattle (fermé, __END_AT défini) et Los Angeles (actuel, __END_AT = NULL).
  • Fork : deux versions. Catégorie Couverts (fermée) et Catégorie Repas (actuelle).
  • Serviette et sous-verre : une version chacun (nouvellement insérée, __END_AT = NULL).
  • Tous les autres produits : une version chacun (__END_AT = NULL).

Étape 5 : Traiter les suppressions à travers le pipeline​

Maintenant, simulez deux produits abandonnés en les supprimant de la table source :

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

Ces événements de suppression sont enregistrés dans le log CDF mais la table de streaming ne les a pas encore vus. Refresh la table de streaming pour traiter les nouveaux événements :

SQL
REFRESH STREAMING TABLE products_history;

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

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

Bowl et Glass sont maintenant clos avec le __END_AT défini, les marquant comme abandonnés. Tous les autres produits actuels restent inchangés. La table de streaming a seulement traité les nouveaux événements de suppression sans retraiter les insertions et les mises à jour de la refresh précédente.

Étape 6 : créer une table de faits avec un flux REPLACE WHERE​

La table de dimensions des é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 des événements, tels que les ventes, et vous ne retraitez généralement qu'une fenêtre récente, par exemple, pour corriger les enregistrements arrivés en retard. Pour ce modèle, utilisez un flux REPLACE WHERE incrémentiel, qui remplace 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, consultez Quand utiliser REPLACE WHERE plutôt que les vues matérialisées.

  1. Créez une table product_sales et chargez quelques événements de ventes, certains récents et d'autres plus anciens :

    SQL
    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 une table de faits qui ne recalcule que les Ventes des 7 derniers jours à chaque refresh :

    SQL
    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 refresh, supprime les lignes dans la cible qui correspondent au prédicat, recalculer la query source pour cette même plage et insère les résultats. Les lignes en dehors de la fenêtre de 7 jours ne sont pas modifiées.
    • SCHEDULE REFRESH EVERY 1 DAY: refresh la table selon le même calendrier quotidien que la table de dimensions.
    • BY NAME: obligatoire. Fait correspondre les colonnes par nom au lieu de par position.

    La vente datée de 10 jours en dehors du prédicat, le flux ne la charge pas. Seules les ventes des 7 derniers jours sont conservées.

  3. query la table de faits :

    SQL
    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 valeurs sale_date dépendent du jour où vous exécutez le tutoriel :

    sale_id

    ID du produit

    quantité

    date_vente

    101

    1

    3

    Aujourd'hui

    102

    2

    1

    Aujourd'hui

    103

    4

    5

    Il y a 2 jours

    sale_id

    ID du produit

    quantité

    date_vente

    101

    1

    3

    Aujourd'hui

    102

    2

    1

    Aujourd'hui

    103

    4

    5

    Il y a 2 jours

    La vente 104 ne figure pas dans le résultat car sa date se situe en dehors de la fenêtre de 7 jours.

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

Maintenant que vous disposez d'une table de dimensions qui reste à jour avec les modifications de la source, vous pouvez ajouter une couche de reporting par-dessus.

Une vue matérialisée stocke les résultats de query précalculés sous forme de table physique. Contrairement à une vue normale, qui réexécute la requête chaque fois que vous y lisez, une vue matérialisée conserve les résultats et ne recalcule les lignes affectées par les modifications en amont qu'à chaque refresh. Cela le rend bien adapté aux tableaux de bord et aux rapports où les performances des query sont importantes.

SQL
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 se refresh selon un calendrier quotidien. Combiné avec le même calendrier sur la table de streaming, vous avez maintenant un pipeline à trois étapes où les modifications de la table source sont transmises par la dimension et dans l'agrégat à chaque cycle de refresh. Vous n'avez pas besoin d'effectuer manuellement le refresh de la dimension ou de l'agrégat.

SQL
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, apportez une modification à la table source et enregistrez une vente arrivée en retard pour hier :

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

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

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

  1. products enregistre l'événement de modification dans le flux de données de modification.
  2. products_history traite l'événement et ajoute une nouvelle version pour le Couteau.
  3. products_by_category recalcule uniquement la ligne Couverts affectée.

La table de faits se refresh selon son propre calendrier. Pour prendre en compte les ventes arrivées en retard maintenant au lieu d’attendre la prochaine refresh prévue, refresh manuellement la table de faits :

SQL
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 :

SQL
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;

Nettoyer​

Pour nettoyer les Ressources créées par ce tutoriel, utilisez le SQL suivant :

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;

Ressources supplémentaires​

Sur cette page