Utiliser Databricks SQL dans un travail Azure Databricks

Article
05/08/2024

Vous pouvez utiliser le type de tâche SQL dans un travail Azure Databricks, ce qui vous permet de créer, de planifier, d’utiliser et de surveiller des flux de travail qui incluent des objets Databricks SQL tels que des requêtes, des tableaux de bord hérités et des alertes. Par exemple, votre flux de travail peut ingérer des données, préparer les données, effectuer une analyse à l’aide de requêtes Databricks SQL, puis afficher les résultats dans un tableau de bord hérité.

Cet article fournit un exemple de flux de travail qui crée un tableau de bord hérité affichant des métriques pour les contributions GitHub. Dans cet exemple, vous allez :

Ingérer des données GitHub à l’aide d’un script Python et de l’API REST GitHub.
Transformer les données GitHub à l’aide d’un pipeline Delta Live Tables.
Déclencher des requêtes Databricks SQL en effectuant une analyse sur les données préparées.
Affichez l’analyse dans un tableau de bord hérité.

Tableau de bord d’analyse GitHub

Avant de commencer

Les éléments suivants sont nécessaires pour effectuer cette procédure pas à pas :

Un jeton d’accès personnel GitHub. Ce jeton doit disposer de l’autorisation du référentiel.
Entrepôt SQL serverless ou entrepôt SQL professionnel. Consulter Types d’entrepôts SQL.
L’étendue de secret Databricks. L’étendue du secret est utilisée pour stocker en toute sécurité le jeton GitHub. Consultez Étape 1 : Stocker le jeton GitHub dans un secret.

Étape 1 : Stocker le jeton GitHub dans un secret

Au lieu de coder en dur les informations d’identification, telles que le jeton d’accès personnel GitHub dans un travail, Databricks recommande d’utiliser une étendue de secret pour stocker et gérer les secrets en toute sécurité. Les commandes CLI Databricks suivantes sont un exemple de création d’une étendue de secret et de stockage du jeton GitHub dans un secret dans cette étendue :

databricks secrets create-scope <scope-name>
databricks secrets put-secret <scope-name> <token-key> --string-value <token>

Remplacez <scope-name par le nom d’une étendue de secret Azure Databricks pour stocker le jeton.
Remplacez <token-key> par le nom d’une clé à affecter au jeton.
Remplacer <token> par la valeur du jeton d'accès personnel GitHub.

Étape 2 : Créer un script pour extraire des données GitHub

Le script Python suivant utilise l’API REST GitHub pour extraire des données sur les validations et les contributions d’un référentiel GitHub. Les arguments d’entrée spécifient le référentiel GitHub. Les enregistrements sont enregistrés à un emplacement dans DBFS spécifié par un autre argument d’entrée.

Cet exemple utilise DBFS pour stocker le script Python, mais vous pouvez également utiliser Dossiers Git Databricks ou des fichiers d’espace de travail pour stocker et gérer le script.

Enregistrez ce script à un emplacement sur votre disque local :

import json
import requests
import sys

api_url = "https://api.github.com"

def get_commits(owner, repo, token, path):
  page = 1
  request_url =  f"{api_url}/repos/{owner}/{repo}/commits"
  more = True

  get_response(request_url, f"{path}/commits", token)

def get_contributors(owner, repo, token, path):
  page = 1
  request_url =  f"{api_url}/repos/{owner}/{repo}/contributors"
  more = True

  get_response(request_url, f"{path}/contributors", token)

def get_response(request_url, path, token):
  page = 1
  more = True

  while more:
    response = requests.get(request_url, params={'page': page}, headers={'Authorization': "token " + token})
    if response.text != "[]":
      write(path + "/records-" + str(page) + ".json", response.text)
      page += 1
    else:
      more = False

def write(filename, contents):
  dbutils.fs.put(filename, contents)

def main():
  args = sys.argv[1:]
  if len(args) < 6:
    print("Usage: github-api.py owner repo request output-dir secret-scope secret-key")
    sys.exit(1)

  owner = sys.argv[1]
  repo = sys.argv[2]
  request = sys.argv[3]
  output_path = sys.argv[4]
  secret_scope = sys.argv[5]
  secret_key = sys.argv[6]

  token = dbutils.secrets.get(scope=secret_scope, key=secret_key)

  if (request == "commits"):
    get_commits(owner, repo, token, output_path)
  elif (request == "contributors"):
    get_contributors(owner, repo, token, output_path)

