Aller au contenu principal
Dernière mise à jour le

Utiliser le flux de données de modification sur Databricks

Le flux de données de modification (CDF) assure le suivi des modifications au niveau des lignes entre les versions d'une table Delta Lake ou d'une table Apache Iceberg v3.

Databricks prend en charge deux approches :

  • Flux de données de modification automatique : Calcule les modifications pendant les lectures de table à l’aide de métadonnées de lignage de ligne. Cela ne nécessite pas de configuration de table individuelle et fonctionne sur les tables Delta Lake et Apache Iceberg v3. Consultez l'option Flux de données de modification automatique.
  • **Flux de données de modification hérité** : matérialise les modifications pendant les écritures de table. Prend uniquement en charge les tables Delta Lake. Nécessite une configuration de table individuelle. Consultez le flux de données de modification hérité pour Delta Lake.

Vous pouvez utiliser le flux de données de modification pour les cas d'utilisation courants d'utilisation des données, y compris :

  • Pipelines ETL incrémentiels qui traitent uniquement les lignes qui ont changé depuis la dernière exécution du pipeline.
  • Pistes d'audit qui suivent les modifications de données pour les exigences de conformité et de gouvernance.
  • Charges de travail de réplication de données qui synchronisent les modifications avec les tables en aval, les caches ou les systèmes externes.

Flux de données de modification automatique​

info

Aperçu

Cette fonctionnalité est en Aperçu public. Les administrateurs du Workspace peuvent contrôler l'accès à cette fonctionnalité à partir de la page Previews . Consultez Gérer les aperçus Databricks.

Le flux de données de modification automatique calcule les modifications au niveau des lignes au moment de la query, plutôt qu'au moment de l'écriture, en utilisant le suivi des lignes pour Delta Lake et la lignée des lignes pour Apache Iceberg v3. Contrairement au flux de données de modification hérité, le flux de données de modification automatique ne nécessite pas de configuration de table individuelle et fonctionne sur les tables Delta Lake et Apache Iceberg v3.

Étant donné que les modifications ne sont pas calculées à chaque écriture pour les opérations MERGE INTO et UPDATE, le flux de données de modification automatique améliore les performances d'écriture et réduit les coûts de stockage, par rapport au flux de données de modification hérité.

Le flux de données de modification automatique utilise les mêmes API table_changes() et readChangeFeed que le flux de données de modification hérité et fonctionne avec les requêtes par batch, Structured Streaming et Databricks-to-Databricks Delta Lake Sharing. Consultez Lecture des modifications dans les batch queries et Traitement incrémentiel des données de modification.

Exigences​

  • Databricks Runtime 18 ou une version ultérieure
  • Un format de table pris en charge et enregistré dans Unity Catalog :
    • Une table gérée au format Delta Lake avec le suivi des lignes activé ou au format Iceberg v3.
    • Une table externe au format Delta Lake avec suivi des lignes activé.

Consultez les types de table de Databricks Unity Catalog.

remarque

Le flux de données de modification ne fait pas partie de la spécification Apache Iceberg. Les lecteurs Databricks peuvent interroger le flux de données de modification automatique pour les tables Apache Iceberg v3, mais les lecteurs Iceberg externes ne le peuvent pas. Consultez la spécification de table Iceberg.

Pour Delta Lake, seuls les lecteurs Databricks peuvent query le flux de données de modification automatique.

Utiliser le flux de données de modification​

Pour utiliser le flux de données de modification, vérifiez que vous utilisez une table qui répond aux exigences. Consultez la page Conditions requises.

Pour lire batch le flux de données de modification, procédez comme suit :

Python
spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("<table_name>")

Pour plus d'information sur les lectures par batch pour le flux de données de modification, consultez Lire les modifications dans les queries par batch.

Pour lire en continu le Stream de données de modification, procédez comme suit :

Python
(spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")
)

Pour plus d'information sur les lectures en streaming pour le flux de données de modification, consultez Traiter les données de modification de manière incrémentielle.

Migrer à partir d'un flux de données de modification hérité​

Pour migrer une table Delta Lake d'un flux de données de modification hérité vers un flux de données de modification automatique, procédez comme suit :

  1. Vérifiez que votre table répond aux exigences.
  2. Désactivez le flux de données de modification hérité en exécutant la commande suivante :
SQL
ALTER TABLE <table_name> UNSET TBLPROPERTIES ('delta.enableChangeDataFeed');

