Les APIs AUTO CDC : Simplifiez la capture des modifications de données grâce aux pipelines

Les pipelines Lakeflow simplifient la capture des données modifiées (CDC) avec les API AUTO CDC et AUTO CDC FROM SNAPSHOT. Ces API automatisent la complexité du calcul des dimensions à variation lente (SCD) Type 1 et Type 2 à partir d’un flux CDC ou d’instantanés de base de données. L’API AUTO CDC prend également en charge le suivi bitemporel, qui enregistre les modifications dans deux dimensions temporelles (bêta). Pour en savoir plus sur SCD Type 1 et Type 2, consultez Captures de données modifiées et captures instantanées. Pour plus d’informations sur le suivi bitemporal, consultez Bitemporal AUTO CDC.

Note

Les AUTO CDC API remplacent les APPLY CHANGES API et ont la même syntaxe. Les APPLY CHANGES API sont toujours disponibles, mais Databricks recommande d’utiliser les AUTO CDC API à leur place.

L’API que vous utilisez dépend de la source de vos données modifiées :

  • AUTO CDC: utilisez cette option lorsque la base de données source a un flux CDC activé. AUTO CDC traite les modifications d’un flux de données modifiées (CDF). Il est pris en charge dans les interfaces SQL et Python du pipeline.
  • AUTO CDC FROM SNAPSHOT : à utiliser lorsque la CDC n’est pas activée sur la base de données source et que seuls des instantanés sont disponibles. Cette API compare les instantanés pour déterminer les modifications, puis les traite. Il est pris en charge dans les interfaces SQL et Python du pipeline.

Pour les cibles de type SCD 1, on peut aussi combiner un AUTO CDC FROM SNAPSHOT flow avec un ou plusieurs AUTO CDC flows. Utilisez ce schéma lorsqu’une source fournit des instantanés autoritaires périodiques et un flux CDC à faible latence entre les instantanés.

Les deux API prennent en charge la mise à jour des tables à l’aide de SCD Type 1 et de Type 2 :

  • Utilisez SCD Type 1 pour mettre à jour les enregistrements directement. L’historique n’est pas conservé pour les enregistrements mis à jour.
  • Utilisez SCD Type 2 pour conserver un historique des enregistrements, soit sur toutes les mises à jour, soit sur les mises à jour d’un ensemble de colonnes spécifié.

Pour AUTO CDC uniquement, vous pouvez également utiliser le stockage bitemporel, qui étend l’historique de type SCD 2 pour suivre les modifications selon deux dimensions temporelles : le temps métier et le temps système. Bitemporal est en version bêta. Voir Bitemporal AUTO CDC.

AUTO CDC prend également en charge les mises à jour partielles, où un enregistrement de modification met à jour uniquement un sous-ensemble de colonnes. Consultez Appliquer des mises à jour partielles.

Les AUTO CDC API ne sont pas prises en charge par les pipelines déclaratifs Apache Spark.

Pour la syntaxe et d'autres références, consultez AUTO CDC INTO (pipelines), create_auto_cdc_flow et create_auto_cdc_from_snapshot_flow.

Note

Utilisez ces API pour mettre à jour les tables dans vos pipelines en fonction des changements dans les données sources. Pour savoir comment enregistrer et interroger des informations de modification au niveau des lignes pour les tables Delta, consultez Utiliser le flux de données modifiées sur Azure Databricks.

Spécifications

Pour utiliser les API CDC, votre pipeline doit être configuré pour utiliser les pipelines Lakeflow serverless ou l’une desPro Lakeflow pipelines Advanced.

Comment fonctionne AUTO CDC

Pour effectuer le traitement CDC avec AUTO CDC, créez une table de diffusion en continu, puis utilisez l’instruction AUTO CDC ... INTO dans SQL ou la fonction create_auto_cdc_flow() en Python pour spécifier la source, les clés et le séquencement du flux de modification. Pour obtenir une explication sur le fonctionnement du séquencement et de la logique SCD, consultez Modifier la capture et les instantanés de données. Consultez les exemples AUTO CDC.

