ノートブックのオーケストレーションとノートブック内のコードのモジュール化
ノートブックを調整し、ノートブックでコードをモジュール化する方法について説明します。 例を参照し、ノートブック オーケストレーションの代替方法を使用する場合を理解します。
オーケストレーションとコードのモジュール化の方法
次の表は、ノートブックのオーケストレーションとノートブック内のコードのモジュール化に使用できる方法を比較したものです。
メソッド |
ユースケース |
注 |
---|---|---|
ノートブック オーケストレーション (推奨) |
ノートブックのオーケストレーションに推奨される方法。 タスクの依存関係、スケジューリング、トリガーを含む複雑なワークフローをサポートします。 本番運用ワークロードに対して堅牢でスケーラブルなアプローチを提供しますが、セットアップと構成が必要です。 |
|
ノートブックのオーケストレーション |
ジョブがユースケースをサポートできない場合 (動的なパラメーターセットでノートブックをループするなど) は、 コールごとに新しいエフェメラルジョブを開始するため、オーバーヘッドが増加し、高度なスケジューリング機能が不足します。 |
|
コードのモジュール化 (推奨) |
コードをモジュール化するための推奨される方法。 ワークスペースに格納された再利用可能なコード ファイルにコードをモジュール化します。 リポジトリによるバージョン管理と IDEs との統合をサポートし、デバッグと単体テストを改善します。 ファイルパスと依存関係を管理するには、追加の設定が必要です。 |
|
コードのモジュール化 |
ワークスペース ファイルにアクセスできない場合は、 他のノートブックから関数や変数をインラインで実行してインポートするだけです。 プロトタイプ作成には便利ですが、密結合されたコードになり、保守が難しくなる可能性があります。 パラメーターの受け渡しやバージョン管理はサポートされていません。 |
%run
対dbutils.notebook.run()
%run
コマンドを使用すると、ノートブック内に別のノートブックを含めることができます。%run
を使用して、サポート関数を別のノートブックに配置することで、コードをモジュール化できます。また、これを使用して、分析のステップを実装するノートブックを連結することもできます。 %run
を使用すると、呼び出されたノートブックがすぐに実行され、そのノートブックで定義された関数と変数が呼び出し元のノートブックで使用できるようになります。
dbutils.notebook
API は、ノートブックにパラメーターを渡したり、ノートブックから値を返すことができるため、%run
を補完します。これにより、依存関係を持つ複雑なワークフローとパイプラインを構築できます。 たとえば、ディレクトリ内のファイルのリストを取得し、その名前を別のノートブックに渡すことができますが、これは %run
では不可能です。 また、戻り値に基づいて if-then-else ワークフローを作成することもできます。
%run
とは異なり、dbutils.notebook.run()
メソッドでは、新しいジョブを開始してノートブックを実行します。
すべてのdbutils
APIsと同様に、これらの方法は Python と Scalaでのみ使用できます。 ただし、 dbutils.notebook.run()
を使用して R ノートブックを呼び出すことはできます。
%run
を使ってノートブックをインポートする
この例では、最初のノートブックは関数reverse
を定義しています。この関数は、%run
マジックを使用してshared-code-notebook
を実行した後に2番目のノートブックで使用できます。
両方のノートブックはワークスペース内の同じディレクトリにあるため、./shared-code-notebook
のプレフィックス ./
を使用して、現在実行中のノートブックを基準にしてパスを解決する必要があることを示します。ノートブックを %run ./dir/notebook
などのディレクトリに整理することも、 %run /Users/username@organization.com/directory/notebook
のような絶対パスを使用することもできます。
注
%run
はノートブック全体をインラインで実行するため、単独でセル内に置く必要があります。%run
を使用して Python ファイルを実行し、そのファイルで定義されているエンティティをノートブックにimport
することはできません 。Python ファイルからインポートするには、「 ファイルを使用したコードのモジュール化」を参照してください。 または、ファイルを Python ライブラリにパッケージ化し、その Python ライブラリから Databricks ライブラリを作成して、ノートブックの実行に使用する クラスターにライブラリをインストールします 。%run
を使用してウィジェットを含むノートブックを実行する場合、指定したノートブックはウィジェットのデフォルト値を使用して by デフォルト 実行されます。ウィジェットに値を渡すこともできます。 「Databricks ウィジェットを %run と共に使用する」を参照してください。
dbutils.notebook.run
を使用して新しいジョブを開始する
ノートブックを実行し、その終了値を返します。このメソッドは、すぐに実行される一時的なジョブを開始するものです。
dbutils.notebook
APIで使用できるメソッドは、run
とexit
です。パラメーターと戻り値はどちらも文字列でなければなりません。
run(path: String, timeout_seconds: int, arguments: Map): String
timeout_seconds
パラメーターは、実行のタイムアウトを制御します (0 はタイムアウトなしを意味します)。run
の呼び出しは、指定された時間内に終了しない場合、例外をスローします。Databricks が 10 分以上ダウンしている場合、ノートブックの実行は timeout_seconds
に関係なく失敗します。
arguments
パラメーターは、ターゲットノートブックのウィジェット値を設定します。具体的には、実行中のノートブックにA
というウィジェットがあり、run()
呼び出しの引数パラメーターの一部として("A": "B")
というキーと値のペアを渡した場合、ウィジェットA
の値を取得すると"B"
が返されます。ウィジェットの作成と操作については、「Databricksのウィジェット」を参照してください。
注
arguments
パラメーターは、ラテン文字(ASCII文字セット)のみを受け入れます。非ASCII文字を使用すると、エラーが返されます。dbutils.notebook
APIを使用して作成されたジョブは、30日以内に完了する必要があります。
ノートブック間で構造化データを受け渡す
このセクションでは、ノートブック間で構造化データを渡す方法を説明します。
# 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))
// 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))
エラーに対応する
このセクションでは、エラーに対応する方法について説明します。
# 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)
// 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)