Utiliser des tables de streaming autonomes
Une table de streaming autonome est une table enregistrée dans Unity Catalog avec un support supplémentaire pour le streaming ou le traitement incrémentiel des données, définie en dehors d'un pipeline Lakeflow. Un pipeline est automatiquement créé pour chaque table de streaming. Vous pouvez utiliser des tables de streaming pour le chargement incrémentiel de données à partir de Kafka et du stockage d'objets cloud.
Vous pouvez créer et refresh des streaming tables autonomes à partir d'un warehouse Databricks SQL ou d'un Notebook s'exécutant sur un compute général serverless. Pour plus de détails sur les différences entre les deux options de compute, consultez Conditions requises pour les pipelines autonomes.
Pour créer et refresh des tables de streaming autonomes avec Python à partir d’un Notebook, consultez Utiliser Python avec des pipeline autonomes.
Pour savoir comment utiliser les tables Delta Lake en tant que sources et puits de streaming, consultez Lectures et écritures en streaming de tables Delta Lake.
Conditions requises
Pour les options de compute, les autorisations et les autres exigences relatives à la création, à l'actualisation et à l'interrogation de tables de streaming autonomes, voir Exigences pour les pipelines autonomes.
Créer des tables de streaming
Une table de streaming est définie par une query SQL dans Databricks SQL. Lorsque vous créez une table de streaming, les données actuellement présentes dans les tables sources sont utilisées pour créer la table de streaming. Après cela, vous refresh la table, généralement selon un calendrier, pour intégrer les données ajoutées dans les tables sources à ajouter à la table de streaming.
Lorsque vous créez une table de streaming, vous êtes considéré comme le propriétaire de la table.
Pour créer une table de streaming à partir d’une table existante, utilisez l’ instruction CREATE STREAMING TABLE, comme dans l’exemple suivant :
CREATE OR REFRESH STREAMING TABLE sales
SCHEDULE EVERY 1 hour
AS SELECT product, price FROM STREAM raw_data;
Dans ce cas, la table de streaming sales est créée à partir de colonnes spécifiques de la table raw_data, avec un planning de refresh toutes les heures. La query utilisée doit être une query de streaming . Utilisez le mot-clé STREAM pour utiliser la sémantique de streaming pour lire depuis la source.
Compute utilisé pour l'refresh
Lorsque vous créez une table de streaming à l’aide de l’instruction CREATE OR REFRESH STREAMING TABLE, le refresh initial des données et le remplissage commencent immédiatement. Ces opérations ne consomment pas de compute Databricks SQL warehouse. Au lieu de cela, les tables de streaming s’appuient sur des pipelines serverless pour la création et le refresh. Un pipeline serverless dédié est automatiquement créé et géré par le système pour chaque table de streaming.
Charger des fichiers avec Auto Loader
Pour créer une table de streaming à partir de fichiers situés dans un volume, vous devez utiliser Auto Loader. Utilisez Auto Loader pour la plupart des tâches d'ingestion de données à partir de stockage d'objets cloud. Auto Loader et les pipelines sont conçus pour charger de manière incrémentielle et idempotente des données toujours croissantes au fur et à mesure de leur arrivée dans le stockage cloud.
Pour utiliser Auto Loader dans Databricks SQL, utilisez la fonction read_files. Les exemples suivants montrent l'utilisation d'Auto Loader pour lire un volume de fichiers JSON dans une table de streaming :
CREATE OR REFRESH STREAMING TABLE sales
SCHEDULE EVERY 1 hour
AS SELECT * FROM STREAM read_files(
"/Volumes/my_catalog/my_schema/my_volume/path/to/data",
format => "json"
);
Pour lire les données à partir du stockage cloud, vous pouvez également utiliser Auto Loader :
CREATE OR REFRESH STREAMING TABLE sales
SCHEDULE EVERY 1 hour
AS SELECT *
FROM STREAM read_files(
's3://mybucket/analysis/*/*/*.json',
format => "json"
);
Pour en savoir plus sur Auto Loader, consultez Qu’est-ce qu’Auto Loader ?. Pour en savoir plus sur l’utilisation d’Auto Loader dans SQL, avec des exemples, consultez Charger des données à partir du stockage d’objets.
Ingestion en streaming à partir d'autres sources
Pour un exemple d’ingestion à partir d’autres sources, y compris Kafka, voir Charger des données dans des pipelines.
Appliquer une capture de données de changement (CDC) avec des flux Auto CDC
Utilisez la clause FLOW AUTO CDC pour traiter les enregistrements CDC (Change Data Capture) d'une source dans une table de streaming. Auparavant, l'instruction MERGE INTO était couramment utilisée pour le traitement des enregistrements CDC sur Databricks. Cependant, MERGE INTO peut produire des résultats incorrects en raison d'enregistrements hors séquence ou nécessite une logique complexe pour réorganiser les enregistrements. Voir Capture et instantanés des données de modification.
AUTO CDC simplifie la CDC en traitant automatiquement les enregistrements en désordre. Vous spécifiez des clés pour identifier les enregistrements, une colonne de séquence pour le tri et si les résultats doivent être stockés en SCD type 1 (mises à jour directes) ou SCD type 2 (suivi de l’historique).
L’exemple suivant crée une table de streaming qui applique les changements CDC en utilisant le SCD de type 1 :
CREATE OR REFRESH STREAMING TABLE target
FLOW AUTO CDC
FROM stream(cdc_data.users)
KEYS (userId)
SEQUENCE BY sequenceNum
STORED AS SCD TYPE 1;
L’exemple suivant utilise les SCD de type 2 pour conserver un historique des modifications :
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;
Pour plus de détails sur les options et le comportement d’Auto CDC, voir Les APIs AUTO CDC : Simplifier la capture de données de changement avec des pipeline. Pour la référence complète de la syntaxe, voir CREATE STREAMING TABLE.
Appliquer un remplacement par batch sélectif avec les flux REPLACE WHERE
Utilisez la clause FLOW REPLACE WHERE pour recalculer et écraser un sous-ensemble ciblé d’une table de streaming sans retraiter tout l’historique de votre table. Les flux REPLACE WHERE sont bien adaptés au traitement par batch incrémentiel des jointures et des agrégations, aux données arrivées en retard, au retraitement en amont, à l’évolution des schémas et aux remplissages de données.
Pour obtenir tous les détails sur les flux REPLACE WHERE, y compris les exigences, les substitutions de prédicat et le refresh incrémentiel, consultez Flux REPLACE WHERE pour les tables de streaming autonomes.
Appliquer le remplacement partiel d’instantané avec les flux REPLACE USING
Bêta
Les flux REPLACE USING sont en version bêta.
Utilisez la clause FLOW REPLACE USING pour maintenir une table de streaming synchronisée avec un flux de snapshots partiels. À chaque mise à jour, un flux REPLACE USING remplace toutes les lignes qui correspondent aux colonnes de clé spécifiées et laisse toutes les autres lignes inchangées. Une colonne SEQUENCE BY ordonne les mises à jour afin que la séquence la plus élevée pour une clé soit toujours prioritaire, même si les mises à jour arrivent dans le désordre. Par exemple :
CREATE OR REFRESH STREAMING TABLE payments_current
FLOW REPLACE USING (payment_id) SEQUENCE BY payment_date BY NAME
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);
BY NAME est requis. Il fait correspondre les colonnes par leur nom plutôt que par leur position.
REPLACE USING se comporte de la même manière pour les tables de streaming autonomes que pour les pipelines LakeFlow Pipelines. Pour en savoir plus sur son fonctionnement, le séquencement, les attentes, les limitations et des exemples, consultez la section Remplacement partiel de snapshot avec les flux REPLACE USING. Les différences suivantes s'appliquent aux tables de streaming autonomes :
- Définissez le flux en SQL. Créez le flux REPLACE USING avec la clause SQL inline
FLOW REPLACE USINGsurCREATE OR REFRESH STREAMING TABLE. L’instructionCREATE FLOWautonome est une construction de pipeline Lakeflow et n’est pas utilisée pour les tables de streaming autonomes. - Compute est géré pour vous. Les tables de streaming autonomes s'exécutent sur des pipelines serverless gérés par le système et nécessitent Databricks Runtime 18.2 et versions ultérieures. Vous ne choisissez pas entre le compute classique et le compute Serverless.
Ingérer uniquement les nouvelles données
Par default, la fonction read_files lit toutes les données existantes dans le dossier source lors de la création de la table, puis traite les nouveaux enregistrements arrivants à chaque refresh.
Pour éviter l'ingestion de données déjà présentes dans le dossier source au moment de la création de la table, définissez l'option includeExistingFiles sur false. Cela signifie que seules les données arrivant dans le dossier après la création de la table sont traitées. Par exemple :
CREATE OR REFRESH STREAMING TABLE sales
SCHEDULE EVERY 1 hour
AS SELECT *
FROM STREAM read_files(
'/path/to/files',
includeExistingFiles => false
);
Version de Runtime
Les tables de streaming s'exécutent toujours sur la dernière version de l'environnement d'exécution Databricks SQL. La propriété de table pipelines.channel, précédemment utilisée pour sélectionner un Canal de distribution d’exécution preview ou current, n’est plus prise en charge et n’a aucun effet. Si une définition existante inclut cette propriété, elle est ignorée en toute sécurité et vous n’avez pas besoin de la supprimer.
Cacher les données sensibles
Vous pouvez utiliser des tables de streaming pour masquer les données sensibles des utilisateurs qui accèdent à la table. Une approche consiste à définir la query de manière à ce qu'elle exclue complètement les colonnes ou les lignes sensibles. Vous pouvez également appliquer des masques de colonne ou des filtres de ligne en fonction des autorisations de l'utilisateur qui exécute la query. Par exemple, vous pouvez masquer la colonne tax_id pour les utilisateurs qui ne sont pas dans le groupe HumanResourcesDept. Pour ce faire, utilisez la syntaxe ROW FILTER et MASK lors de la création de la table de streaming. Pour plus d'informations, voir Filtres de lignes et masques de colonne.
Refresh une table de streaming
Les tables de streaming créent et utilisent automatiquement des pipelines serverless pour traiter les opérations de refresh. Le refresh est géré par le pipeline et le refresh est surveillé par le warehouse Databricks SQL utilisé pour créer la table de streaming. Les tables de streaming peuvent être mises à jour à l’aide d’un pipeline s’exécutant selon une planification.
Même si une refresh planifiée est configurée, vous pouvez déclencher une refresh manuelle à tout moment. Les refresh sont gérés par le même pipeline que celui créé automatiquement lors de la création de la table de streaming.
Pour refresh une table de streaming :
REFRESH STREAMING TABLE sales;
Vous pouvez vérifier l’état du dernier refresh avec DESCRIBE TABLE EXTENDED.
Vous devrez peut-être refresh votre streaming table avant d'utiliser des query time travel.
Pour savoir comment programmer une refresh, consultez Planifier les refresh. Les refresh programmées peuvent avoir des notifications de mise à jour, et vous pouvez définir le mode de performance du refresh.
Comment fonctionne le refresh
Une refresh de table de streaming évalue uniquement les nouvelles lignes arrivées après la dernière mise à jour et ajoute uniquement les nouvelles données.
Chaque refresh utilise la définition actuelle de la table de streaming pour traiter ces nouvelles données. La modification d’une définition de table de streaming ne recalcule pas automatiquement les données existantes. Si une modification est incompatible avec les données existantes (par exemple, la modification d’un type de données), la prochaine refresh échouera avec une erreur.
Les exemples suivants expliquent comment les modifications apportées à la définition d’une table de streaming affectent le comportement de refresh :
- La suppression d’un filtre ne permet pas de retraiter les lignes précédemment filtrées.
- La modification des projections de colonnes n’affecte pas la façon dont les données existantes ont été traitées.
- Les jointures avec des instantanés statiques utilisent l’état de l’instantané au moment du traitement initial. Les données arrivées en retard qui auraient correspondu à l’instantané mis à jour sont ignorées. Cela peut entraîner l’élimination de faits si les dimensions sont en retard.
- La modification du CAST d'une colonne existante provoque une erreur.
Si vos données changent de manière à ne pas pouvoir être prises en charge dans la table de streaming existante, vous pouvez effectuer un refresh complet.
refresh complètement une table de streaming
Les refresh complètes retraitent toutes les données disponibles dans la source avec la dernière définition. Il n'est pas recommandé d'appeler des refresh complètes sur des sources qui ne conservent pas l'intégralité de l'historique des données ou qui ont de courtes périodes de rétention, telles que Kafka, car le refresh tronque les données existantes. Vous ne pourrez peut-être pas récupérer d'anciennes données si elles ne sont plus disponibles dans la source.
Par exemple :
REFRESH STREAMING TABLE sales FULL;
Planifier et surveiller les refresh
Vous pouvez refresh automatiquement une table de streaming selon un calendrier ou lorsque les données en amont changent, et vous pouvez configurer des délais d'expiration, des notifications et des modes de performance pour le refresh. Voir Planifier les refresh.
Contrôler l'accès aux tables de streaming
Les tables de streaming prennent en charge des contrôles d’accès enrichis pour favoriser le Data Sharing tout en évitant l’exposition de données potentiellement privées. Un propriétaire de table de streaming ou un utilisateur disposant du privilège MANAGE peut accorder des privilèges SELECT à d’autres utilisateurs. Les utilisateurs disposant d’un accès SELECT à la table de streaming n’ont pas besoin d’un accès SELECT aux tables référencées par la table de streaming. Ce contrôle d’accès facilite le Data Sharing tout en contrôlant l’accès aux données sous-jacentes.
Vous pouvez également modifier le propriétaire d’une table de streaming.
Accorder des privilèges à une table de streaming
Pour accorder l'accès à une table de streaming, utilisez l'instruction GRANT:
GRANT <privilege_type> ON <st_name> TO <principal>;
Les privilege_type peuvent être :
SELECT- l’utilisateur peutSELECTla table de streaming.REFRESH- l'utilisateur peutREFRESHla table de streaming. Les refresh sont exécutées en utilisant les autorisations du propriétaire.
L'exemple suivant crée une table en streaming et accorde des privilèges de sélection et de refresh aux utilisateurs :
CREATE OR REFRESH STREAMING TABLE st_name AS SELECT * FROM STREAM source_table;
-- Grant read-only access:
GRANT SELECT ON st_name TO read_only_user;
-- Grant read and refresh access:
GRANT SELECT ON st_name TO refresh_user;
GRANT REFRESH ON st_name TO refresh_user;
Pour plus d'informations sur l'octroi de privilèges sur les objets sécurisables Unity Catalog, reportez-vous à la référence des privilèges Unity Catalog.
Révoquer les privilèges d'une table de streaming
Pour révoquer l'accès à une table de streaming, utilisez l' instruction REVOKE:
REVOKE privilege_type ON <st_name> FROM principal;
Lorsque les privilèges SELECT sur une table source sont retirés au propriétaire de la table streaming ou à tout autre utilisateur auquel des privilèges MANAGE ou SELECT ont été accordés sur la table streaming, ou si la table source est supprimée, le propriétaire de la table streaming ou l’utilisateur auquel l’accès a été accordé peut toujours query la table streaming. Toutefois, le comportement suivant se produit :
- Le propriétaire de la table de streaming ou les autres personnes ayant perdu l'accès à une table de streaming ne peuvent plus
REFRESHcette table de streaming, et la table de streaming devient obsolète avec le temps. - En cas d’automatisation avec un planning, le prochain
REFRESHprogrammé échouera ou ne sera pas exécuté.
L’exemple suivant révoque le privilège SELECT de read_only_user:
REVOKE SELECT ON st_name FROM read_only_user;
Changer le propriétaire d'une table de streaming
Un utilisateur disposant des autorisations MANAGE sur une table de streaming autonome peut définir un nouveau propriétaire via l'explorateur de catalogues. Le nouveau propriétaire peut être lui-même ou un Service Principal sur lequel il dispose du rôle Utilisateur de Service Principal .
-
Depuis votre workspace Databricks, cliquez sur
Catalog pour ouvrir l'Explorateur de catalogues. -
Sélectionnez la table de streaming que vous souhaitez mettre à jour.
-
Dans la barre latérale droite, sous À propos de cette table de streaming , recherchez le Propriétaire et cliquez sur
modifier.
Si un message vous demande de mettre à jour le propriétaire en changeant l’utilisateur Exécuter en tant que dans les paramètres du pipeline, alors la table de streaming est définie dans un LakeFlow Pipelines, et non en tant que table autonome. Le message inclut un link vers les paramètres du pipeline, où vous pouvez changer l’utilisateur Exécuter en tant que .
-
Sélectionnez un nouveau propriétaire pour la table de streaming.
Les propriétaires ont automatiquement les privilèges
MANAGEetSELECTsur les tables streaming qu’ils possèdent. Si vous définissez un service principal comme propriétaire d’une table streaming que vous possédez, et que vous ne disposez pas explicitement des privilègesSELECTouMANAGEsur la table streaming, alors ce changement vous fera perdre tout accès à la table streaming. Dans ce cas, vous êtes invité à fournir explicitement ces privilèges.Sélectionnez les privilèges Accorder GÉRER et Accorder SÉLECTIONNER pour les fournir lors de l' Enregistrement .
-
Cliquez sur Enregistrer pour changer le propriétaire.
Le propriétaire de la table de streaming est mis à jour. Toutes les futures mises à jour sont exécutées à l’aide de l’identité du nouveau propriétaire.
Quand le propriétaire perd les privilèges des tables sources
Si vous changez le propriétaire et que le nouveau propriétaire n’a pas accès aux tables source (ou si les privilèges SELECT sont révoqués sur les tables source sous-jacentes), les utilisateurs peuvent toujours query la table de streaming. Cependant :
- Ils ne peuvent pas
REFRESHla table de streaming. - Le prochain refresh planifié de la table de streaming échoue.
La perte de l'accès aux données sources empêche les mises à jour, mais n'invalide pas immédiatement la table de streaming existante pour la lecture.
Supprimer définitivement les enregistrements d'une table de streaming
Aperçu
La prise en charge de l’instruction REORG avec des tables de streaming est en aperçu public.
- L'utilisation d'une instruction
REORGavec une table de streaming nécessite Databricks Runtime 15.4 et versions ultérieures. - Bien que vous puissiez utiliser l’instruction
REORGavec n’importe quelle table de streaming, elle n’est requise que lors de la suppression d’enregistrements d’une table de streaming avec vecteurs de suppression activés. La commande n’a aucun effet lorsqu’elle est utilisée avec une table de streaming sans vecteurs de suppression activés.
Pour supprimer physiquement des enregistrements du stockage sous-jacent d'une table de streaming avec des vecteurs de suppression activés, comme pour la conformité GDPR, des étapes supplémentaires doivent être effectuées pour garantir qu'une opération VACUUM s'exécute sur les données de la table de streaming.
Pour supprimer physiquement des enregistrements du stockage sous-jacent :
- Mettez à jour ou supprimez des enregistrements de la table de streaming.
- Exécutez une instruction
REORGsur la table de streaming, en spécifiant le parameterAPPLY (PURGE). Par exempleREORG TABLE <streaming-table-name> APPLY (PURGE);. - Attendez que la période de rétention des données de la table de streaming soit écoulée. La période de rétention des données default est de sept jours, mais elle peut être configurée avec la propriété de table
delta.deletedFileRetentionDuration. Consultez Configurer la rétention des données pour les query time travel. REFRESHla table de streaming. Voir refresh une table de streaming. Dans les 24 heures suivant l'opérationREFRESH, les tâches de maintenance du pipeline, y compris l'opérationVACUUMqui est nécessaire pour garantir que les enregistrements sont supprimés définitivement, sont exécutées automatiquement.
Surveiller les exécutions via l'historique des query
Vous pouvez utiliser la page de l'historique des query pour accéder aux détails des query et aux profils des query, ce qui peut vous aider à identifier les query peu performantes et les goulots d'étranglement dans le pipeline utilisé pour exécuter les mises à jour de vos tables de streaming. Pour un aperçu des types d’informations disponibles dans les historiques et les profils de query, consultez la page consacrée à l’historique des query et celle sur le profil des query.
Aperçu
Cette fonctionnalité est en aperçu public. Les administrateurs de workspace peuvent contrôler l'accès à cette fonctionnalité à partir de la page Prévisualisations . Consultez Gérer les aperçus Databricks.
Toutes les déclarations concernant les tables en streaming sont consultables dans l’historique des query. Vous pouvez utiliser le filtre déroulant Instruction pour sélectionner n’importe quelle commande et inspecter les queries associées. Toutes les instructions CREATE sont suivies d’une instruction REFRESH qui s’exécute de manière asynchrone sur un pipeline. Les instructions REFRESH incluent généralement des plans de query détaillés qui fournissent des insights sur l'optimisation des performances.
Pour accéder à REFRESH déclarations dans l’interface utilisateur de l’historique des query, procédez comme suit :
- Cliquez sur
dans la barre latérale gauche pour ouvrir l'interface utilisateur de Historique des query . - Sélectionnez la case à cocher REFRESH dans le filtre déroulant Instruction .
- Cliquez sur le nom de la déclaration de query pour afficher des détails résumés tels que la durée de la query et les métriques agrégées.
- Cliquez sur Voir le profil de la query pour ouvrir le profil de la query. Voir Profil de la query pour plus de détails sur la navigation dans le profil de la query.
- Vous pouvez également utiliser les liens de la section Source de la query pour ouvrir la query ou le pipeline associé.
Vous pouvez également accéder aux détails de la query en utilisant des Link dans l'éditeur SQL ou depuis un Notebook attaché à un SQL Warehouse.
Accéder aux tables de streaming depuis des clients externes
Pour accéder aux tables en streaming à partir de clients externes Delta Lake ou Iceberg qui ne prennent pas en charge les APIs ouvertes, vous pouvez utiliser le Mode. Le Mode de compatibilité crée une version en lecture seule de votre table en streaming accessible par n’importe quel client Delta Lake ou Iceberg.