Pour une hydratation initiale à partir d'une source avec un flux de modification, utilisez AUTO CDC avec un flux once, puis continuez à traiter le flux de modification. Consultez Répliquer une table RDBMS externe à l’aide de l’AUTO CDC.

Pour plus d’informations sur la syntaxe, consultez AUTO CDC INTO (pipelines) ou create_auto_cdc_flow.

Fonctionnement de AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT détermine les modifications apportées aux données sources en comparant les instantanés dans l’ordre. Vous pouvez lire des instantanés à partir d’une table Delta, de fichiers de stockage cloud ou de JDBC directement. Il est pris en charge dans les interfaces SQL et Python du pipeline.

Pour effectuer un traitement CDC avec AUTO CDC FROM SNAPSHOT, créez une table de streaming puis définissez le flux :

  • En SQL, utilisez la forme AUTO CDC ... FROM SNAPSHOT de l’instruction CREATE FLOW ou la clause intégrée FLOW AUTO CDC de CREATE STREAMING TABLE. La source est une clause FROM SNAPSHOT (snapshot_query) requise et une clause WITH VERSION (version_query) facultative qui sélectionne la prochaine version de l'instantané à traiter. Voir CREATE FLOW (pipelines).
  • En Python, utilisez la fonction create_auto_cdc_from_snapshot_flow() pour spécifier l’instantané, les clés et d’autres arguments. Voir create_auto_cdc_from_snapshot_flow.

Pour plus d’informations sur les deux modèles d’ingestion et sur le moment où les utiliser, consultez les modèles de traitement par instantané. Consultez les exemples AUTO CDC FROM SNAPSHOT.

Combinez AUTO CDC FROM SNAPSHOT et AUTO CDC

La version de l’instantané et la colonne de séquencement CDC appartiennent au même domaine d’ordonnancement. Un événement CDC plus récent que le dernier instantané prime sur le snapshot. Un instantané plus récent prime sur les anciens événements CDC et considère les clés absentes du instantané comme supprimées dans la version du snapshot. Les événements CDC plus récents peuvent préserver ou restaurer ces clés.

Une cible unifiée doit répondre aux exigences suivantes :

  • La cible utilise le SCD Type 1.
  • La cible a exactement un AUTO CDC FROM SNAPSHOT flux et un ou plusieurs AUTO CDC flux.
  • Chaque AUTO CDC flux porte un nom unique.
  • Tous les flux utilisent le même nombre de clés, dans le même ordre. Les noms des clés du flux d’instantané sont comparés sans tenir compte de la casse aux noms des clés AUTO CDC. Les flux multiples AUTO CDC doivent utiliser des noms de clés et une casse identiques.
  • La version de snapshot et chaque colonne de séquençage CDC ont exactement le même type de données.
  • Le AUTO CDC FROM SNAPSHOT flow ne définit pas les attentes.
  • Les AUTO CDC flux n’utilisent pas IGNORE NULL UPDATES ni ses variantes de liste de colonnes.
  • Le pipeline utilise le mode déclenché.
  • Si vous ajoutez un AUTO CDC flux à une cible existante AUTO CDC FROM SNAPSHOT , la cible ne doit pas contenir de colonnes utilisateur dont les noms entrent en conflit avec les colonnes système réservées AUTO CDC .

Les flux AUTO CDC et AUTO CDC FROM SNAPSHOT peuvent utiliser l’interface de pipeline Python ou SQL. Vous pouvez combiner des flux SQL et Python dans une même cible.

