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

Lakeflow Spark Declarative Pipelines (SDP) simplifie la capture de changements de données (CDC) avec les API AUTO CDC et AUTO CDC FROM SNAPSHOT.

L’interface que vous utilisez dépend de la source des données modifiées :

• Utilisez AUTO CDC pour traiter les modifications d’un flux de données de modifications (CDF).
• Utilisez AUTO CDC FROM SNAPSHOT (préversion publique et disponible uniquement pour Python) pour traiter les modifications apportées aux instantanés de base de données.

Auparavant, l’instruction MERGE INTO était couramment utilisée pour traiter les enregistrements CDC sur Azure Databricks. Toutefois, MERGE INTO il peut produire des résultats incorrects en raison d’enregistrements hors séquence ou nécessite une logique complexe pour réorganiser les enregistrements.

L’API AUTO CDC est prise en charge dans les interfaces SQL et Python pipelline. L’API AUTO CDC FROM SNAPSHOT est prise en charge dans l’interface Python. Les AUTO CDC API ne sont pas prises en charge par les pipelines déclaratifs Apache Spark.

Les deux AUTO CDC et AUTO CDC FROM SNAPSHOT prennent en charge la mise à jour des tables en utilisant le SCD de 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 un ensemble de colonnes spécifié.

Pour obtenir la syntaxe et d'autres références, consultez AUTO CDC pour les pipelines (SQL),AUTO CDC pour les pipelines (Python) et AUTO CDC FROM SNAPSHOT pour les pipelines (Python).

Spécifications

Pour utiliser les API cdc, votre pipeline doit être configuré pour utiliser le SDP serverless ou le SDP Pro ou Advancedles éditions.

Comment s'effectue la mise en œuvre de CDC avec l'API AUTO CDC ?

En gérant automatiquement les enregistrements hors séquence, l’API CDC AUTO garantit le traitement correct des enregistrements CDC et supprime la nécessité de développer une logique complexe pour gérer les enregistrements hors séquence. Vous devez spécifier une colonne dans les données sources sur lesquelles séquencer les enregistrements, que les API interprètent comme une représentation monotonique de l’ordre approprié des données sources. Les pipelines gèrent automatiquement les données qui arrivent dans le désordre. Pour les modifications du type SCD 2, les pipelines propagent les valeurs de séquence correctes vers les colonnes __START_AT et __END_AT de la table cible. Il doit y avoir une mise à jour distincte par clé à chaque valeur de séquencement, et les valeurs de séquencement NULL ne sont pas prises en charge.

Pour effectuer le traitement CDC avec AUTO CDC, vous créez d’abord 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 créer la table de diffusion en continu cible, utilisez l’instruction CREATE OR REFRESH STREAMING TABLE dans SQL ou la create_streaming_table() fonction en Python. Consultez les exemples de traitement SCD de type 1 et de type 2 .

Pour plus d’informations sur la syntaxe, consultez la référence SQL des pipelines ou la référence Python.

Comment le CDC est-il implémenté avec l'API AUTO CDC FROM SNAPSHOT ?

AUTO CDC FROM SNAPSHOT est une API déclarative qui détermine efficacement les modifications apportées aux données sources en comparant une série d’instantanés dans l’ordre, puis exécute le traitement requis pour le traitement CDC des enregistrements dans les instantanés. AUTO CDC FROM SNAPSHOT est pris en charge uniquement par l’interface de pipelines Python.

AUTO CDC FROM SNAPSHOT prend en charge l’importation d’instantanés à partir de plusieurs types de sources :

• Utilisez l’ingestion périodique d’instantanés pour charger des instantanés à partir d’une table ou d’une vue existante. AUTO CDC FROM SNAPSHOT dispose d’une interface simple et simplifiée pour prendre en charge l’ingestion périodique d’instantanés à partir d’un objet de base de données existant. Un nouvel instantané est ingéré avec chaque mise à jour du pipeline et le temps d’ingestion est utilisé comme version de l’instantané. Lorsqu’un pipeline est exécuté en mode continu, plusieurs instantanés sont ingérés avec chaque mise à jour de pipeline sur une période déterminée par le paramètre d’intervalle de déclencheur pour le flux qui contient le AUTO CDC FROM SNAPSHOT traitement.
• Utilisez l’ingestion d’instantanés historiques pour traiter les fichiers contenant des instantanés de base de données, tels que les instantanés générés à partir d’une base de données Oracle ou MySQL ou d’un entrepôt de données.

Pour effectuer le traitement CDC à partir de n’importe quel type de source avec AUTO CDC FROM SNAPSHOT, vous créez d’abord une table de streaming, puis utilisez la fonction create_auto_cdc_from_snapshot_flow() en Python afin de spécifier l’instantané, les clés et d’autres arguments requis pour implémenter le traitement. Voir les exemples d’ingestion d’instantanés périodiques et d’ingestion d’instantanés historiques.

Les instantanés passés à l’API doivent être dans l’ordre croissant par version. Si SDP détecte un instantané hors séquence, une erreur est générée.

