Fournir des données rentables en temps réel avec dbt et Databricks

À mesure que les entreprises se développent, les volumes de données passent de Go à To (ou plus), et les exigences de latence passent d'heures à minutes (ou moins), ce qui rend de plus en plus coûteux de fournir des informations récentes à l'entreprise. Historiquement, les ingénieurs de données Python et Scala se sont tournés vers le streaming pour répondre à ces demandes, traitant efficacement les nouvelles données en temps réel, mais les ingénieurs analytiques qui avaient besoin de mettre à l'échelle les pipelines dbt basés sur SQL n'avaient pas cette option.

Ce n'est plus le cas ! Ce blog vise à illustrer comment nous pouvons utiliser les nouvelles tables de streaming et les vues matérialisées sur Databricks pour fournir des informations fraîches et en temps réel aux entreprises avec la simplicité de SQL et dbt.

Contexte

Lors du Data + AI Summit 2023, nous avons introduit les Tables de streaming et Vues matérialisées dans Databricks SQL. Cette fonctionnalité impressionnante a donné aux utilisateurs de Databricks SQL un accès facile à de nouvelles matérialisations de table puissantes introduites pour la première fois dans Delta Live Tables, leur permettant d'incrémenter de grandes requêtes, de streamer directement à partir de sources de données d'événements, et plus encore.

En plus d'utiliser nativement les Tables de streaming et les Vues matérialisées au sein d'un environnement Databricks, elles fonctionnent également pour les utilisateurs dbt sur Databricks. dbt-databricks est devenu l'un des moyens les plus populaires de construire des modèles de données sur Databricks, en tirant parti de toutes les capacités puissantes de Databricks SQL, y compris le moteur de calcul Photon, la mise à l'échelle instantanée des entrepôts SQL Serverless et le modèle de gouvernance Unity Catalog, avec l'ubiquité du framework de transformation de dbt.

Qu'est-ce qui a changé dans dbt-databricks ?

Depuis dbt v1.6+, dbt-databricks a évolué sur trois aspects clés :

1. Nouvelles matérialisations : "streaming_table" et "materialized_view"
2. Nouvelle syntaxe pour lire directement à partir du stockage de données cloud sans mettre en scène vos sources sous forme de table
3. Accès à des concepts de streaming avancés tels que les agrégations par fenêtre, le watermarking et les jointures stream-stream

Remarque : Gardez un œil sur la prochaine version de dbt v1.7.3 qui affinera davantage les capacités ci-dessus !

Jetons un coup d'œil à la façon dont nous pouvons utiliser ces nouvelles fonctionnalités avec la démo Airline Trips.

La démo Airline Trips

La démo Airline Trips a été créée pour démontrer comment ingérer et transformer de manière incrémentielle des données d'événements en direct pour des informations commerciales à jour sur Databricks, qu'il s'agisse d'un tableau de bord ou d'un modèle d'IA. L'ensemble de données représente tous les voyages aériens effectués aux États-Unis au fil du temps, capturant les retards des départs et des arrivées pour chaque voyage.

Un notebook d'aide inclus établit un flux simulé à partir de cet ensemble de données, tandis que le projet dbt présente un modèle de données qui prend ces événements JSON bruts et les transforme via un ETL en streaming en une couche de vues matérialisées, de tables de fonctionnalités, et plus encore.

Le dépôt est disponible publiquement ici, et utilise des données d'exemple empaquetées dans tous les espaces de travail Databricks prêts à l'emploi. N'hésitez pas à suivre !

Ingérer des données à partir du stockage de données cloud

L'une des façons les plus simples de commencer à utiliser les tables de streaming est l'ingestion de données à partir du stockage de données cloud, comme S3 pour AWS ou ADLS pour Azure. Vous pouvez avoir une source de données en amont générant des données d'événements à haut volume, et un processus pour les déposer sous forme de fichiers bruts dans un emplacement de stockage, généralement json, csv, parquet ou avro.

Dans notre démo, imaginons que nous recevions un flux en direct de chaque voyage aérien effectué aux États-Unis d'un tiers, et que nous voulions l'ingérer de manière incrémentielle au fur et à mesure.

Au lieu de mettre en scène les fichiers en tant que table externe, ou d'utiliser un outil tiers pour matérialiser une table Delta pour la source de données, nous pouvons simplement utiliser les tables de streaming pour résoudre ce problème. Prenons le modèle ci-dessous pour notre flux de voyages aériens bronze :

Les deux points clés à noter sont :

• La stratégie de matérialisation est définie sur 'streaming_table'
  • Cela exécutera une commande CREATE OR REFRESH STREAMING TABLE dans Databricks