Utilisez once=True dans Python ou ONCE dans SQL dans le flux AUTO CDC FROM SNAPSHOT pour remplir un instantané une seule fois pendant que les flux AUTO CDC continuent de traiter les nouveaux événements lors des mises à jour suivantes. Une actualisation complète de la cible réexécute le flux d’instantané à exécution unique. Pour un exemple complet, voir Ajouter un remplissage à une table AUTO CDC SCD Type 1.

Ce modèle ne prend pas en compte les cibles SCD Type 2 ni bitemporelles.

Utiliser plusieurs colonnes pour le séquencement

Pour séquencer sur plusieurs colonnes (par exemple, un horodatage et un ID afin de départager les égalités), utilisez une STRUCT pour les combiner. L’API commande d’abord par le premier champ, et en cas d’égalité, considère le deuxième champ, et ainsi de suite.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

sequence_by = struct("timestamp_col", "id_col")

Exemples AUTO de CDC

Les exemples suivants illustrent le traitement SCD Type 1 et Type 2 à l’aide d’une source de flux de données modifiées. L’exemple de données crée de nouveaux enregistrements utilisateur, supprime un enregistrement utilisateur et met à jour les enregistrements utilisateur. Dans l’exemple SCD Type 1, les dernières UPDATE opérations arrivent en retard et sont supprimées de la table cible, ce qui illustre la gestion des événements hors séquence.

Voici les enregistrements d’entrée utilisés dans ces exemples. Ces données sont créées en exécutant la requête dans la section Créer des exemples de données .

userId nom city opération Numéro de séquence
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lys Cancun INSERT 2
123 zéro zéro Supprimer 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Si vous supprimez les marques de commentaire de la ligne finale dans l’exemple de requête de génération de données, celle-ci insère l’enregistrement suivant, qui indique de tronquer la table (vider la table) lorsque sequenceNum=3 :

userId nom city opération Numéro de séquence
zéro zéro zéro TRONQUER 3

Note

Tous les exemples suivants incluent des options pour spécifier à la fois DELETE et TRUNCATE des opérations, mais chacune est facultative.

Créer des exemples de données

Exécutez les instructions suivantes pour créer un exemple de jeu de données. Ce code n’est pas destiné à être exécuté dans le cadre d’une définition de pipeline. Exécutez-le à partir du dossier d’exploration de votre pipeline, plutôt que du dossier des transformations.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Traiter les mises à jour SCD Type 1

SCD Type 1 conserve uniquement la dernière version de chaque enregistrement. L’exemple suivant lit le flux des changements de données créé ci-dessus et applique les changements à une table de streaming cible. Vous avez besoin d’un pipeline pour exécuter ce code. Consultez Développer et déboguer des pipelines ETL avec l’éditeur de pipelines Lakeflow.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Après avoir exécuté l’exemple SCD Type 1, la table cible contient les enregistrements suivants :

userId nom city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lys Cancun

L’utilisateur 123 (Isabel) a été supprimé et n’apparaît pas. L’utilisateur 125 (Mercedes) affiche uniquement la dernière ville (Guadalajara), car SCD Type 1 remplace les valeurs précédentes. La version antérieure UPDATE a été supprimée à sequenceNum=5, car une mise à jour ultérieure a été reçue à sequenceNum=6.

Après avoir exécuté l’exemple avec l’enregistrement TRUNCATE non commenté, la table est vidée à sequenceNum=3. Cela signifie que les enregistrements 124 et 126 ne figurent pas dans la table et que la table cible finale contient uniquement l’enregistrement suivant :

userId nom city
125 Mercedes Guadalajara

Traiter les mises à jour SCD Type 2

SCD Type 2 conserve un historique complet des modifications en créant de nouvelles lignes pour chaque version d’un enregistrement, avec __START_AT et __END_AT des colonnes indiquant quand chaque version était active.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Après avoir exécuté l’exemple SCD Type 2, la table cible contient les enregistrements suivants :

userId nom city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 zéro
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 zéro
126 Lys Cancun 2 zéro