Pour plus d’informations sur la syntaxe, consultez la référence Python des pipelines.

Utiliser plusieurs colonnes pour le séquencement

Vous pouvez séquencer par plusieurs colonnes (par exemple, un horodatage et un ID pour rompre les liens), vous pouvez utiliser un STRUCT pour les combiner : il les trie par le premier champ du STRUCT, et en cas de liaison, considère le deuxième champ, et ainsi de suite.

Exemple dans SQL :

SEQUENCE BY STRUCT(timestamp_col, id_col)

Exemple en Python :

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

Limites

La colonne utilisée pour le séquencement doit être un type de données triable.

Exemple : traitement de type SCD 1 et type SCD 2 avec les données source CDF

Les sections suivantes fournissent des exemples de requêtes SCD de type 1 et de type 2 qui mettent à jour les tables cibles en fonction des événements sources à partir d’un flux de données de modification qui :

1. Crée de nouveaux enregistrements utilisateur.
2. Supprime un enregistrement utilisateur.
3. 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 commande.

Les exemples suivants supposent une connaissance de la configuration et de la mise à jour des pipelines. Consultez le tutoriel : Créer un pipeline ETL à l’aide de la capture de données modifiées.

Pour exécuter ces exemples, vous devez commencer par créer un exemple de jeu de données. Consultez Générer des données de test.

Voici les enregistrements d’entrée pour ces exemples :

Si vous supprimez les marques de commentaire de la ligne finale dans les exemples de données, il insère l’enregistrement suivant qui spécifie où les enregistrements doivent être tronqués :

Traiter les mises à jour de type SCD 1

L’exemple suivant illustre le traitement des mises à jour de type SCD 1 :

Python

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

@dp.view
def users():
  return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW flowname AS AUTO CDC INTO
  target
FROM
  stream(cdc_data.users)
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 :

Après avoir exécuté l’exemple SCD type 1 avec l’enregistrement supplémentaire TRUNCATE , les enregistrements 124 et 126 sont tronqués en raison de l’opération TRUNCATE sur sequenceNum=3, et la table cible contient l’enregistrement suivant :

Traiter les mises à jour de type SCD 2

L’exemple suivant illustre le traitement des mises à jour de type SCD 2 :

Python

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

@dp.view
def users():
  return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_flow
AS AUTO CDC INTO
  target
FROM
  stream(cdc_data.users)
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 :

Une requête SCD type 2 peut également spécifier un sous-ensemble de colonnes de sortie à suivre pour l’historique dans la table cible. Les modifications apportées à d’autres colonnes sont actualisées directement au lieu de générer de nouvelles entrées d'historique. L’exemple suivant illustre l’exclusion de la city colonne du suivi :

L’exemple suivant illustre l’utilisation de l’historique de suivi avec le type SCD 2 :

Python

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

@dp.view
def users():
  return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_flow
AS AUTO CDC INTO
  target
FROM
  stream(cdc_data.users)
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)

Après avoir exécuté cet exemple sans l’enregistrement supplémentaire TRUNCATE , la table cible contient les enregistrements suivants :

Générer des données de test

Le code ci-dessous est fourni pour générer un exemple de jeu de données à utiliser dans les exemples de requêtes présents dans ce didacticiel. En supposant que vous disposez des informations d’identification appropriées pour créer un schéma et créer une table, vous pouvez exécuter ces instructions avec un notebook ou Databricks SQL. Le code suivant n’est pas destiné à être exécuté dans le cadre d’une définition de pipeline :

CREATE SCHEMA IF NOT EXISTS cdc_data;

CREATE TABLE
  cdc_data.users
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 above batch at sequenceNum 6 will be the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Exemple : traitement périodique des instantanés

L’exemple suivant illustre le traitement SCD de type 2 qui ingère des instantanés d’une table stockée à l’adresse mycatalog.myschema.mytable. Les résultats du traitement sont écrits dans une table nommée target.

mycatalog.myschema.mytable enregistrements à l'horodatage 2024-01-01 00:00:00

mycatalog.myschema.mytable enregistrements à l'horodatage 2024-01-01 12:00:00

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
 return spark.read.table("mycatalog.myschema.mytable")

dp.create_streaming_table("target")

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

Après avoir traité les instantanés, la table cible contient les enregistrements suivants :

Exemple : traitement d’instantanés historiques

L’exemple suivant illustre le traitement SCD de type 2 qui met à jour une table cible en fonction des événements sources de deux instantanés stockés dans un système de stockage cloud :

Instantané sur timestamp, stocké dans /<PATH>/filename1.csv

Instantané sur timestamp + 5, stocké dans /<PATH>/filename2.csv

L’exemple de code suivant illustre le traitement des mises à jour de type SCD 2 avec ces instantanés :

from pyspark import pipelines as dp

def exist(file_name):
  # Storage system-dependent function that returns true if file_name exists, false otherwise

