Orchestrer des notebooks et modulariser le code dans les notebooks

Découvrez comment organiser les notebooks et modulariser du code dans des notebooks. Consultez des exemples et comprenez quand utiliser d’autres méthodes pour l’orchestration de notebooks.

Méthodes d’orchestration et de modularisation du code

Le tableau suivant compare les méthodes disponibles pour orchestrer des notebooks et modulariser le code dans les notebooks.

%run et dbutils.notebook.run()

La commande %run vous permet d’inclure un autre carnet dans un carnet. Vous pouvez utiliser %run pour modulariser votre code en plaçant des fonctions de prise en charge dans un bloc-notes distinct. Vous pouvez également l’utiliser pour concaténer les notebooks qui implémentent les étapes d'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 complète %run, car elle vous permet de transmettre des paramètres à un notebook et de retourner des valeurs à partir d’un 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 passer les noms à un autre bloc-notes, ce qui est impossible avec %run. Vous pouvez également créer des flux de travail if-then-else en fonction des valeurs de retour.

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

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

Utiliser %run pour importer un notebook

Dans cet exemple, le premier notebook définit une fonction reverse, celle-ci est disponible dans le deuxième notebook après l’utilisation de la commande magic %run pour exécuter shared-code-notebook.

Étant donné que les deux blocs-notes se trouvent dans le même répertoire dans l’espace de travail, utilisez le préfixe ./ dans ./shared-code-notebook pour indiquer que le chemin d’accès doit être résolu par rapport au bloc-notes en cours d’exécution. Vous pouvez organiser les notebooks en répertoires, tels que %run ./dir/notebook, ou utiliser un chemin absolu comme %run /Users/username@organization.com/directory/notebook.

Utiliser dbutils.notebook.run pour démarrer un nouveau travail

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

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

Le paramètre timeout_seconds contrôle le délai d’expiration de l’exécution (0 signifie qu’aucun délai d’attente n’est nécessaire). L’appel à run lance 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 indépendamment de 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 permettant de créer et d’utiliser des widgets dans la page des widgets Databricks .

run utilisation

Python

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

Langage de programmation Scala

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

Passer des données structurées entre des notebooks

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").save("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
import json

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

Langage de programmation 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, 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 */

// 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)

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)

Langage de programmation 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 simultanément plusieurs notebooks

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

1. Téléchargez les quatre blocs-notes suivants. Les notebooks sont écrits en Scala.
2. Importez les notebooks dans un dossier unique de l’espace de travail.
3. Exécutez le notebook Exécuter simultanément.

Exécuter un notebook en parallèle

Obtenir l'ordinateur portable

Exécuter dans un notebook parallèle

Obtenir l'ordinateur portable

Notebook de test

Obtenir l'ordinateur portable

Notebook de test 2

Obtenir l'ordinateur portable