Exécuter un notebook Databricks à partir d’un autre notebook

Notes

Pour la plupart des cas d’usage d’orchestration, Databricks recommande d’utiliser des travaux Databricks. Utilisez l’API dbutils.notebook décrite dans cet article uniquement lorsque vous ne pouvez pas implémenter votre cas d’usage au moyen de travaux multitâches. C’est le cas, par exemple, pour l’exécution conditionnelle et pour le bouclage de notebooks sur un ensemble dynamique de paramètres.

Cet article explique comment utiliser des notebooks Databricks pour coder des workflows complexes qui utilisent du code modulaire, des notebooks liés ou incorporés, et une logique if-then-else.

Comparaison de %run et dbutils.notebook.run()

La commande %run vous permet d’inclure un autre notebook dans un notebook. Vous pouvez utiliser %run pour modulariser votre code, par exemple en plaçant des fonctions de prise en charge dans un autre notebook. Vous pouvez également l’utiliser pour concaténer les notebooks qui implémentent les étapes dans une analyse. Lorsque vous utilisez %run, le notebook appelé est immédiatement exécuté et les fonctions et les variables définies dans celui-ci sont disponibles dans le notebook appelant.

L’API dbutils.notebook vient en complément de %run, car elle vous permet de passer des paramètres à un notebook et de recevoir en retour des valeurs du notebook. Cela vous permet de créer des flux de travail et des pipelines complexes avec des dépendances. Par exemple, vous pouvez obtenir une liste de fichiers dans un répertoire et transmettre les noms à un autre notebook, ce qui n’est pas possible avec %run. Vous pouvez également créer des flux de travail if-then-else basés sur des valeurs renvoyées ou appeler d’autres notebooks en utilisant des chemins d’accès relatifs.

Contrairement à %run, la méthode dbutils.notebook.run() démarre un nouveau travail pour exécuter le notebook.

Ces méthodes, comme toutes les API dbutils, sont uniquement disponibles dans Python et Scala. Toutefois, vous pouvez utiliser dbutils.notebook.run() pour appeler un notebook R.

Notes

Les travaux créés avec l’API dbutils.notebook doivent être terminés dans un délai maximum de 30 jours.

API

Les méthodes disponibles dans l’API dbutils.notebook sont run et exit. Les paramètres et les valeurs renvoyées doivent être des chaînes.

run(path: String, timeout_seconds: int, arguments: Map): String

Exécutez un notebook et renvoyez sa valeur de sortie. La méthode démarre un travail éphémère qui s’exécute immédiatement.

Le paramètre timeout_seconds contrôle le délai d’attente de l’exécution (0 signifie aucun délai d’expiration) : l’appel à run lève une exception s’il ne se termine pas dans le délai spécifié. Si Azure Databricks est en panne pendant plus de 10 minutes, l’exécution du notebook échoue, peu importe timeout_seconds.

Le paramètre arguments définit les valeurs des widgets du notebook cible. Plus précisément, si le notebook que vous exécutez possède un widget nommé A et que vous transmettez une paire clé-valeur ("A": "B") dans le paramètre arguments à l’appel run(), la récupération de la valeur du widget A retourne "B". Vous trouverez les instructions relatives à la création et à l’utilisation des widgets dans l’article Widgets Databricks.

Notes

Le paramètre accepte uniquement les arguments caractères latins (jeu de caractères ASCII). L’utilisation de caractères non-ASCII retourne une erreur.

run Utilisation

Python

dbutils.notebook.run("notebook-name", 60, {"argument": "data", "argument2": "data2", ...})

Scala

dbutils.notebook.run("notebook-name", 60, Map("argument" -> "data", "argument2" -> "data2", ...))

run Exemple

Supposons que vous avez un notebook nommé workflows avec un widget nommé foo qui imprime la valeur du widget :

dbutils.widgets.text("foo", "fooDefault", "fooEmptyLabel")
print dbutils.widgets.get("foo")

L’exécution de dbutils.notebook.run("workflows", 60, {"foo": "bar"}) génère les résultats suivants :

Notebook avec widget

Le widget avait la valeur que vous aviez passée avec dbutils.notebook.run(), soit "bar", à la place de la valeur par défaut.

exit(value: String): void Quitter un notebook avec une valeur. Si vous appelez un notebook à l’aide de la méthode run, il s’agit de la valeur retournée.

dbutils.notebook.exit("returnValue")

Si vous appelez dbutils.notebook.exit dans un travail, le notebook se termine correctement. Si vous souhaitez provoquer l’échec du travail, levez une exception.

Exemple

Dans l’exemple suivant, vous transmettez des arguments à DataImportNotebook et vous exécutez différents notebookq (DataCleaningNotebook ou ErrorHandlingNotebook) en fonction du résultat de DataImportNotebook.

Exemple if-else

Durant l’exécution du code, vous voyez un lien vers le notebook en cours d’exécution :

Lien vers le notebook en cours d’exécution

Pour voir les détails de l’exécution, cliquez sur le lien du notebook Notebook job #xxxx.

Résultat de l’exécution du notebook éphémère

Transmettre des données structurées

Cette section explique comment passer des données structurées entre des notebooks.

Python

