Ce que le destinataire fait concrètement avec une table partagée
par Jordan Soldo et Nikhil Tallam
De nombreuses architectures de partage de données commencent par un schéma familier :
La première partie de ce blog a décrit pourquoi une table partagée peut constituer une meilleure alternative à un fichier récurrent ou à un flux API. Examinons maintenant ce que le destinataire fait réellement de cette table.
Cette procédure se concentre sur le partage entre environnements Databricks (D2D). Le destinataire dispose d'un espace de travail Databricks compatible avec Unity Catalog et accède à la table partagée via un catalogue partagé. Le fournisseur partage la table avec son historique, ce qui permet au destinataire de l'utiliser comme source Structured Streaming. Nous reviendrons sur le partage vers des environnements ouverts (D2O) plus loin dans cet article.
Une table Delta conserve un historique ordonné de chaque modification qui lui est apportée. OpenSharing expose cette table via un protocole ouvert sans la copier. Les données restent dans le stockage du fournisseur et le destinataire les lit en temps réel.
Du côté du destinataire, vous pointez un lecteur Structured Streaming vers la table partagée. Spark enregistre sa progression dans un point de contrôle, de sorte que chaque exécution ne récupère que les modifications apportées depuis la dernière. Le suivi des « nouveautés depuis ma dernière consultation » devient la tâche du moteur, s'exécutant sur des données gouvernées dont le lignage et le contrôle d'accès sont préservés.
Prenons un exemple : un distributeur gère une table de commandes en direct et la partage avec un destinataire externe utilisant Databricks. Au lieu de recevoir un fichier CSV chaque nuit, le destinataire peut interroger directement la table de commandes partagée. Si le destinataire a besoin d'une copie locale pour l'analytique ou pour des charges de travail en aval, il peut matérialiser les données de manière incrémentielle dans une table Delta. La configuration s'effectue une seule fois et est répartie entre le fournisseur et le destinataire.
Créez la table source en activant Change Data Feed, puis publiez-la dans un partage avec l'option WITH HISTORY. L'historique est ce qui fait de la table une source incrémentielle. Sans lui, le destinataire n'obtient qu'un instantané statique. L'activation de Change Data Feed (delta.enableChangeDataFeed) est facultative pour la lecture incrémentielle principale ci-dessous. Ne l'activez que si vous prévoyez d'utiliser ultérieurement le parcours des upserts et des suppressions. Si vous utilisez Databricks Runtime 19 ou une version ultérieure, vous pouvez ignorer cette propriété et vous appuyer sur la fonction automatic Change Data Feed, qui calcule les modifications au niveau des lignes au moment de l'interrogation pour les tables dont le suivi des lignes est activé, et qui fonctionne avec le partage OpenSharing de Databricks à Databricks.
Enregistrez le destinataire par son identifiant de partage, puis accordez les droits sur le partage :
Veuillez noter que les commandes CREATE SHARE et CREATE RECIPIENT nécessitent des privilèges au niveau du métastore et non pas seulement des droits d'administrateur d'espace de travail.
Le partage apparaît désormais comme un fournisseur dans votre métastore. Montez-le en tant que catalogue et vous pourrez le lire comme n'importe quelle autre table.
Une fois le partage monté sous le nom shared_in, la lecture est gérée par Structured Streaming.
Trigger.AvailableNow traite toutes les données actuellement disponibles et non traitées à l'aide d'un ou de plusieurs micro-lots, puis s'arrête. Cela en fait une solution idéale pour les tâches incrémentielles planifiées. Au lieu d'un calendrier fixe, vous pouvez également configurer la tâche avec un déclencheur de mise à jour de table sur la table partagée, de sorte que la tâche ne s'exécute que lorsque le fournisseur valide de nouvelles données. Il est recommandé de stocker le point de contrôle dans un stockage cloud durable tel que UC Volumes.
Remarque : Cet exemple lit une table partagée via Delta Sharing, il utilise donc ignoreChanges. Si vous diffusez des flux directement à partir d'une table Delta classique, les runtimes Databricks plus récents utilisent skipChangeCommits pour le traitement en mode ajout uniquement.
Pour prouver qu'il s'agit bien d'un traitement incrémentiel, exécutez le flux une première fois sur un ensemble initial de lignes. Réexécutez-le après avoir ajouté de nouvelles lignes chez le fournisseur. L'historique des validations de la table cible affichera deux écritures en streaming :
Version | Opération | Métriques de l'opération |
|---|---|---|
2 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 200, numOutputBytes: 2307, numAddedFiles: 1 |
1 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 1000, numOutputBytes: 4958, numAddedFiles: 1 |
0 | CREATE TABLE | {} |
La deuxième exécution a écrit 200 lignes, et non 1200. Le point de contrôle a suivi la dernière version lue, de sorte que l'exécution n'a extrait que les nouvelles.
ignoreChanges convient à ce type de chargement incrémentiel, mais il peut réémettre des lignes et ne propage pas les suppressions. Si les consommateurs en aval ont besoin d'insertions, de mises à jour et de suppressions, lisez plutôt le flux de modifications (Change Data Feed) de la table partagée :
Lorsqu'un nouveau client est ajouté, le fournisseur n'a pas besoin de créer un nouveau point de terminaison SFTP ni de mettre en œuvre une nouvelle tâche d'exportation. Le fournisseur accorde au destinataire l'accès au partage. Le destinataire utilise le même schéma général tout en maintenant distinctes ses propres tables cibles et sa logique métier en aval. La différence opérationnelle est significative :
Tout ce qui précède fonctionne parce que le destinataire exécute Spark sur Databricks, ce qui permet d'utiliser le streaming avec points de contrôle et le CDF. D'autres destinataires (pandas, outils décisionnels, Snowflake) se connectent via OpenSharing à l'aide d'un fichier d'informations d'identification plutôt que d'un identifiant de partage de métastore.
S'ils exécutent Spark Open Source, la lecture utilise le même code spark.readStream via le connecteur Delta Sharing. Sans Spark, il n'y a pas de flux continu, ils récupèrent donc les modifications par plage de versions :
Un premier projet utile n'a pas besoin de résoudre tous les scénarios de partage. Commencez par une seule table Delta classique et un seul destinataire.
Avant de mettre en œuvre le flux, répondez à ces questions :
Une fois ces réponses claires, validez le flux de bout en bout le plus simple possible : créez le partage, lisez un instantané initial, écrivez une nouvelle ligne, mettez à jour une ligne, supprimez une ligne et confirmez que le destinataire obtient les résultats attendus.
C'est la transformation pratique qu'OpenSharing rend possible. Au lieu de créer un nouveau flux de transport et d'ingestion pour chaque client, vous pouvez établir un contrat de partage de données gouverné et permettre à chaque destinataire de consommer les données via le parcours qui convient à sa plateforme.
Vous débutez avec OpenSharing ? Commencez par Qu'est-ce que OpenSharing ? et Créer des partages pour OpenSharing, puis configurez le flux de bout en bout minimal ci-dessus avec une table et un destinataire.
(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.