Rubriques avancées sur l'AUTO CDC

Au-delà des API de base AUTO CDC et AUTO CDC FROM SNAPSHOT, vous pouvez exécuter des instructions DML sur des tables cibles, lire des flux de données de changement à partir de cibles CDC, surveiller les métriques de traitement, appliquer des mises à jour partielles et suivre les modifications grâce au stockage bitemporel. Pour une présentation des API AUTO CDC, consultez Les API AUTO CDC : simplifier la capture des données modifiées grâce à des pipelines.

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 le pipeline utilise la colonne spécifiée SEQUENCE BY et propage des valeurs de séquence appropriées aux colonnes __START_AT et __END_AT de la table cible (pour le SCD de type 2), vous devez vous assurer que les instructions DML utilisent des valeurs valides pour ces colonnes afin de préserver l’ordre correct des enregistrements. Découvrez comment fonctionne 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);

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

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 d’une table cible de streaming créée dans un pipeline Lakeflow de la même manière que vous lisez un flux de données de modification provenant 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 dans Python et SQL, consultez Utiliser le flux de données modifiées sur Azure Databricks.

Lire un flux de données de changement à partir d’une vue matérialisée

Vous pouvez lire un flux de données de changement à partir d’une vue matérialisée créée dans un pipeline Lakeflow ou dans Databricks SQL. Utilisez cela pour reproduire les modifications matérialisées de la vue vers des destinations en dehors d’Azure Databricks, ou pour conserver un historique des modifications matérialisées de la vue pour l’audit et le rapport.

Les vues matérialisées utilisent le flux automatique de données de changement, donc vous n’activez pas le flux de données de changement lui-même. À la place, vous activez le flux de données de modification sur chaque vue matérialisée sur laquelle vous en avez besoin en répondant aux exigences suivantes. Consultez le flux de données de modification automatique.

• Pour lire le flux de données de modifications, vous devez utiliser Databricks Runtime 18 LTS ou une version ultérieure, avec le calcul classique, le calcul serverless ou Databricks SQL.

• La vue matérialisée, le pipeline qui la crée, ou le pipeline qui la lit doit utiliser le canal PREVIEW.

• La vue matérialisée doit avoir le suivi des lignes activé. Les vues matérialisées utilisant un calcul serverless ont le suivi des lignes activé par défaut. Consultez le suivi des lignes dans Azure Databricks. Pour vérifier si le suivi des lignes est activé sur une vue matérialisée, exécutez :

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• Pour lire le flux de données de modification à partir d’une vue matérialisée, activez l’indicateur de métadonnées externes sur le pipeline ou sur la vue matérialisée. Pour les instructions, voir Comment activer l’accès à un jeu de données.

Vous lisez le flux de données de changement depuis une vue matérialisée de la même manière que dans d’autres tables Delta, en utilisant la table_changes() fonction, une lecture en flux ou l’option readChangeFeed . Pour la syntaxe et des exemples en SQL et Python, voir Utiliser le flux de données de changement sur Azure Databricks.

Vous pouvez lire un flux de données de changement de vue matérialisé depuis l’intérieur d’une vue matérialisée SQL de Databricks ou d’une table de streaming :

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)

Limitations

En plus des limitations du flux automatique de données de changement, les éléments suivants s’appliquent lorsque vous lisez un flux de données de changement depuis une vue matérialisée :

• Le flux de données de changement inclut les lignes inchangées lorsque la vue matérialisée est entièrement réécrite, et il ne consolide pas plusieurs mises à jour sur la même ligne en un seul événement. Pour filtrer ces éléments, agrégez le flux de données de changement en regroupant toutes les colonnes pour trouver les inserts et suppressions qui partagent les mêmes valeurs de lignes.
• Seul Azure Databricks peut interroger le flux de données de changement pour obtenir une vue matérialisée. Les clients externes de Delta Lake et Iceberg ne le peuvent pas.
• Dans les pipelines Lakeflow, vous pouvez lire le flux de données de changement d’une vue matérialisée uniquement à partir d’un autre pipeline, et ce pipeline doit utiliser le canal PREVIEW. La lecture du flux de données de modification d’une vue matérialisée dans le même pipeline qui la crée n’est pas prise en charge.
• Vous ne pouvez pas créer un index de recherche vectorielle à partir d’une vue matérialisée.

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.

Appliquer des mises à jour partielles