Vous ne pouvez pas utiliser les flux de données de modification hérités et automatiques ensemble.

Schéma du flux de données de modification​

Lorsque vous lisez le flux de données de modification d'une table, la query utilise le schéma de la dernière version de la table. Databricks prend en charge la plupart des Opérations de modification et d'évolution de schéma, mais les tables avec mappage de colonnes ont des limitations. Consultez Tables avec mappage de colonnes.

En plus des colonnes de données du schéma de la table Delta Lake, le flux de données de modification contient des colonnes de métadonnées qui identifient le type d'événement de modification :

Nom de colonne

Type

Valeurs

_change_type

Chaîne

Contient : insert, update_preimage, update_postimage, delete.

preimage est la valeur avant la mise à jour, postimage est la valeur après la mise à jour.

_commit_version

Long

Contient : le log Delta ou la version de la table contenant la modification.

_commit_timestamp

Horodatage

Contient : le timestamp associé à la création du commit.

Nom de colonne

Type

Valeurs

_change_type

Chaîne

Contient : insert, update_preimage, update_postimage, delete.

preimage est la valeur avant la mise à jour, postimage est la valeur après la mise à jour.

_commit_version

Long

Contient : le log Delta ou la version de la table contenant la modification.

_commit_timestamp

Horodatage

Contient : le timestamp associé à la création du commit.

Si le schéma contient des colonnes portant les mêmes noms que ces colonnes de métadonnées, vous ne pouvez pas utiliser le flux de données de modification sur une table. Avant d'activer le flux de données de modification, renommez les colonnes de votre table pour résoudre ce conflit.

Traiter de manière incrémentielle les données de modification​

Databricks recommande d'utiliser le flux de données de modification conjointement avec Structured Streaming pour traiter progressivement les modifications provenant des tables. Vous devez utiliser Structured Streaming pour que Databricks suive automatiquement les versions du flux de données de modification de votre table. Pour le traitement CDC avec les tables SCD de type 1 ou de type 2, consultez Les API AUTO CDC : Simplifier la capture des modifications de données avec des pipelines.

Lorsque le Stream first starts pour la première fois, le flux de données de modification renvoie le dernier instantané de la table sous forme de INSERT enregistrements, puis renvoie les changements futurs en tant que données de modification. Les flux de données de modification commit à la fois les données de modification et les nouvelles lignes de données dans le journal des transactions de la table simultanément.

Pour configurer un Stream afin de lire le flux de données de modification d’une table, définissez l’option readChangeFeed sur true comme suit :

Python
(spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")
)

Limites de débit​

Databricks prend en charge les limites de débit (maxFilesPerTrigger, maxBytesPerTrigger) et excludeRegex lors de la lecture des données de modification. Pour une liste complète des options de streaming de Delta Lake, voir Delta Lake.

En option, vous pouvez spécifier une version de départ, voir Spécifier une version de départ. Pour les versions autres que l'instantané de départ, les limites de débit s'appliquent atomiquement à l'intégralité des commits. Soit le batch actuel inclut le commit entier, soit le batch actuel diffère le commit au batch suivant.

Relire l'historique de la table​

Un flux de données de modification n’est pas destiné à servir de registre permanent de toutes les modifications apportées à une table. Il enregistre uniquement les modifications qui surviennent après l’activation du flux de données de modification. Vous pouvez start une nouvelle lecture en streaming pour capturer la version actuelle et toutes les modifications ultérieures.

Les enregistrements du flux de données de modification sont transitoires et accessibles uniquement pendant une fenêtre de rétention spécifiée. Les Logs de transactions suppriment les versions de table et leurs versions de flux de données de modification correspondantes à intervalles réguliers. Lorsqu'une version est supprimée, vous ne pouvez plus lire le flux de données de modification pour cette version.

Archiver les données de modification pour l'historique permanent​

Si votre cas d'utilisation vous oblige à maintenir un historique permanent de toutes les modifications apportées à une table, utilisez une logique incrémentielle pour écrire les enregistrements du flux de données de modification dans une nouvelle table.

L’exemple suivant montre comment utiliser trigger.AvailableNow pour traiter les données disponibles en tant que charge de travail par batch pour l’audit ou les relectures complètes des modifications :

Python
(spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("target_table")
)

Spécifier une version de démarrage​

