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 modifications de produits au fil du temps. Vous créez une table source, capturez les événements de modification, créez une table de dimension qui préserve l'historique complet de chaque produit, et ajoutez une couche de rapport agrégé par-dessus.

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. Création d'une vue matérialisée qui maintient de manière incrémentielle un rapport d'agrégation.
  6. 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​

info

Bêta

AUTO CDC est en Bêta. Nécessite Databricks Runtime 17.3 ou une version ultérieure.

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 STREAM products WITH (readChangeFeed = true)
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.
  • 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 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 refresh selon un planning quotidien. Combiné au même planning sur la table de streaming, vous disposez maintenant d'un pipeline à trois étapes où les modifications apportées à la table source se répercutent à travers la dimension et dans l'agrégat à chaque cycle de refresh. Il n'y a pas de refresh manuel à exécuter.

SQL
SELECT * FROM products_by_category ORDER BY active_products DESC;

Étape 7 : Vérifiez la cascade de bout en bout.​

Pour vérifier la cascade complète du pipeline, apportez une modification à la table source :

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

The Knife se déplace de Denver à Seattle. Ce changement DML unique Trigger la cascade complète du pipeline, démontrant comment les trois étapes fonctionnent ensemble :

  1. products enregistre l'événement de modification via CDF.
  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.

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;

Nettoyer​

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

SQL
DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP STREAMING TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;

Ressources supplémentaires​

Sur cette page