if __name__ == "__main__":
    main()

Chargez le script sur DBFS :
1. Accédez à votre page de destination Azure Databricks et cliquez sur l'Catalogue dans la barre latérale.
2. Cliquez sur Parcourir DBFS.
3. Dans l’explorateur de fichiers DBFS, cliquez sur Charger. La boîte de dialogue Charger des données dans DBFS s’affiche.
4. Entrez un chemin d’accès dans DBFS pour stocker le script, cliquez sur Supprimer les fichiers à charger ou cliquez pour parcourir, puis sélectionnez le script Python.
5. Cliquez sur Done.

Étape 3 : Créer un pipeline Delta Live Tables pour traiter les données GitHub

Dans cette section, vous allez créer un pipeline Delta Live Tables pour convertir les données GitHub brutes en tables qui peuvent être analysées par des requêtes Databricks SQL. Pour créer le pipeline, procédez comme suit :

Cliquez sur New (Nouveau) dans la barre latérale, puis sélectionnez Notebook dans le menu. La boîte de dialogue Create Notebook s’affiche.
Dans Langage par défaut, saisissez un nom, puis sélectionnez Python. Vous pouvez conserver la valeur par défaut du paramètre Cluster. Le runtime Delta Live Tables crée un cluster avant d’exécuter votre pipeline.
Cliquez sur Créer.

Copiez l’exemple de code Python et collez-le dans votre nouveau notebook. Vous pouvez ajouter l’exemple de code à une seule cellule du notebook ou à plusieurs cellules.

import dlt
from pyspark.sql.functions import *

def parse(df):
   return (df
     .withColumn("author_date", to_timestamp(col("commit.author.date")))
     .withColumn("author_email", col("commit.author.email"))
     .withColumn("author_name", col("commit.author.name"))
     .withColumn("comment_count", col("commit.comment_count"))
     .withColumn("committer_date", to_timestamp(col("commit.committer.date")))
     .withColumn("committer_email", col("commit.committer.email"))
     .withColumn("committer_name", col("commit.committer.name"))
     .withColumn("message", col("commit.message"))
     .withColumn("sha", col("commit.tree.sha"))
     .withColumn("tree_url", col("commit.tree.url"))
     .withColumn("url", col("commit.url"))
     .withColumn("verification_payload", col("commit.verification.payload"))
     .withColumn("verification_reason", col("commit.verification.reason"))
     .withColumn("verification_signature", col("commit.verification.signature"))
     .withColumn("verification_verified", col("commit.verification.signature").cast("string"))
     .drop("commit")
   )

@dlt.table(
   comment="Raw GitHub commits"
)
def github_commits_raw():
  df = spark.read.json(spark.conf.get("commits-path"))
  return parse(df.select("commit"))

@dlt.table(
  comment="Info on the author of a commit"
)
def commits_by_author():
  return (
    dlt.read("github_commits_raw")
      .withColumnRenamed("author_date", "date")
      .withColumnRenamed("author_email", "email")
      .withColumnRenamed("author_name", "name")
      .select("sha", "date", "email", "name")
  )

@dlt.table(
  comment="GitHub repository contributors"
)
def github_contributors_raw():
  return(
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load(spark.conf.get("contribs-path"))
  )

