CREATE STREAMING TABLE ... FLOW AUTO CDC

S’applique à :case cochée oui Databricks SQL

Utilisez la FLOW AUTO CDC clause avec CREATE STREAMING TABLE pour traiter les enregistrements de capture de données modifiées (CDC) d’une source dans une table de diffusion en continu.

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.

AUTO CDC simplifie la capture de données modifiées en gérant automatiquement les enregistrements hors commande. Vous spécifiez des clés pour identifier les enregistrements, une colonne de séquence pour l’ordre et si vous souhaitez stocker les résultats en tant que SCD type 1 (mises à jour directes) ou SCD type 2 (suivi de l’historique).

Syntaxe

CREATE OR REFRESH STREAMING TABLE table_name
FLOW AUTO CDC
FROM source
KEYS (keys)
[IGNORE NULL UPDATES [ON {columnList | * EXCEPT (exceptColumnList)}]]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[SYSTEM SEQUENCE BY systemOrderByColumn]
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2 | BITEMPORAL}]
[TRACK HISTORY ON {columnList | * EXCEPT (exceptColumnList)}]
[COLUMNS TO UPDATE columnName]

Le comportement par défaut pour INSERT et UPDATE les événements consiste à mettre à jour les événements CDC à partir de la source : mettez à jour les lignes de la table cible qui correspondent aux clés spécifiées ou insérez une nouvelle ligne lorsqu’un enregistrement correspondant n’existe pas dans la table cible. DELETE La gestion des événements peut être spécifiée par la condition APPLY AS DELETE WHEN.