La table conserve l’historique complet. L’utilisateur 123 a deux versions (terminées à la séquence 6 lors de la suppression). L’utilisateur 125 a trois versions montrant les changements de ville. Les enregistrements avec __END_AT = null sont actuellement actifs.

Suivre un sous-ensemble de colonnes avec SCD Type 2

Par défaut, SCD Type 2 crée une nouvelle version chaque fois qu’une valeur de colonne change. Vous pouvez spécifier un sous-ensemble de colonnes à suivre, afin que les modifications apportées à d’autres colonnes mettent à jour la version actuelle en place plutôt que de générer un nouvel enregistrement d’historique.

L’exemple suivant exclut la colonne city du suivi de l’historique :

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Étant donné que les changements de city ne sont pas suivis, les mises à jour de cette colonne écrasent la ligne actuelle au lieu de créer une nouvelle version. La table cible contient les enregistrements suivants :

userId nom city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 zéro
125 Mercedes Guadalajara 2 zéro
126 Lys Cancun 2 zéro

Exemples AUTO CDC FROM SNAPSHOT

Les sections suivantes fournissent des exemples d’utilisation AUTO CDC FROM SNAPSHOT pour traiter des instantanés dans des tables cibles SCD Type 1 ou Type 2. Pour plus d’informations sur l’utilisation de cette API, consultez Capture des modifications de données et instantanés.

Les exemples de cette section incluent des implémentations Python et SQL. Pour une syntaxe SQL complète, voir CREATE FLOW (pipelines).

Exemple : Instantanés périodiques de processus

Utilisez cette approche lorsque les instantanés arrivent régulièrement et dans l’ordre. L’interface Python utilise l’ordre de mise à jour du pipeline pour la gestion des versions. Pour SQL, l’exemple utilise une table de version séparée pour identifier quand un nouvel instantané est disponible.

Vous pouvez lire des captures instantanées à partir de plusieurs types sources, notamment les tables Delta, les fichiers de stockage cloud et les connexions JDBC.

Étape 1 : Créer des exemples de données

Créez une table contenant des données instantanées. Exécutez le code suivant à partir d’un notebook ou de Databricks SQL dans le explorations dossier de votre pipeline :

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

Étape 2 : Exécuter AUTO CDC À PARTIR DU SNAPSHOT

Vous avez besoin d’un pipeline pour exécuter le code à cette étape. Consultez Développer et déboguer des pipelines ETL avec l’éditeur de pipelines Lakeflow.

Pour Python, choisissez un type source pour la vue instantanée (le code de création d’exemple génère une table Delta) :

Option A : Lecture à partir d’une table Delta 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
Option B : Lecture à partir du stockage cloud 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
Option C : Lecture à partir de JDBC (calcul classique uniquement) 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Ensuite, ajoutez le tableau cible et le flux. L’implémentation SQL utilise la snapshot_version table pour identifier lorsqu’un nouvel instantané est disponible :

Python

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

Après la première exécution du pipeline, tous les enregistrements sont insérés en tant que lignes actives :

userId city __START_AT __END_AT
1 Oaxaca 0 zéro
2 Monterrey 0 zéro
3 Tijuana 0 zéro

Note

Pour utiliser SCD Type 1 à la place et ne garder que l’état actuel, définissez stored_as_scd_type=1 en Python ou STORED AS SCD TYPE 1 en SQL. Dans ce cas, la table cible n'inclut pas les colonnes __START_AT et __END_AT.

Étape 3 : Simuler un nouvel instantané et réexécuter

Mettez à jour la table source pour simuler l’arrivée d’un nouvel instantané (exécutez ce code depuis un notebook ou un fichier SQL dans le explorations dossier de votre pipeline) :

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

Exécuter de nouveau le pipeline. AUTO CDC FROM SNAPSHOT compare la nouvelle capture instantanée à la précédente et détecte que l’utilisateur 1 a été supprimé, les utilisateurs 2 et 3 ont été mis à jour, et les utilisateurs 4 et 6 ont été insérés. Cela génère un flux de modification et utilise AUTO CDC pour créer la table de sortie.