Cliquez sur l’Flux de travail dans la barre latérale, cliquez sur l’onglet Delta Live Tables, puis sur Créer un pipeline.
Donnez un nom au pipeline, par exemple, Transform GitHub data.
Dans le champ Bibliothèques de notebooks, saisissez le chemin d’accès à votre notebook ou cliquez sur l’ pour sélectionner le notebook.
Cliquez sur Ajouter une configuration. Dans la zone de texte Key, saisissez commits-path. Dans la zone de texte Value, saisissez le chemin d’accès DBFS où les enregistrements GitHub seront écrits. Il peut s’agir de n’importe quel chemin d’accès que vous choisissez et il s’agit du même chemin d’accès que celui que vous utiliserez lors de la configuration de la première tâche Python lorsque vous créez le flux de travail.
Cliquez à nouveau sur Ajouter une configuration. Dans la zone de texte Key, saisissez contribs-path. Dans la zone de texte Value, saisissez le chemin d’accès DBFS où les enregistrements GitHub seront écrits. Il peut s’agir de n’importe quel chemin d’accès que vous choisissez et il s’agit du même chemin d’accès que celui que vous utiliserez lors de la configuration de la deuxième tâche Python lorsque vous créez le flux de travail.
Dans le champ Cible, saisissez une base de données cible, par exemple, github_tables. La définition d’une base de données cible publie les données de sortie dans le metastore et est nécessaire pour les requêtes en aval analysant les données produites par le pipeline.
Cliquez sur Enregistrer.

Étape 4 : Créer un flux de travail pour ingérer et transformer des données GitHub

Avant d’analyser et de visualiser les données GitHub avec Databricks SQL, vous devez ingérer et préparer les données. Pour créer un flux de travail afin d’effectuer ces tâches, procédez comme suit :

Créer un travail Azure Databricks et ajouter la première tâche

Accédez à votre page d’accueil Azure Databricks et effectuez l’une des opérations suivantes :
- Dans la barre latérale, cliquez sur l’Workflows et cliquez sur le .
- Dans la barre latérale, cliquez sur Nouvelle et sélectionnez Travail dans le menu.
Dans la boîte de dialogue de la tâche qui s’affiche sous l’onglet Tâches, remplacez Ajouter un nom à votre travail... par votre nom de travail, par exemple GitHub analysis workflow.
Pour Nom de la tâche, entrez un nom pour la tâche, par exemple get_commits.
Dans Type, sélectionnez Script Python.
Dans Source, sélectionnez DBFS/S3.
Dans Chemin d’accès, saisissez le chemin d’accès au script dans DBFS.
Dans Paramètres, saisissez les arguments suivants pour le script Python :

["<owner>","<repo>","commits","<DBFS-output-dir>","<scope-name>","<github-token-key>"]
- Remplacez <owner> par le nom du propriétaire du référentiel. Par exemple, pour extraire des enregistrements du référentiel github.com/databrickslabs/overwatch, saisissez databrickslabs.
- Remplacez <repo> par le nom du référentiel, par exemple, overwatch.
- Remplacez <DBFS-output-dir> par un chemin d’accès dans DBFS pour stocker les enregistrements extraits de GitHub.
- Remplacez <scope-name> par le nom de l’étendue de secret que vous avez créée pour stocker le jeton GitHub.
- Remplacez <github-token-key> par le nom de la clé que vous avez affectée au jeton GitHub.
Cliquez sur Enregistrer la tâche.

Ajouter une autre tâche

Cliquez sur en dessous de la tâche que vous venez de créer.
Pour Nom de la tâche, entrez un nom pour la tâche, par exemple get_contributors.
Dans Type, sélectionnez le type de tâche Script Python.
Dans Source, sélectionnez DBFS/S3.
Dans Chemin d’accès, saisissez le chemin d’accès au script dans DBFS.
Dans Paramètres, saisissez les arguments suivants pour le script Python :

["<owner>","<repo>","contributors","<DBFS-output-dir>","<scope-name>","<github-token-key>"]
- Remplacez <owner> par le nom du propriétaire du référentiel. Par exemple, pour extraire des enregistrements du référentiel github.com/databrickslabs/overwatch, saisissez databrickslabs.
- Remplacez <repo> par le nom du référentiel, par exemple, overwatch.
- Remplacez <DBFS-output-dir> par un chemin d’accès dans DBFS pour stocker les enregistrements extraits de GitHub.
- Remplacez <scope-name> par le nom de l’étendue de secret que vous avez créée pour stocker le jeton GitHub.
- Remplacez <github-token-key> par le nom de la clé que vous avez affectée au jeton GitHub.
Cliquez sur Enregistrer la tâche.

Ajouter une tâche pour transformer les données