Lorsqu’une source envoie uniquement les colonnes qui ont changé, AUTO CDC doit faire la distinction entre une colonne absente d’un enregistrement de modification, qui doit laisser la valeur cible inchangée et une colonne définie nullexplicitement sur , qui doit remplacer la valeur cible par null. Par défaut, IGNORE NULL UPDATES traite chaque null comme un marqueur « ne pas mettre à jour », il ne peut donc pas appliquer un null explicite. Pour résoudre cette ambiguïté, choisissez l’une des trois méthodes suivantes :

COLUMNS TO UPDATE ne peut pas être combiné avec IGNORE NULL UPDATES, et n’est pas pris en charge pour les tables bitemporales.

En règle générale, choisissez COLUMNS TO UPDATE quand le producteur sait quelles colonnes ont changé dans chaque enregistrement et peuvent porter ces informations dans une colonne source, par exemple lorsque plusieurs producteurs écrivent dans la même source ou que l’ensemble de colonnes pouvant être mises à jour augmente au fil du temps. Choisissez IGNORE NULL UPDATES ON quand le propriétaire du pipeline connaît l’ensemble fixe de colonnes pouvant être mises à jour à l’avance et préfère les contrôler dans le code du pipeline.

L’exemple suivant utilise une colonne source nommée columnsToUpdate pour contrôler les colonnes que chaque enregistrement de modification met à jour, y compris les colonnes définies explicitement sur null :

Python

from pyspark import pipelines as dp

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "sequenceNum",
  stored_as_scd_type = 1,
  columns_to_update = "columnsToUpdate"
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  target
FROM
  stream(cdc_source)
KEYS
  (id)
SEQUENCE BY
  sequenceNum
STORED AS
  SCD TYPE 1
COLUMNS TO UPDATE
  columnsToUpdate;

Pour obtenir la référence complète des paramètres, consultez AUTO CDC INTO (pipelines) et create_auto_cdc_flow.

Bitemporal AUTO CDC

Le type SCD 1 et le type 2 sont unitemporals : ils effectuent le suivi des modifications dans une dimension de temps unique. Bitemporal étend l’historique SCD Type 2 pour suivre les modifications sur deux dimensions temporelles et faire la distinction entre deux perspectives :

• Temps d’affaires : quand l’événement s’est réellement produit.
• Heure système : lorsque le système a enregistré ou ingéré l’événement.

Comme SCD Type 2, bitemporal conserve un historique complet des enregistrements. Il ajoute une deuxième chronologie pour vous permettre de reconstruire à la fois ce que les données ont montrées et ce que le système croyait à n’importe quel moment dans le passé.

Par exemple, un fonds spéculatif ingère des données boursières depuis un système source. Le cours des actions d’Acme Corp change le 1er janvier, mais le fonds n’ingère pas cette mise à jour jusqu’au 5 janvier. Bitemporal AUTO CDC permet au fonds de répondre à deux questions distinctes : ce que le cours réel des actions d’Acme Corp était le 1er janvier (temps d’affaires) et quel prix le système croyait lorsque le fonds a pris des décisions commerciales le 3 janvier (heure système). La possibilité de faire la distinction entre ces délais est utile pour l’audit, les rapports réglementaires et la prise de décision financière.

Pour activer le traitement bitemporel, définissez STORED AS BITEMPORAL (SQL) ou stored_as_scd_type="bitemporal" (Python), utilisez SEQUENCE BY pour la colonne de temps métier, et utilisez SYSTEM SEQUENCE BY pour la colonne de temps système. La table cible ajoute les colonnes __SYSTEM_START_AT et __SYSTEM_END_AT, en plus des colonnes __START_AT et __END_AT de type SCD 2. Pour plus d’informations sur la syntaxe, consultez AUTO CDC INTO (pipelines) ou create_auto_cdc_flow.

Exemples AUTO CDC bitemporels

L’exemple suivant crée une table cible bitemporelle à partir d’un petit ensemble d’événements CDC synthétiques. La bt colonne a l’heure d’activité et la st colonne a l’heure système.

Python

from pyspark import pipelines as dp

# Source: synthetic CDC events
dp.create_streaming_table(name="cdc_source")

@dp.append_flow(target="cdc_source", once=True)
def load_cdc_source():
  return spark.createDataFrame(
    [
      (1, "x10", "y10", 10, 100),
      (1, "x20", "y20", 20, 200)
    ],
    schema="id INT, x STRING, y STRING, bt INT, st INT",
  )

# Target: bitemporal table
dp.create_streaming_table(name="target_bitemporal")

dp.create_auto_cdc_flow(
  target = "target_bitemporal",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "bt",
  system_sequence_by = "st",
  stored_as_scd_type = "bitemporal"
)

SQL

-- Source: synthetic CDC events
CREATE OR REFRESH STREAMING TABLE cdc_source_sql;