Après la deuxième exécution avec SCD Type 2, la table cible contient les enregistrements suivants :

userId city __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 zéro
3 Tijuana 0 1
3 Los Angeles 1 zéro
4 Vallée de la mort 1 zéro
6 Kings Canyon 1 zéro

L’utilisateur 1 a été supprimé. Les utilisateurs 2 et 3 ont chacun deux versions affichant leurs changements de ville. Les utilisateurs 4 et 6 ont été récemment insérés.

Après la deuxième exécution avec SCD Type 1, la table cible affiche uniquement l’état actuel :

userId city
2 Carmel
3 Los Angeles
4 Vallée de la mort
6 Kings Canyon

Exemple : Traiter des instantanés à l’aide de fonctions de version

Utilisez cette approche lorsque vous avez besoin d’un contrôle explicite sur l’ordre des instantanés. Par exemple, utilisez cette approche lorsque plusieurs instantanés arrivent en même temps ou que les instantanés arrivent dans le désordre. Vous définissez comment sélectionner le prochain instantané et son numéro de version. L’API traite les instantanés dans l’ordre de version croissant :

  • Si plusieurs instantanés sont dans le stockage, le sélecteur d’exemples traite les instantanés disponibles dans l’ordre des versions.
  • Si un instantané arrive dans le désordre (par exemple, snapshot_3 après snapshot_4), le sélecteur d’exemple l’exclut car sa version n’est pas supérieure à la dernière version validée.
  • S’il n’y a pas de nouveaux instantanés, la fonction de version ou la requête ne retourne aucun résultat et aucun traitement n’a lieu.

Étape 1 : Préparer les fichiers de snapshots

Créez des fichiers CSV contenant des données d’instantané et ajoutez-les à un volume ou à un emplacement de stockage cloud. Nommez les fichiers par ordre chronologique (par exemple, snapshot_1.csv, snapshot_2.csv).

Chaque fichier doit contenir des colonnes pour userId et city. Par exemple:

snapshot_1.csv :

userId city
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv :

userId city
2 Carmel
3 Los Angeles
4 Vallée de la mort

Étape 2 : Exécuter AUTO CDC FROM SNAPSHOT avec une fonction de version

Dans un pipeline, créez un nouveau fichier Python ou SQL dans le transformations dossier, puis ajoutez le code suivant. Exécutez ensuite le pipeline. Consultez Développer et déboguer des pipelines ETL avec l’éditeur de pipelines Lakeflow.

Python

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Note

Pour utiliser SCD Type 1 à la place, définissez stored_as_scd_type=1 en Python ou STORED AS SCD TYPE 1 en SQL.

Après le traitement snapshot_1.csv, la table cible contient les enregistrements suivants :

userId city __START_AT __END_AT
1 Oaxaca 1 zéro
2 Monterrey 1 zéro
3 Tijuana 1 zéro

Après le traitement snapshot_2.csv, la table cible contient les enregistrements suivants :

userId city __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 zéro
3 Tijuana 1 2
3 Los Angeles 2 zéro
4 Vallée de la mort 2 zéro

Note

N’oubliez pas que, pour SCD Type 1, la table ressemble exactement à la capture instantanée la plus récente. La différence est que les requêtes en aval peuvent utiliser le flux de modification pour traiter uniquement les enregistrements modifiés.

Étape 3 : Ajouter de nouveaux instantanés

Ajoutez un nouveau fichier CSV à l’emplacement de stockage avec des données modifiées (par exemple, des valeurs de ville modifiées, de nouvelles lignes ou des lignes supprimées). Ensuite, réexécutez le pipeline pour traiter le nouvel instantané.

Limites

Ressources supplémentaires

  • Dernière mise à jour :