Cliquez sur en dessous de la tâche que vous venez de créer.
Pour Nom de la tâche, entrez un nom pour la tâche, par exemple transform_github_data.
Dans Type, sélectionnez Pipeline Delta Live Tables et saisissez un nom pour la tâche.
Dans Pipeline, sélectionnez le pipeline créé à l’étape 3 : Créer un pipeline Delta Live Tables pour traiter les données GitHub.
Cliquez sur Créer.

Étape 5 : Exécuter le flux de travail de transformation des données

Cliquez sur le pour exécuter le flux de travail. Pour voir les détails de l’exécution, cliquez sur le lien dans la colonne Heure de début de l’exécution dans la vue Exécutions du travail. Cliquez sur chaque tâche pour afficher les détails de l’exécution de la tâche.

Étape 6 : (facultatif) Pour afficher les données de sortie une fois l’exécution du flux de travail terminée, procédez comme suit :

Dans la vue détails de l’exécution, cliquez sur la tâche Delta Live Tables.
Dans le panneau Détails de l’exécution de la tâche, cliquez sur le nom du pipeline sous Pipeline. La page Détails du pipeline s’affiche.
Sélectionnez la table commits_by_author dans le DAG de pipeline.
Cliquez sur le nom de la table en regard de Metastore dans le panneau commits_by_author. La page Catalog Explorer s'ouvre.

Dans Catalog Explorer, vous pouvez afficher le schéma de tables, les exemples de données et d'autres détails pour les données. Suivez les mêmes étapes pour afficher les données de la table github_contributors_raw.

Étape 7 : Supprimer les données GitHub

Dans une application réelle, vous pouvez ingérer et traiter des données en continu. Étant donné que cet exemple télécharge et traite l’ensemble du jeu de données, vous devez supprimer les données GitHub déjà téléchargées pour éviter une erreur lors de la réexécution du workflow. Pour supprimer les données téléchargées, procédez comme suit :

Créez un notebook et entrez les commandes suivantes dans la première cellule :
```
dbutils.fs.rm("<commits-path", True)
dbutils.fs.rm("<contributors-path", True)
```
Remplacez <commits-path> et <contributors-path> par les chemins d’accès DBFS que vous avez configurés lors de la création des tâches Python.
Cliquez sur le , puis sélectionnez Exécuter la cellule.

Vous pouvez également ajouter ce notebook en tant que tâche dans le flux de travail.

Étape 8 : Créer des requêtes Databricks SQL

Après avoir exécuté le flux de travail et créé les tables requises, créez des requêtes pour analyser les données préparées. Pour créer les exemples de requêtes et de visualisations, procédez comme suit :

Afficher les 10 premiers contributeurs par mois

Cliquez sur l’icône située sous le logo Databricks dans la barre latérale, puis sélectionnez SQL.
Cliquez sur Créer une requête pour ouvrir l’éditeur de requêtes Databricks SQL.
Certifiez-vous que le catalogue soit défini sur hive_metastore. Cliquez sur la valeur par défaut à côté de hive_metastore et définissez la base de données sur la valeur Cible que vous avez définie dans le pipeline Delta Live Tables.

Sous l’onglet Nouvelle requête, entrez la requête suivante :

SELECT
  date_part('YEAR', date) AS year,
  date_part('MONTH', date) AS month,
  name,
  count(1)
FROM
  commits_by_author
WHERE
  name IN (
    SELECT
      name
    FROM
      commits_by_author
    GROUP BY
      name
    ORDER BY
      count(name) DESC
    LIMIT 10
  )
  AND
    date_part('YEAR', date) >= 2022
GROUP BY
  name, year, month
ORDER BY
  year, month, name

Cliquez sur l’onglet Nouvelle requête et renommez la requête, par exemple,. Commits by month top 10 contributors.
Par défaut, les résultats sont affichés dans une table. Pour modifier la façon dont les données sont visualisées, par exemple, à l’aide d’un graphique à barres, dans le volet Résultats, cliquez sur , puis cliquez sur Modifier.
Dans Type de visualisation, sélectionnez Barre.
Dans la Colonne X, sélectionnez mois.
Dans les Colonnes Y, sélectionnez count(1).
Dans Regrouper par, sélectionnez nom.
Cliquez sur Enregistrer.

Afficher les 20 premiers contributeurs

