De la conformité bitemporelle aux mises à jour partielles d'enregistrements : fournir une capture de données modifiées robuste et prête pour l'audit, sans code personnalisé
par Josh Seidel, Shanelle Roman et Sudhanva Huruli
La capture de données modifiées (CDC) est l'une des tâches les plus courantes que les ingénieurs de données conçoivent sur Spark, et l'une des plus fastidieuses à réaliser correctement à la main. Dans notre article précédent, Arrêtez de coder à la main vos pipelines de capture de données modifiées, nous avons présenté comment AUTO CDC dans Apache™ Spark Declarative Pipelines (SDP) automatise le SCD Type 1, le SCD Type 2 et le Snapshot CDC en remplaçant des centaines de lignes de logique MERGE fragiles par quelques déclarations simples.
À mesure que les exigences des pipelines évoluent, les ingénieurs sont confrontés à des situations que ces modèles CDC standards ont du mal à résoudre :
Aujourd'hui, nous portons AUTO CDC au niveau supérieur pour résoudre précisément ces défis du monde réel, et nous étendons ces fonctionnalités à la version open source d'Apache Spark 4.2.
Les tables SCD Type 2 standards peuvent vous indiquer quand un fait a changé dans le monde réel, mais elles ne peuvent pas vous dire ce que votre système croyait à un moment précis.
En vertu de la règle SEC 17a-4 et des règles de conservation des registres de la FINRA, les entreprises doivent être en mesure de reconstruire les enregistrements tels qu'ils existaient à un instant T ; la seule campagne de contrôle de la SEC sur la conservation des registres a entraîné plus de 2 milliards de dollars d'amendes pour plus de 100 entreprises depuis 2021. Le plus difficile est rarement de stocker la valeur d'aujourd'hui. C'est de pouvoir répondre, des mois plus tard, à ce que disaient les données de référence à la date de reporting, et à ce que nos systèmes croyaient à ce moment-là.
Le SCD Type 2 standard suit une seule chronologie : le moment où un fait a changé. AUTO CDC bitemporel en suit deux, de manière indépendante :
Chaque table cible reçoit quatre colonnes gérées par le système : __START_AT et __END_AT pour le temps métier, __SYSTEM_START_AT et __SYSTEM_END_AT pour le temps système. Un seul fait logique peut avoir plusieurs lignes physiques, une par combinaison version métier/version système, ce qui permet de reconstruire les données à un instant T le long de l'un ou l'autre axe. La garantie de comportement clé : les événements peuvent arriver dans n'importe quel ordre sur l'une ou l'autre chronologie.
Lorsqu'une correction apparaît avec un temps métier ou un temps système antérieur à un élément déjà traité, le moteur réécrit l'historique concerné au lieu de simplement l'ajouter à la fin. Aucune logique écrite à la main, il suffit de déclarer les deux colonnes de séquençage et le moteur maintient les deux intervalles. Cela fonctionne tout aussi bien pour les tables de dimensions, comme les référentiels de symboles, que pour les tables de faits, comme l'historique des transactions ou les relevés de capteurs, qui nécessitent une audibilité stricte. Voici à quoi cela ressemble par rapport aux données de référence FINRA CAT :
Notez que la clause SQL exacte est STORED AS BITEMPORAL, et non STORED AS SCD TYPE BITEMPORAL, et qu'elle nécessite à la fois SEQUENCE BY et SYSTEM SEQUENCE BY. Supposons que l'indicateur de déclaration d'Acme change le 1er janvier (temps métier), mais que le flux ne le reçoive que le 5 janvier (temps système). Une correction rétroactive arrive ensuite le 8 janvier, indiquant que le changement réel a eu lieu le 1er janvier mais avec une valeur différente. AUTO CDC bitemporel peut répondre aux deux questions :
Le 3 janvier, la première requête ne renvoie rien, ce qui est la réponse correcte et auditable pour ce que le système affichait à ce moment-là. La deuxième requête, exécutée aujourd'hui, reflète la vérité corrigée. Deux horloges, deux réponses, toutes deux correctes. Les colonnes de séquençage doivent être de types triables, sans valeurs de séquençage NULL. La fonctionnalité s'exécute sur SDP serverless ou sur les éditions de produits Pro/Advanced, et est actuellement en version bêta, alors épinglez le pipeline sur le canal : PREVIEW.
Lorsqu'un modèle est entraîné sur des données de référence ou de caractéristiques (features), la reproductibilité signifie être capable de reconstruire le jeu de données exact utilisé par le modèle, des mois plus tard, lors d'un examen ou d'un audit. Le premier réflexe est de se tourner vers le voyage dans le temps de Delta Lake, mais il s'agit d'une propriété de l'historique des fichiers de la table, et non d'un enregistrement permanent. VACUUM supprime définitivement les fichiers de données qui ne sont plus référencés par les versions récentes ; une fois la fenêtre de rétention par défaut de 7 jours dépassée, un TIMESTAMP AS OF enregistré au moment de l'entraînement peut discrètement cesser de se résoudre. Une table bitemporelle stocke cet historique sous forme de données, et non de versions de fichiers. VACUUM et OPTIMIZE compactent les fichiers mais ne touchent jamais à l'historique logique, de sorte que chaque version métier ou système passée reste une ligne interrogeable. Il existe deux façons d'obtenir de la reproductibilité à partir de cela : enregistrer deux instants « as-of » (temps métier et temps système) en tant que paramètres MLflow, et épingler la requête d'entraînement à cet état de croyance :
Or, si la table expose une vue actuelle, enregistrez un seul instant système au moment de l'entraînement et reconstruisez-le plus tard avec une requête de temps système à ce timestamp :
Dans tous les cas, le contrat de reproductibilité se résume à quelques timestamps dans l'exécution MLflow, et parce que l'historique bitemporel est stocké sous forme de lignes, ce contrat tient même après que VACUUM a nettoyé les fichiers sous-jacents.
Toutes les sources de capture de données modifiées (CDC) n'émettent pas des lignes complètes pour les mises à jour. Au lieu de cela, beaucoup n'envoient que les champs qui ont changé, représentant toutes les autres colonnes par la valeur NULL. Sans traitement particulier, ces valeurs NULL peuvent écraser involontairement les données existantes dans la table cible. Jusqu'à présent, les clients devaient concevoir une logique personnalisée pour contourner ce comportement. Avec les mises à jour partielles AutoCDC, cela est désormais géré automatiquement.
Les mises à jour partielles étendent AutoCDC en permettant aux événements de mise à jour de modifier uniquement un sous-ensemble de colonnes. Pour les colonnes sélectionnées, les valeurs NULL dans une mise à jour entrante sont interprétées comme « ne pas mettre à jour » plutôt que d'écraser la valeur existante.
Ceci est particulièrement utile pour les sources CDC qui omettent les valeurs inchangées en émettant NULL. Sans les mises à jour partielles, ces valeurs NULL écraseraient les données existantes dans la table cible.
Par exemple, supposons que la table cible contienne : (1, 'A', 20)
Un événement de mise à jour entrant contient : (1, NULL, 30)
Par défaut, AutoCDC mettrait à jour la ligne comme suit : (1, NULL, 30).
Avec les mises à jour partielles activées, la valeur NULL dans name est traitée comme « laisser la valeur existante inchangée », ce qui donne : (1, 'A', 30).
L'activation des mises à jour partielles nécessite simplement l'ajout d'un paramètre à votre définition AutoCDC. Vous pouvez choisir parmi trois méthodes pour spécifier les colonnes à traiter comme des mises à jour partielles :
IGNORE NULL UPDATES ON columnListIGNORE NULL UPDATES ON * EXCEPT (columnList)COLUMNS TO UPDATEPour obtenir la syntaxe complète, des exemples et des conseils d'utilisation, consultez la documentation sur l'application des mises à jour partielles.
Spark Declarative Pipelines étant open source, son type de flux le plus largement utilisé devrait l'être aussi. Nous commençons par contribuer l'API Python pour AUTO CDC Type 1 à Apache Spark 4.2.
Nous y avons contribué de la même manière que le reste de Spark évolue : sous la forme d'une série de propositions examinées et de pull requests, et non d'un dépôt de code unique (voir le SPIP et SPARK-56249).
La correction des données hors séquence est intégrée d'office : une petite table auxiliaire suit l'état des événements arrivant en avance, comme les tombstones de suppression, les micro-lots rejoués convergent au lieu de corrompre la cible, et comme elle repose sur les abstractions de streaming et de table de Spark plutôt que sur un format de stockage, elle s'exécute à la fois sur Delta Lake et Apache Iceberg.
La suite, en open source :
CREATE FLOW ... AS AUTO CDC INTO) dans la branche master, qui sera disponible dans la prochaine version d'Apache Spark.NULL d'écraser les données cibles.Que vous souhaitiez mettre en œuvre la conformité bitemporelle, configurer des mises à jour partielles ou explorer AutoCDC open source dans Apache Spark, consultez les ressources ci-dessous pour commencer :
(Cet article de blog a été traduit à l'aide d'outils basés sur l'intelligence artificielle) Article original
Abonnez-vous à notre blog et recevez les derniers articles directement dans votre boîte mail.