Pour lire les modifications à partir d’un point spécifique, veuillez spécifier une version de départ à l’aide d’un Timestamp ou d’un numéro de version. Les versions de départ sont requises pour les lectures en batch. Optionnellement, vous pouvez spécifier une version finale pour limiter la plage. Pour en savoir plus sur l'historique des tables, consultez Time travel.

Lorsque vous configurez des charges de travail Structured Streaming qui utilisent un flux de données de modification, la spécification d'une version de départ pourrait affecter les performances de traitement :

  • Les nouveaux pipelines de traitement de données bénéficient généralement du comportement par default, qui enregistre tous les enregistrements existants dans la table en tant qu'opérations INSERT lorsque le Stream démarre pour la première fois.
  • Si votre table cible contient déjà tous les enregistrements avec les modifications appropriées jusqu'à un certain point, spécifiez une version de départ pour éviter de traiter l'état de la table source en tant qu'événements INSERT.

L'exemple suivant montre comment récupérer d'une défaillance de streaming avec un point de contrôle corrompu. Dans cet exemple, supposez les conditions suivantes :

  1. Le flux de données de modification a été activé sur la table source lors de la création de la table.
  2. La table en aval cible a traité toutes les modifications jusqu'à la version 75 incluse.
  3. L'historique des versions de la table source est disponible pour les versions 70 et supérieures.

Lorsque vous définissez le write Stream vers la table cible existante, vous devez spécifier un nouvel emplacement de point de contrôle :

Python
(spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")
)
important

Si vous spécifiez une version de départ et que cette version n'est pas disponible dans l'historique de la table, le Stream ne start pas à partir d'un nouveau point de contrôle. Étant donné que les tables gérées nettoient automatiquement les versions historiques, toutes les versions de départ spécifiées sont finalement supprimées.

Consultez Revoir l’historique de la table.

Lire les changements dans les queries par batch​

Vous pouvez utiliser la syntaxe de requête batch pour lire toutes les modifications à partir d'une version particulière ou pour lire les modifications au sein d'une plage de versions spécifiée, comme suit :

  • Spécifiez les versions sous forme d'entiers et les Timestamp sous forme de chaînes de caractères au format yyyy-MM-dd[ HH:mm:ss[.SSS]].
  • Les versions de start et de fin sont inclusives. Pour lire d'une version de départ à la dernière version, spécifiez uniquement la version de départ.
  • Si vous spécifiez une version antérieure à l'activation du flux de données de modification, cela génère une erreur.

Pour utiliser les lectures par batch avec les options de version de début et de fin, procédez comme suit :

Pour lire de la version 0 à 10, procédez comme suit :

SQL
SELECT * FROM table_changes('tableName', 0, 10)

Pour lire entre deux versions de timestamp, procédez comme suit :

SQL
--
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')

Pour lire d'une version de départ à la dernière, procédez comme suit :

SQL
SELECT * FROM table_changes('tableName', 0)

Pour lire les modifications d’une table dont le nom contient des caractères spéciaux, procédez comme suit :

SQL
SELECT * FROM table_changes('`schema`.`dotted.tableName`', '2021-04-21 06:45:46', '2021-05-21 12:00:00')

Consultez table_changes fonction à valeur de table.

Gérer les versions hors plage​

Par default, si vous spécifiez une version ou un timestamp qui dépasse le dernier commit, la query retourne l'erreur timestampGreaterThanLatestCommit.

Dans Databricks Runtime 11.3 LTS et versions ultérieures, vous pouvez activer la tolérance pour les versions hors de portée comme suit :

SQL
SET spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled = true;

Lorsque cette configuration est activée, la query renvoie des résultats différents comme suit :

  • Une version ou un Timestamp de départ au-delà du dernier commit renvoie un résultat vide.
  • Une version de fin ou un Timestamp au-delà du dernier commit renvoie toutes les modifications du start au dernier commit.

Flux de données de modification hérité pour Delta Lake​

Le flux de données de modification hérité nécessite une configuration manuelle pour les tables Delta Lake individuelles. Étant donné que le flux de données de modification n'est pas inclus dans la spécification Apache Iceberg, les tables Apache Iceberg ne sont pas prises en charge. Databricks vous recommande de migrer vers le flux de données de modification automatique. Consultez Migrer depuis le flux de données de modification hérité.