# Example 1 - returning data through temporary views.
# You can only return one string using dbutils.notebook.exit(), but since called notebooks reside in the same JVM, you can
# return a name referencing data stored in a temporary view.

## In callee notebook
spark.range(5).toDF("value").createOrReplaceGlobalTempView("my_data")
dbutils.notebook.exit("my_data")

## In caller notebook
returned_table = dbutils.notebook.run("LOCATION_OF_CALLEE_NOTEBOOK", 60)
global_temp_db = spark.conf.get("spark.sql.globalTempDatabase")
display(table(global_temp_db + "." + returned_table))

# Example 2 - returning data through DBFS.
# For larger datasets, you can write the results to DBFS and then return the DBFS path of the stored data.

## In callee notebook
dbutils.fs.rm("/tmp/results/my_data", recurse=True)
spark.range(5).toDF("value").write.format("parquet").load("dbfs:/tmp/results/my_data")
dbutils.notebook.exit("dbfs:/tmp/results/my_data")

## In caller notebook
returned_table = dbutils.notebook.run("LOCATION_OF_CALLEE_NOTEBOOK", 60)
display(spark.read.format("parquet").load(returned_table))

# Example 3 - returning JSON data.
# To return multiple values, you can use standard JSON libraries to serialize and deserialize results.

## In callee notebook
import json
dbutils.notebook.exit(json.dumps({
  "status": "OK",
  "table": "my_data"
}))

## In caller notebook
result = dbutils.notebook.run("LOCATION_OF_CALLEE_NOTEBOOK", 60)
print(json.loads(result))

Scala

// Example 1 - returning data through temporary views.
// You can only return one string using dbutils.notebook.exit(), but since called notebooks reside in the same JVM, you can
// return a name referencing data stored in a temporary view.

/** In callee notebook */
sc.parallelize(1 to 5).toDF().createOrReplaceGlobalTempView("my_data")
dbutils.notebook.exit("my_data")

/** In caller notebook */
val returned_table = dbutils.notebook.run("LOCATION_OF_CALLEE_NOTEBOOK", 60)
val global_temp_db = spark.conf.get("spark.sql.globalTempDatabase")
display(table(global_temp_db + "." + returned_table))

// Example 2 - returning data through DBFS.
// For larger datasets, you can write the results to DBFS and then return the DBFS path of the stored data.

/** In callee notebook */
dbutils.fs.rm("/tmp/results/my_data", recurse=true)
sc.parallelize(1 to 5).toDF().write.format("parquet").save("dbfs:/tmp/results/my_data")
dbutils.notebook.exit("dbfs:/tmp/results/my_data")

/** In caller notebook */
val returned_table = dbutils.notebook.run("LOCATION_OF_CALLEE_NOTEBOOK", 60)
display(sqlContext.read.format("parquet").load(returned_table))

// Example 3 - returning JSON data.
// To return multiple values, you can use standard JSON libraries to serialize and deserialize results.

/** In callee notebook */

// Import jackson json libraries
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.databind.ObjectMapper

// Create a json serializer
val jsonMapper = new ObjectMapper with ScalaObjectMapper
jsonMapper.registerModule(DefaultScalaModule)

// Exit with json
dbutils.notebook.exit(jsonMapper.writeValueAsString(Map("status" -> "OK", "table" -> "my_data")))

/** In caller notebook */
val result = dbutils.notebook.run("LOCATION_OF_CALLEE_NOTEBOOK", 60)
println(jsonMapper.readValue[Map[String, String]](result))

Gérer les erreurs

Cette section montre comment gérer les erreurs.

Python

# Errors throw a WorkflowException.

def run_with_retry(notebook, timeout, args = {}, max_retries = 3):
  num_retries = 0
  while True:
    try:
      return dbutils.notebook.run(notebook, timeout, args)
    except Exception as e:
      if num_retries > max_retries:
        raise e
      else:
        print("Retrying error", e)
        num_retries += 1

run_with_retry("LOCATION_OF_CALLEE_NOTEBOOK", 60, max_retries = 5)

Scala

// Errors throw a WorkflowException.

import com.databricks.WorkflowException

// Since dbutils.notebook.run() is just a function call, you can retry failures using standard Scala try-catch
// control flow. Here we show an example of retrying a notebook a number of times.
def runRetry(notebook: String, timeout: Int, args: Map[String, String] = Map.empty, maxTries: Int = 3): String = {
  var numTries = 0
  while (true) {
    try {
      return dbutils.notebook.run(notebook, timeout, args)
    } catch {
      case e: WorkflowException if numTries < maxTries =>
        println("Error, retrying: " + e)
    }
    numTries += 1
  }
  "" // not reached
}

runRetry("LOCATION_OF_CALLEE_NOTEBOOK", timeout = 60, maxTries = 5)

Exécuter plusieurs notebooks simultanément

Vous pouvez exécuter plusieurs notebooks en même temps à l’aide de constructions Scala et Python standard, telles que des Threads (Scala, Python) et Futures (Scala, Python). Les exemples de notebooks illustrent comment utiliser ces constructions. Pour exécuter l'exemple :

  1. Téléchargez l’archive de notebook. Ces notebooks sont écrits en Scala.
  2. Importez l’archive dans un espace de travail.
  3. Exécutez le notebook Notebooks concurrents.