Cliquez sur + > Créer une requête et assurez-vous que le catalogue est défini sur hive_metastore. Cliquez sur la valeur par défaut à côté de hive_metastore et définissez la base de données sur la valeur Cible que vous avez définie dans le pipeline Delta Live Tables.

Entrez la requête suivante :

SELECT
  login,
  cast(contributions AS INTEGER)
FROM
  github_contributors_raw
ORDER BY
  contributions DESC
LIMIT 20

Cliquez sur l’onglet Nouvelle requête et renommez la requête, par exemple, Top 20 contributors.
Pour modifier la visualisation à partir de la table par défaut, dans le volet Résultats, cliquez sur , puis cliquez sur Modifier.
Dans Type de visualisation, sélectionnez Barre.
Dans la Colonne X, sélectionnez connexion.
Dans les Colonnes Y, sélectionnez contributions.
Cliquez sur Enregistrer.

Afficher le nombre total de validations par auteur

Cliquez sur + > Créer une requête et assurez-vous que le catalogue est défini sur hive_metastore. Cliquez sur la valeur par défaut à côté de hive_metastore et définissez la base de données sur la valeur Cible que vous avez définie dans le pipeline Delta Live Tables.

Entrez la requête suivante :

SELECT
  name,
  count(1) commits
FROM
  commits_by_author
GROUP BY
  name
ORDER BY
  commits DESC
LIMIT 10

Cliquez sur l’onglet Nouvelle requête et renommez la requête, par exemple, Total commits by author.
Pour modifier la visualisation à partir de la table par défaut, dans le volet Résultats, cliquez sur , puis cliquez sur Modifier.
Dans Type de visualisation, sélectionnez Barre.
Dans la Colonne X, sélectionnez nom.
Dans les Colonnes Y, sélectionnez validations.
Cliquez sur Enregistrer.

Étape 9 : Créer un tableau de bord

Dans la barre latérale, cliquez sur Tableaux de bord.
Cliquez sur Créer un tableau de bord.
Entrez un nom pour le tableau de bord, par exemple, GitHub analysis.
Pour chaque requête et visualisation créée à l’Étape 8 : Créer les requêtes Databricks SQL, cliquez sur Ajouter > Visualisation et sélectionnez chaque visualisation.

Étape 10 : Ajouter les tâches SQL au flux de travail

Pour ajouter les nouvelles tâches de requête au flux de travail que vous avez créé dans Créer un travail Azure Databricks et ajouter la première tâche, pour chaque requête que vous avez créée à l’Étape 8 : Créer les requêtes Databricks SQL :

Cliquez sur Workflows dans la barre latérale.
Dans la colonne Nom, cliquez sur le nom d’un travail.
Cliquez sur l'onglet Tâches.
Cliquez sur le sous la dernière tâche.
Entrez un nom pour la tâche, dans Type, sélectionnez SQL et, dans Tâche SQL, sélectionnez Requête.
Sélectionnez la requête dans la requête SQL.
Dans l’entrepôt SQL, sélectionnez un entrepôt SQL serverless ou un entrepôt SQL professionnel pour exécuter la tâche.
Cliquez sur Créer.

Étape 11 : Ajouter une tâche de tableau de bord

Cliquez sur le sous la dernière tâche.
Entrez un nom pour la tâche, dans Type, sélectionnez SQL et, dans Tâche SQL, sélectionnez Tableau de bord hérité.
Sélectionnez le tableau de bord créé à l’Étape 9 : Créer un tableau de bord.
Dans l’entrepôt SQL, sélectionnez un entrepôt SQL serverless ou un entrepôt SQL professionnel pour exécuter la tâche.
Cliquez sur Créer.

Étape 12 : Exécuter le flux de travail complet

Pour exécuter le workflow, cliquez sur le . Pour voir les détails de l’exécution, cliquez sur le lien dans la colonne Heure de début de l’exécution dans la vue Exécutions du travail.

Étape 13 : Afficher les résultats

Pour afficher les résultats une fois l’exécution terminée, cliquez sur la tâche de tableau de bord finale, puis cliquez sur le nom du tableau de bord sous tableau de bord SQL dans le volet droit.

Share via