CREATE FLOW cdc_source_sql AS INSERT INTO ONCE
  cdc_source_sql BY NAME
SELECT * FROM VALUES
  (1, 'x10', 'y10', 10, 100),
  (1, 'x20', 'y20', 20, 200)
  AS t(id, x, y, bt, st);

-- Target: bitemporal table
CREATE OR REFRESH STREAMING TABLE target_bitemporal_sql;

CREATE FLOW target_bitemporal_sql AS AUTO CDC INTO
  target_bitemporal_sql
FROM
  stream(cdc_source_sql)
KEYS
  (id)
SEQUENCE BY
  bt
SYSTEM SEQUENCE BY
  st
STORED AS
  BITEMPORAL;

La séquence de modifications suivante montre comment une table bitemporelle enregistre une insertion, une mise à jour, une mise à jour non chronologique et une suppression pour une entreprise unique. La colonne de séquencement génère les __START_AT colonnes et __END_AT (temps d’activité) et la colonne de séquencement système génère les __SYSTEM_START_AT colonnes et __SYSTEM_END_AT (heure système) :

Le système gère les événements qui arrivent dans n’importe quel ordre dans les deux chronologies. Lorsqu’un événement arrive avec une date métier ou une date système antérieure à celles des événements déjà traités, le système corrige l’historique concerné plutôt que de simplement ajouter les données en fin de séquence.

Modification 1 : Insertion

La société A est ajoutée le 18/07/2025 à 10:01:00 (heure métier), mais n’est pas ingérée avant 10:05:00 (heure système).

Entrée :

Output:

XFv1 est valide à partir de 10:01:00 sans fin connue. Le système a eu connaissance de ce fait à l’heure système de 10:05:00, sans heure de fin connue.

Modification 2 : Mise à jour

La société A est mise à jour le 18/07/2025 à 12:15:43 (heure métier), et le système traite l’événement à 12:20:00 (heure système). Le système conserve à la fois ce qu’il croyait avant que la mise à jour ait été connue et l’historique des activités corrigé après l’ingestion de la mise à jour.

Entrée :

Output:

XFv1 était considéré valide de 10:01:00 sans fin connue, et le système a maintenu cette croyance de 10:05:00 jusqu’à 12:20:00. XFv1 n’est désormais considéré comme valide que jusqu’à 12:15:43 ; un historique corrigé prend effet à partir de l’heure système 12:20:00, sans heure de fin connue. XFv2 est valide à partir de 12:15:43, sans fin connue, et a été découvert à l’heure du système 12:20:00.

Modification 3 : Mise à jour hors commande

Une mise à jour reçue dans le désordre indique que la société A a en réalité été mise à jour le 18/07/2025 à 12:05:00 (heure métier), mais qu’elle n’est ingérée qu’à 12:25:00 (heure système). Lorsqu’une mise à jour arrive tardivement en temps système, mais avec un temps métier antérieur, le système corrige l’historique métier et conserve à la fois l’état qu’il avait initialement interprété avant l’événement reçu dans le désordre, ainsi que l’historique corrigé.

Entrée :

Output:

XFv1 a été considéré valide de 10:01:00 à 12:15:43, et cette croyance est maintenant valide dans le temps système jusqu’à 12:25:00. La nouvelle mise à jour corrige la validité métier de XFv1 en la fixant à 12:05:00, avec un historique corrigé prenant effet à compter de l’heure système 12:25:00. XFv3 est désormais considéré comme valide de 12:05:00 à 12:15:43, cette information étant valable dans le temps système à partir de 12:25:00, sans fin connue.

Modification 4 : Supprimer

La société A est supprimée le 18/07/2025 à 12:30:00, et le système traite l’événement à 12:30:00. Étant donné qu’une opération de suppression représente la fin de l’existence métier de l’entité, le système ne crée aucune ligne de remplacement. XFv2 apparaît sur deux lignes, conservant une piste d’audit complète à la fois du moment où l’entreprise a cessé d’exister et de celui où le système a eu connaissance de la suppression.

Entrée :

Output:

XFv2 était valide de 12:15:43 sans fin connue, et le système a tenu cette croyance de 12:20:00 à 12:30:00. Une fois la suppression ingérée, XFv2 n’est considéré comme valide que jusqu’à 12:30:00, avec un historique corrigé prenant effet à l’heure système 12:30:00.

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 voyez une vue nommée dp_cdc_target et une table nommée __apply_changes_storage_dp_cdc_target dans le metastore. Interrogez la vue pour accéder aux données traitées. Ne modifiez pas directement la table de support.