Lorsque le flux de données de modification hérité est activé, le runtime enregistre les événements de modification pour toutes les données écrites dans la table. Cela inclut les données de ligne ainsi que les métadonnées indiquant si la ligne spécifiée a été insérée, supprimée ou mise à jour.

Le flux de données de modification hérité utilise les mêmes API de lecture readChangeFeed et table_changes() que le flux de données de modification automatique. Voir Traiter les données de modification de manière incrémentielle et Lire les modifications dans les query batch.

Activer le flux de données de modification hérité​

Vous devez activer explicitement le flux de données de modification héritées sur les tables individuelles. Utilisez l'une des méthodes suivantes :

Nouvelle table​

Définissez la propriété de table delta.enableChangeDataFeed = true dans la commande CREATE TABLE.

SQL
CREATE TABLE student (id INT, name STRING, age INT)
  TBLPROPERTIES (delta.enableChangeDataFeed = true)
remarque

Si vous désactivez l'ancien flux de données de modification pour un intervalle de temps, puis le réactivez, l'intervalle ne pourra pas être interrogé. Utilisez le flux de données de modification automatique pour interroger les modifications pendant l'intervalle. Consultez l'alimentation automatique de données de modification.

Table existante​

Définissez la propriété de table delta.enableChangeDataFeed = true dans la commande ALTER TABLE.

SQL
ALTER TABLE myDeltaTable
  SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

Considérations relatives au stockage​

Les tables gérées enregistrent les modifications de données efficacement et peuvent utiliser d'autres fonctionnalités pour optimiser le Layout de stockage.

Avec le flux de données de modification hérité, vous devez prendre en compte le comportement de stockage suivant :

  • Vous pourriez constater une légère augmentation des coûts de stockage, car les modifications pourraient être enregistrées dans des fichiers distincts.
  • Certaines opérations, telles que les suppressions uniquement d'insertion ou de partition complète, ne génèrent pas de fichiers de données de modification. Databricks calcule le flux de données de modification directement à partir du journal des transactions.
  • Les fichiers de données de modification utilisent la politique de conservation de la table. La commande VACUUM supprime les fichiers de données de modification, et les modifications du transaction log utilisent la politique de conservation des points de contrôle.

Databricks vous recommande de ne pas tenter de reconstituer le flux de données de modification en interrogeant directement les fichiers de données de modification. Toujours utiliser les APIs Delta Lake et Apache Iceberg.

Limitations​

Veuillez tenir compte des limites suivantes pour les flux de données de modification :

Tables avec mappage de colonnes​

Avec le mappage de colonnes activé sur une table Delta Lake, vous pouvez supprimer ou renommer des colonnes sans réécrire les fichiers de données. Consultez Renommer et supprimer des colonnes avec le mappage de colonnes Delta Lake.

Cependant, les flux de données de modification ont des limitations après des changements de schéma non additifs. Les modifications non additives du schéma incluent les opérations suivantes :

  • Renommer ou supprimer des colonnes.
  • Modifier les types de données des colonnes.
  • Modifiez la propriété de nullité de la colonne, comme avec ALTER COLUMN ... SET NOT NULL. Voir la NOT NULL contrainte.

Vous ne pouvez pas lire les flux de données de modification pour une transaction ou une plage dans laquelle un changement de schéma non additif se produit.

Pour permettre des modifications non additives du schéma avant ou après la plage spécifiée de lectures par batch, les requêtes utilisent le schéma de la version de fin de la plage plutôt que la dernière version de la table. Les queries échouent toujours si la plage de versions s'étend sur un changement de schéma non additif.

Flux de données de modification automatique​

  • Étant donné que le flux de données de modification n'est pas pris en charge dans les spécifications d'Apache Iceberg, les clients Iceberg externes ne peuvent pas interroger le flux de données de modification automatique. Consultez la spécification de la table Iceberg.
  • Pour les transactions multi-instructions, si la table source a été modifiée pendant la transaction, l'alimentation automatique des données de modification n'est pas prise en charge.
  • Le flux de données de modification automatique n'est pas pris en charge sur les tables avec des filtres de ligne ou des masques de colonne. Voir les filtres de lignes et masques de colonnes.
  • Les queries du Change Data Feed ne peuvent pas couvrir les versions de table où un changement de schéma non additif s'est produit, tel qu'un renommage de colonne, une suppression ou un changement de type de données. Divisez la query en plages avant et après la modification du schéma.
Sur cette page