Paramètres

  • source

    Source des données. La source doit être une source de diffusion en continu. Utilisez le mot clé STREAM pour utiliser la sémantique de streaming pour lire à partir de la source. Si la lecture détecte une modification ou une suppression concernant un enregistrement existant, une erreur est générée. Il est plus sûr de lire depuis des sources statiques ou d’ajout uniquement.

    Pour plus d’informations sur la diffusion en continu des données, consultez Transformer des données avec des pipelines.

  • KEYS

    Colonne ou combinaison de colonnes qui identifient de manière unique une ligne dans les données sources. Les valeurs de ces colonnes sont utilisées pour identifier les événements CDC qui s’appliquent à des enregistrements spécifiques dans la table cible.

    Pour définir une combinaison de colonnes, utilisez une liste séparée par des virgules de colonnes.

    Cette clause est requise.

  • IGNORE NULL UPDATES

    Permet d’ingérer des mises à jour contenant un sous-ensemble des colonnes cibles. Lorsqu’un événement CDC correspond à une ligne existante et IGNORE NULL UPDATES est spécifié, les colonnes avec une null valeur conservent leurs valeurs existantes dans la cible. Cela s’applique également aux colonnes imbriquées avec une null valeur.

    Pour les mises à jour partielles, ajoutez une ON clause pour contrôler les colonnes qui ignorent null les valeurs :

    • IGNORE NULL UPDATES ON columnList: seules les colonnes répertoriées conservent leurs valeurs existantes lorsque la valeur entrante est null. Toutes les autres colonnes appliquent des valeurs explicites null .
    • IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList): toutes les colonnes à l’exception de celles répertoriées conservent leurs valeurs existantes lorsque la valeur entrante est null. Les colonnes répertoriées appliquent des valeurs explicites null .

    Pour plus d’informations, consultez Appliquer des mises à jour partielles.

    Cette clause est facultative.

    La valeur par défaut consiste à remplacer les colonnes existantes avec des valeurs null.

  • APPLY AS DELETE WHEN

    Spécifie quand un événement CDC doit être traité comme une DELETE insertion/mise à jour plutôt qu'une.

    Pour les sources SCD de type 2, pour traiter les données hors séquence, la ligne supprimée est temporairement conservée en tant que marqueur de suppression dans la table Delta sous-jacente, et une vue est créée dans le metastore pour filtrer ces marqueurs de suppression. L’intervalle de rétention peut être configuré avec la pipelines.cdc.tombstoneGCThresholdInSecondspropriété de table.

    Cette clause est facultative.

  • APPLY AS TRUNCATE WHEN

    Spécifie quand un événement CDC doit être traité comme une table TRUNCATEcomplète. Étant donné que cette clause supprime toutes les données de la table cible, elle doit être utilisée uniquement pour des cas d'utilisation spécifiques nécessitant cette fonctionnalité.

    La APPLY AS TRUNCATE WHEN clause est prise en charge uniquement pour le type SCD 1. Le type SCD 2 ne prend pas en charge l’opération de troncation.

    Cette clause est facultative.

  • SEQUENCE BY

    Nom de colonne spécifiant l’ordre logique des événements CDC dans les données sources. Le traitement du pipeline utilise ce séquencement pour gérer les événements de modification qui arrivent hors de commande.

    Si plusieurs colonnes sont nécessaires pour l'ordonnancement, utilisez une expression STRUCT : elle ordonnera d'abord par le premier champ de struct, puis par le deuxième champ s'il y a égalité, et ainsi de suite.

    Les colonnes spécifiées doivent être des types de données triables.

    Cette clause est requise.

  • SYSTEM SEQUENCE BY

    Important

    Bitemporal AUTO CDC est en version bêta.

    Nom de colonne spécifiant l’heure système à laquelle chaque événement CDC est connu du système. Utilisé pour STORED AS BITEMPORAL suivre les modifications à la fois dans le temps d’activité (SEQUENCE BY) et l’heure système. Voir Bitemporal AUTO CDC.

    Les colonnes spécifiées doivent être des types de données triables.

    Cette clause est facultative et s’applique uniquement aux tables bitemporales.

  • COLUMNS

    Spécifie un sous-ensemble de colonnes à inclure dans la table cible. Vous pouvez :

    • Spécifiez la liste complète des colonnes à inclure : COLUMNS (userId, name, city).
    • Spécifiez une liste de colonnes à exclure : COLUMNS * EXCEPT (operation, sequenceNum)

    Cette clause est facultative.

    La valeur par défaut consiste à inclure toutes les colonnes de la table cible lorsque la COLUMNS clause n’est pas spécifiée.

  • STORED AS

    Indique s’il faut stocker des enregistrements en tant que type SCD 1, SCD type 2 ou bitemporal.

    Définissez cette option pour BITEMPORAL suivre les modifications à la fois dans le temps d’activité et l’heure système. Bitemporal nécessite SYSTEM SEQUENCE BY et est en bêta. Voir Bitemporal AUTO CDC.

    Cette clause est facultative.

    La valeur par défaut est la méthode SCD de type 1.

  • TRACK HISTORY ON

    Spécifie un sous-ensemble de colonnes de sortie pour générer des enregistrements d’historique lorsqu’il existe des modifications apportées à ces colonnes spécifiées. Vous pouvez :

    • Spécifiez la liste complète des colonnes à suivre : COLUMNS (userId, name, city).
    • Spécifiez une liste de colonnes à exclure du suivi : COLUMNS * EXCEPT (operation, sequenceNum)

    Cette clause est facultative. La valeur par défaut consiste à suivre l’historique de toutes les colonnes de sortie lorsqu’il existe des modifications, équivalentes à TRACK HISTORY ON *.

  • COLUMNS TO UPDATE

    Spécifie le nom d’une colonne source qui contient, pour chaque enregistrement de modification, l’ensemble de colonnes à mettre à jour en tant que tableau de chaînes de nom de colonne (array<string>). Les colonnes qui ne figurent pas dans le tableau conservent leurs valeurs cibles existantes, tandis que les colonnes répertoriées sont écrites à partir de la source, y compris les valeurs explicites null .

    Utilisez cette clause pour les mises à jour partielles lorsque chaque enregistrement de modification met à jour un ensemble différent de colonnes et que vous devez appliquer des valeurs explicites null .

    Vous ne pouvez pas utiliser COLUMNS TO UPDATE avec IGNORE NULL UPDATES, et il n’est pas pris en charge pour les tables bitemporales.

    Pour plus d’informations, consultez Appliquer des mises à jour partielles.

    Cette clause est facultative.

Exemples

-- SCD type 1: apply CDC changes with direct updates (no history)
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

-- SCD type 2: retain a history of changes, with delete handling
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  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;

-- SCD type 2 with history tracking on specific columns
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  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);
  • Last updated on