Memodularisasi ataupun menautkan kode pada notebook
Artikel ini menjelaskan cara untuk menggunakan buku catatan Databricks untuk mengodekan alur kerja kompleks yang menggunakan kode modular, notebook yang tertaut atau disematkan, dan logika if-then-else.
Cara memodularisasi ataupun menautkan notebook
Perintah %run memungkinkan Anda menyertakan buku catatan lain di dalam buku catatan. Anda dapat menggunakan %run untuk memodulasi kode Anda, misalnya dengan menempatkan fungsi pendukung di buku catatan terpisah. Anda juga dapat menggunakannya untuk menggabungkan buku catatan yang menerapkan langkah-langkah dalam analisis. Saat Anda menggunakan %run, buku catatan yang dipanggil segera dijalankan dan fungsi serta variabel yang ditentukan di dalamnya menjadi tersedia di buku catatan panggilan.
Alur kerja buku catatan adalah pelengkap %run karena memungkinkan Anda meneruskan parameter ke dan mengembalikan nilai dari buku catatan. Ini memungkinkan Anda untuk membangun alur kerja dan alur yang kompleks dengan dependensi. Misalnya, Anda bisa mendapatkan daftar file dalam direktori dan meneruskan nama ke buku catatan lain, yang tidak mungkin dengan %run. Anda juga dapat membuat alur kerja if-then-else berdasarkan nilai pengembalian atau memanggil buku catatan lain menggunakan jalur relatif.
Untuk menerapkan alur kerja buku catatan, gunakan metode dbutils.notebook.*. Tidak seperti %run, metode dbutils.notebook.run() ini memulai pekerjaan baru untuk menjalankan buku catatan.
Metode ini, seperti semua API dbutils, hanya tersedia di Python dan Scala. Namun, Anda dapat menggunakan dbutils.notebook.run() untuk memanggil buku catatan R.
Peringatan
Pekerjaan berdasarkan alur kerja buku catatan harus diselesaikan dalam 30 hari atau lebih cepat. Pekerjaan yang berjalan lebih lama berdasarkan tugas notebook yang dimodulasikan atau ditautkan tidak akan didukung.
API
Metode yang dbutils.notebook tersedia dalam API untuk membangun alur kerja buku catatan adalah: run dan exit. Kedua parameter dan nilai kembali harus string.
run(path: String, timeout_seconds: int, arguments: Map): String
Menjalankan buku catatan dan mengembalikan nilai keluarnya. Metode ini memulai pekerjaan singkat yang berjalan dengan segera.
Parameter timeout_seconds mengontrol batas waktu lari (0 berarti tidak ada batas waktu): panggilan untuk run memberikan pengecualian jika tidak selesai dalam waktu yang ditentukan. Jika Azure Databricks turun selama lebih dari 10 menit, menjalankan buku catatan gagal terlepas dari timeout_seconds.
Parameter arguments menetapkan nilai widget dari buku catatan target. Secara khusus, jika buku catatan yang Anda jalankan memiliki widget bernama A, dan Anda melewati pasangan ("A": "B") nilai kunci sebagai bagian dari parameter argumen ke panggilan run(), maka mengambil nilai widget A akan kembali "B". Anda dapat menemukan petunjuk untuk membuat dan bekerja dengan widget di artikel widget Databricks.
Peringatan
Parameter arguments hanya menerima karakter Latin (set karakter ASCII). Menggunakan karakter non-ASCII akan mengembalikan kesalahan. Contoh karakter non-ASCII yang tidak valid adalah kanji Cina, Jepang, dan emoji.
run Penggunaan
Python
dbutils.notebook.run("notebook-name", 60, {"argument": "data", "argument2": "data2", ...})
Scala
dbutils.notebook.run("notebook-name", 60, Map("argument" -> "data", "argument2" -> "data2", ...))
run Contoh
Misalkan Anda memiliki buku catatan workflows bernama dengan widget bernama foo yang mencetak nilai widget:
dbutils.widgets.text("foo", "fooDefault", "fooEmptyLabel")
print dbutils.widgets.get("foo")
Menjalankan dbutils.notebook.run("workflows", 60, {"foo": "bar"}) memberikan hasil berikut:
Widget memiliki nilai yang Anda lewati melalui alur kerja, "bar", bukan default.
exit(value: String): void Keluar dari buku catatan dengan nilai. Jika Anda memanggil buku catatan menggunakan metode ini run, ini adalah nilai yang dikembalikan.
dbutils.notebook.exit("returnValue")
Memanggil dbutils.notebook.exit di pekerjaan menyebabkan buku catatan berhasil diselesaikan. Jika Anda ingin menyebabkan pekerjaan gagal, lemparkan pengecualian.
Contoh
Dalam contoh berikut, Anda meneruskan argumen ke DataImportNotebook dan menjalankan buku catatan yang berbeda (DataCleaningNotebook atau ErrorHandlingNotebook) berdasarkan hasil dari DataImportNotebook.
Saat alur kerja buku catatan berjalan, Anda akan melihat tautan ke buku catatan yang sedang berjalan:
Klik tautan buku catatan Pekerjaan buku catatan #xxxx untuk melihat detail lari:
Meneruskan data terstruktur
Bagian ini mengilustrasikan cara meneruskan data terstruktur antar buku catatan.
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))
Menangani kesalahan
Bagian ini mengilustrasikan cara menangani kesalahan dalam alur kerja buku catatan.
Python
# Errors in workflows thrown 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 in workflows thrown 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)
Menjalankan beberapa buku catatan secara bersamaan
Anda dapat menjalankan beberapa buku catatan secara bersamaan dengan menggunakan konstruksi Scala dan Python standar seperti Threads (Scala, Python) dan Futures (Scala, Python). Buku catatan alur kerja buku catatan tingkat lanjut menunjukkan cara menggunakan konstruksi ini. Buku catatan ada di Scala tetapi Anda dapat dengan mudah menulis yang setara dengan Python. Untuk menjalankan contoh:
- Unduh arsip buku catatan.
- Mengimpor arsip ke ruang kerja.
- Jalankan buku catatan Buku Catatan Bersamaan.