# This function returns a tuple, where the first value is a DataFrame containing the snapshot
# records to process, and the second value is the snapshot version representing the logical
# order of the snapshot.
# Returns None if no snapshot exists.
def next_snapshot_and_version(latest_snapshot_version):
  latest_snapshot_version = latest_snapshot_version or 0
  next_version = latest_snapshot_version + 1
  file_name = "dir_path/filename_" + next_version + ".csv"
  if (exist(file_name)):
    return (spark.read.load(file_name), next_version)
   else:
     # No snapshot available
     return None

dp.create_streaming_live_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = next_snapshot_and_version,
  keys = ["Key"],
  stored_as_scd_type = 2,
  track_history_column_list = ["TrackingCol"]
)

Après avoir traité les instantanés, la table cible contient les enregistrements suivants :

Ajouter, modifier ou supprimer des données dans une table de diffusion en continu cible

Si votre pipeline publie des tables dans le catalogue Unity, vous pouvez utiliser des instructions DML (Langage de manipulation de données), y compris les instructions insert, update, delete et merge, pour modifier les tables de streaming cibles créées par des instructions .

Étant donné que les pipelines déclaratifs Spark Lakeflow utilisent une colonne spécifiée SEQUENCE BY et propagent les valeurs de séquence appropriées vers les colonnes SEQUENCE BY, __END_AT et de la table cible (pour le type SCD 2), vous devez vous assurer que les instructions DML utilisent des valeurs valides pour ces colonnes afin de conserver l'ordre correct des enregistrements. Découvrez comment la CDC est implémentée avec l’API AUTO CDC.

Pour plus d’informations sur l’utilisation d’instructions DML avec des tables de streaming, consultez Ajouter, modifier ou supprimer des données dans une table de diffusion en continu.

L’exemple suivant insère un enregistrement actif avec une séquence de début de 5 :

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

Lire un flux de données modifiées à partir d’une table cible AUTO CDC

Dans Databricks Runtime 15.2 et versions ultérieures, vous pouvez lire un flux de données modifiées à partir d'une table de streaming qui fait l'objet des requêtes AUTO CDC ou AUTO CDC FROM SNAPSHOT de la même façon que vous lisez un flux de données modifiées à partir d'autres tables Delta. Les éléments suivants sont nécessaires pour lire le flux de données modifiées à partir d’une table de diffusion en continu cible :

• La table de streaming cible doit être publiée sur le catalogue Unity. Consultez Utiliser le catalogue Unity avec des pipelines.
• Pour lire le flux de données de modification de la table de diffusion en continu cible, vous devez utiliser Databricks Runtime 15.2 ou version ultérieure. Pour lire le flux de données de modification dans un autre pipeline, le pipeline doit être configuré pour utiliser Databricks Runtime 15.2 ou version ultérieure.

Vous lisez le flux de données de modification à partir d'une table de diffusion en continu cible qui a été créée dans les pipelines déclaratifs de Spark Lakeflow de la même manière que pour la lecture d'un flux de données modifiées à partir d'autres tables Delta. Pour en savoir plus sur l’utilisation de la fonctionnalité de flux de données modifiées Delta, notamment des exemples en Python et SQL, consultez Utiliser le flux de données modifiées Delta Lake sur Azure Databricks.

Obtenir des données sur les enregistrements traités par une requête CDC dans les pipelines

Les métriques suivantes sont capturées par les requêtes AUTO CDC :

• num_upserted_rows: Le nombre de lignes de sortie fusionnées dans le jeu de données lors d'une mise à jour.
• num_deleted_rows: nombre de lignes de sortie existantes supprimées du jeu de données pendant une mise à jour.

La métrique num_output_rows, générée pour les flux non-CDC, n'est pas capturée lors des requêtes AUTO CDC.

Quels objets de données sont utilisés pour le traitement CDC dans un pipeline ?

Lorsque vous déclarez la table cible dans le metastore Hive, deux structures de données sont créées :

• Vue utilisant le nom attribué à la table cible.
• Table de stockage interne utilisée par le pipeline pour gérer le traitement CDC. Cette table est nommée en prédéfinissant __apply_changes_storage_ le nom de la table cible.

Par exemple, si vous déclarez une table cible nommée dp_cdc_target, vous verrez une vue nommée dp_cdc_target et une table nommée __apply_changes_storage_dp_cdc_target dans le metastore. La création d’une vue permet aux pipelines déclaratifs Spark de Lakeflow de filtrer les informations supplémentaires (par exemple, les tombstones et les versions) requises pour gérer les données désordonnées. Pour afficher les données traitées, interrogez la vue cible. Étant donné que le schéma de la __apply_changes_storage_ table peut changer pour prendre en charge les futures fonctionnalités ou améliorations, vous ne devez pas interroger la table pour une utilisation en production. Si vous ajoutez manuellement des données à la table, les enregistrements sont supposés venir avant d’autres modifications, car les colonnes de version sont manquantes.

Ressources supplémentaires

• Tutoriel : Créer un pipeline ETL à l’aide de la capture de données modifiées