• La syntaxe pour lire à partir du stockage cloud utilise Auto Loader sous le capot
  • read_files() listera les nouveaux fichiers JSON dans le dossier spécifié et commencera à les traiter. Puisque nous utilisons dbt, nous avons profité de la fonction var() dans dbt pour passer dynamiquement un chemin de dossier S3 (de la forme "s3://…")
  • Le mot-clé STREAM indique de streamer à partir de cet emplacement. Alternativement, sans lui, nous pouvons toujours utiliser read_files() avec materialized='table' pour effectuer une lecture par lots directement à partir du dossier spécifié

Par ailleurs, bien qu'Auto Loader nécessite le moins de configuration, vous pouvez également streamer directement à partir d'une plateforme de streaming d'événements comme Kafka, Kinesis ou Event Hubs pour une latence encore plus faible en utilisant une syntaxe très similaire. Voir ici pour plus de détails.

Enrichir incrémentiellement les données pour la couche argent

Le streaming ne doit pas s'arrêter à l'étape d'ingestion. Si nous voulons effectuer des jointures en aval ou ajouter une clé surrogate, mais que nous voulons le limiter aux nouvelles données uniquement pour économiser sur le calcul, nous pouvons continuer à utiliser la matérialisation de table de streaming. Par exemple, prenons l'extrait de notre prochain modèle pour la couche argent, le flux enrichi de voyages aériens, où nous joignons des tables de mappage pour les codes d'aéroport aux données brutes :

Encore une fois, nous avons utilisé la matérialisation de table de streaming, et avons pu exploiter la fonctionnalité standard de dbt pour toute notre logique. Cela inclut :

• Utilisation du package dbt_utils pour des raccourcis pratiques comme la génération d'une clé surrogate
• Utilisation de l'instruction ref() pour maintenir une lignée complète

Le seul changement réel apporté à notre SQL a été l'ajout du mot-clé STREAM() autour de l'instruction ref() pour airline_trips_bronze, pour indiquer que cette table est lue de manière incrémentielle, tandis que la table airport_codes jointe est une table de mappage qui est lue dans son intégralité. C'est ce qu'on appelle une jointure stream-statique.

Créer une couche or efficace en calcul avec des vues matérialisées

Avec nos tables argent enrichies prêtes, nous pouvons maintenant réfléchir à la manière dont nous voulons servir des informations agrégées à nos consommateurs commerciaux finaux. Typiquement, si nous utilisons une matérialisation de table, nous devrions recalculer tous les résultats historiques à chaque fois.

Pour tirer parti des tables de streaming en amont qui ne traitent que les nouvelles données à chaque exécution, nous nous tournons plutôt vers les vues matérialisées pour cette tâche !

La bonne nouvelle sur Databricks est qu'un modèle qui construit une vue matérialisée ne ressemble pas à un modèle qui construit une table ! Prenons notre exemple pour une vue matérialisée de couche or afin de calculer le pourcentage de vols retardés chaque jour :

Tout ce que nous avons changé, c'est la configuration de la matérialisation !

N'oubliez pas que les vues matérialisées peuvent être rafraîchies de manière incrémentielle lorsqu'il y a des modifications dans les tables de base. Dans le scénario ci-dessus, à mesure que nous traitons de nouvelles données en streaming, la vue matérialisée détermine quels groupes nécessitent un recalcul et ne calcule que ceux-ci, laissant les agrégations inchangées telles quelles et réduisant les coûts de calcul globaux. C'est plus facile à visualiser dans l'exemple car nous agrégeons sur ArrDate, la date d'arrivée des vols, ce qui signifie que les nouveaux jours de données tomberont naturellement dans de nouveaux groupes et que les groupes existants resteront inchangés.

En analysant les journaux d'événements de la vue matérialisée (illustrée ci-dessous) après plusieurs exécutions du modèle, nous pouvons voir l'incrémentation à l'œuvre. La première exécution est un calcul complet comme pour n'importe quelle table, mais une deuxième exécution pour mettre à jour les agrégations avec de nouvelles données utilise un rafraîchissement incrémentiel ligne par ligne. Une dernière exécution du modèle a reconnu qu'aucune nouvelle donnée n'avait été ingérée en amont et n'a simplement rien fait.

Quoi d'autre puis-je trouver dans le dépôt de démo ?

Nous avons couvert les bases de l'extraction des données directement de la source d'événements jusqu'à une vue matérialisée prête pour la BI, mais le dépôt de démo contient bien plus encore.

Le dépôt contient des exemples sur la façon de surveiller les journaux des tables en flux et des vues matérialisées pour comprendre comment les données sont traitées, ainsi qu'un exemple avancé non couvert dans ce blog sur la façon de joindre deux flux ensemble dans une jointure flux-flux en utilisant uniquement SQL !

Clonez le dépôt dans votre environnement Databricks pour commencer, ou connectez dbt Cloud à Databricks sans frais supplémentaires avec partner connect. Vous pouvez également en apprendre davantage avec la documentation sur les Vues Matérialisées et les Tables en Flux.

(Cet article de blog a été traduit à l'aide d'outils basés sur l'intelligence artificielle) Article original