create_auto_cdc_from_snapshot_flow

La create_auto_cdc_from_snapshot_flow fonction crée un flux qui utilise les fonctionnalités de capture de données modifiées des pipelines Lakeflow (CDC) pour traiter les données sources à partir d’instantanés de base de données. Découvrez comment fonctionne AUTO CDC FROM SNAPSHOT.

Note

Cette fonction remplace la fonction apply_changes_from_snapshot()précédente. Les deux fonctions ont la même signature. Databricks recommande la mise à jour pour utiliser le nouveau nom.

Important

Vous devez disposer d’une table de diffusion en continu cible pour cette opération. Pour créer la table cible requise, vous pouvez utiliser la fonction create_streaming_table().

Syntaxe

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

Note

Pour AUTO CDC FROM SNAPSHOT le traitement, le comportement par défaut consiste à insérer une nouvelle ligne lorsqu’un enregistrement correspondant avec la ou les mêmes clés n’existe pas dans la cible. Si un enregistrement correspondant existe, il est mis à jour uniquement si l’une des valeurs de la ligne a changé. Les lignes avec des clés présentes dans la cible, mais qui ne sont plus présentes dans la source, sont supprimées.

Pour en savoir plus sur le traitement CDC avec des captures instantanées, consultez les API AUTO CDC : simplifiez la capture de données modifiées à l'aide de pipelines. Pour des exemples d’utilisation de la create_auto_cdc_from_snapshot_flow() fonction, voir les exemples d’ingestion périodique d’instantanés, d’ingestion d’instantanés historiques et de remblayage SCD Type 1 .

Paramètres

Paramètre Type Descriptif
target str Obligatoire. Nom de la table à mettre à jour. Vous pouvez utiliser la fonction create_streaming_table() pour créer la table cible avant d’exécuter la create_auto_cdc_from_snapshot_flow() fonction.
source str ou lambda function Obligatoire. Nom d’une table ou d’une vue pour un instantané à intervalles réguliers, ou une fonction lambda Python qui retourne le DataFrame d’instantané à traiter, ainsi que la version de l’instantané. Consultez Implémenter l’argumentsource.
keys list Obligatoire. Colonne ou combinaison de colonnes qui identifient de manière unique une ligne dans les données sources. Cela permet d’identifier les événements cdc qui s’appliquent à des enregistrements spécifiques dans la table cible. Vous pouvez spécifier l’une ou l’autre des options suivantes :
  • Liste de chaînes : ["userId", "orderId"]
  • Liste des fonctions Spark SQL col() : [col("userId"), col("orderId"]. Les arguments des col() fonctions ne peuvent pas inclure de qualificateurs. Par exemple, vous pouvez utiliser col(userId), mais vous ne pouvez pas utiliser col(source.userId).
stored_as_scd_type str ou int Indique s’il faut stocker des enregistrements en tant que type SCD 1 ou SCD type 2. Défini sur pour 1 le type SCD 1 ou 2 pour le type SCD 2. La valeur par défaut est la méthode SCD de type 1.
track_history_column_list ou track_history_except_column_list list Sous-ensemble de colonnes de sortie à suivre pour l’historique dans la table cible. Permet track_history_column_list de spécifier la liste complète des colonnes à suivre. Permet track_history_except_column_list de spécifier les colonnes à exclure du suivi. Vous pouvez déclarer une valeur en tant que liste de chaînes ou en tant que fonctions Spark SQL col() :
  • track_history_column_list = ["userId", "name", "city"]
  • track_history_column_list = [col("userId"), col("name"), col("city")]
  • track_history_except_column_list = ["operation", "sequenceNum"]
  • track_history_except_column_list = [col("operation"), col("sequenceNum")

Les arguments des col() fonctions ne peuvent pas inclure de qualificateurs. Par exemple, vous pouvez utiliser col(userId), mais vous ne pouvez pas utiliser col(source.userId). La valeur par défaut est d’inclure toutes les colonnes de la table cible lorsque ni un argument track_history_column_list ni track_history_except_column_list n'est passé à la fonction.
once bool S’il ne faut lancer le flux d’instantanés qu’une seule fois. Lorsque True, le flux est sauté lors des mises à jour incrémentales suivantes après qu’il a réussi son validation. Un rafraîchissement complet de la cible relance le flux. Utilisez cette option pour un chargement instantané unique ou un remplissage. La valeur par défaut est False.

Notes

Pour les cibles SCD Type 1 dans les pipelines déclenchés, un create_auto_cdc_from_snapshot_flow() flux peut partager une cible avec un ou plusieurs AUTO CDC flux définis en Python ou SQL. Les conditions suivantes s'appliquent :

  • Donnez un nom unique à chaque AUTO CDC flux.
  • Utilisez le même nombre de tonalités dans le même ordre pour tous les flux de jeu. Les noms de clés en flux instantané sont comparés de manière insensible à la cascule avec les AUTO CDC noms de clés. Les flux multiples AUTO CDC doivent utiliser des noms clés et un casing identiques.
  • Utilisez exactement le même type de données pour la version instantanée et la colonne de séquençage de chaque AUTO CDC flux.
  • Ne définissez pas les attentes concernant le AUTO CDC FROM SNAPSHOT flow.
  • N’utilisez pas IGNORE NULL UPDATES. En Python, ne définissez ignore_null_updatespas , ignore_null_updates_column_list, ni ignore_null_updates_except_column_list sur les create_auto_cdc_flow() flux.
  • 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 .
  • N’utilisez pas ce schéma pour les cibles SCD de type 2 ou bitemporelles.

Pour un exemple, voir Ajouter un remplissage à une table AUTO CDC SCD Type 1.

Implémenter l’argument source

La create_auto_cdc_from_snapshot_flow() fonction inclut l’argument source . Pour le traitement des instantanés historiques, l’argument source est censé être une fonction lambda Python qui retourne deux valeurs à la create_auto_cdc_from_snapshot_flow() fonction : un DataFrame Python contenant les données d’instantané à traiter et une version d’instantané.

La première invocation de la fonction doit retourner un DataFrame instantané et une version. Si la première invocation retourne None, la mise à jour du pipeline échoue. Après qu’au moins un instantané a été traité, revenez None pour indiquer qu’aucun instantané supplémentaire n’est disponible.

Voici la signature de la fonction lambda :

lambda Any => Optional[(DataFrame, Any)]
  • L'argument de la fonction lambda est la version instantanée la plus récemment traitée.
  • La valeur de retour de la fonction lambda est None ou un tuple de deux valeurs : la première valeur du tuple est un DataFrame contenant l’instantané à traiter. La deuxième valeur du tuple est la version de l'instantané qui représente l’ordre logique de l’instantané.

Exemple qui implémente et appelle la fonction lambda :

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

Le runtime des pipelines Lakeflow effectue les étapes suivantes chaque fois que le pipeline qui contient la create_auto_cdc_from_snapshot_flow() fonction est déclenché :

  1. Exécute la next_snapshot_and_version fonction pour charger le DataFrame d’instantané suivant et la version d’instantané correspondante.
  2. Si la première invocation ne retourne pas de DataFrame, la mise à jour échoue. Si une invocation ultérieure ne retourne pas de DataFrame, l’exécution se termine et la mise à jour du pipeline est marquée comme terminée.
  3. Détecte les modifications apportées à la nouvelle capture instantanée et les applique de manière incrémentielle à la table cible.
  4. Retourne à l’étape 1 pour charger l’instantané suivant et sa version.
  • Dernière mise à jour :