Databricksにおけるジョブ管理

[Jobs \| Databricks on AWS](https://docs.databricks.com/jobs.html#jobs_acl_user_guide) [2022/4/19時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

> [Databricksクイックスタートガイド](https://qiita.com/taka_yayoi/items/125231c126a602693610)のコンテンツです。

ジョブは非インタラクティブにDatabricksクラスターを実行するための手段です。例えば、抽出、変換、ロード(ETL)ワークロードをインタラクティブあるいはスケジュール処理として実行することができます。また、[ノートブックのUI](https://qiita.com/taka_yayoi/items/24a897cf40bba6d9e305)上でジョブをインタラクティブに実行することもできます。

UIあるいはCLI、Jobs APIを用いてジョブの作成実行が可能です。UIやAPIを用いて失敗あるいはキャンセルしたジョブを復旧し再実行することができます。UI、CLI、API、メールのアラートを用いてジョブの実行結果をモニタリングすることができます。本書では、UIを用いたジョブタスクの実行にフォーカスします。他の方法に関しては、[Jobs CLI](https://docs.databricks.com/dev-tools/cli/jobs-cli.html)と[Jobs CLI](https://docs.databricks.com/dev-tools/cli/jobs-cli.html)を参照ください。

ジョブは単一のタスク、あるいは複雑な依存関係を持つ大規模なマルチタスクワークフローから構成することができます。お使いのジョブ全てのタスクオーケストレーション、クラスター管理、モニタリング、エラーレポートはDatabrickによって管理されます。ジョブを即時実行したり、使いやすいスケジューリングシステムを用いて定期的にジョブを実行することができます。

タスクは、JAR、Databricksノートブック、Delta Live Tablesのパイプライン、そして、Scala、Java、Pythonで記述されたアプリケーションで実装することができます。レガシーなSpark Submitアプリケーションもサポートされています。タスク間の依存関係を指定することでタスクの実行順序を制御することができます。タスクを直列あるいは並列に実行するように設定することができます。以下の図では下のワークフローを示しています。

1. 生のクリックストリームデータを取り込み、レコードをセッション化するために変換
1. 注文データを取り込み、セッション化されたクリックストリームデータと結合し、分析データを準備
1. 準備されたデータから特徴量を抽出
1. 特徴量を永続化し、機械学習モデルをトレーニングするために並列でタスクを実行

![](https://docs.databricks.com/_images/multitask-jobs-workflow.png)

Databricksジョブを用いて最初のワークフローを作成するには、[クイックスタート](https://docs.databricks.com/data-engineering/jobs/jobs-quickstart.html)を参照してください。

:::note warn
**重要！**
- ジョブはData Science & Engineeringワークスペースでのみ作成できます。
- ワークスペースあたり同時実行ジョブ数は1000に制限されています。要求したリクエストが即時実行できない場合には、`429 Too Many Requests`レスポンスが返されます。
- ワークスペースで1時間あたりで作成できるジョブの数は5000("run now"と"runs submit"を含みます)に制限されています。この制限はREST APIで作成されるジョブ、ノートブックワークフローで作成されるジョブにも適用されます。
:::

# ジョブの作成

1. 以下のいずれかを実行します。
    - サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png) **Jobs**をクリックして、![](https://docs.databricks.com/_images/create-job.png)をクリックします。
    - サイドバーで![](https://docs.databricks.com/_images/create-icon.png)**Create**をクリックし、メニューから**Job**を選択します。

    **Task**タブでタスク作成ダイアログが表示されます。
![](https://docs.databricks.com/_images/create-task.png)

1. **Add a name for your job…** にジョブの名前を指定します。
1. **Task name**フィールドにタスク名を指定します。
1. 実行するタスクのタイプを指定します。**Type**ドロップダウンで、**Notebook**、**JAR**、**Spark Submit**、**Python**、**Pipeline**のいずれかを選択します。
    - **Notebook**: ファイルブラウザからノートブックを検索し、ノートブックを選択して**Confirm**をクリックします。
    - **JAR**: **Main class**を指定します。メインメソッドを含むクラスの完全なクラス名を使用します。例えば`org.apache.spark.examples.SparkPi`。そして、**Dependent Libraries**の下の**Add**をクリックしてタスク実行に必要なライブラリを追加します。これらのライブラリの一つにはメインクラスが含まれる必要があります。

        JARタスクの詳細に関しては[JARジョブ](#jarジョブ)を参照ください。
    - **Spark Submit**: **Parameters**テキストボックスでは、メインクラス、ライブラリJARへのパス、すべての引数をJSON配列型式で指定します。以下の例では、Apache Sparkの`DFSReadWriteTest`をspark-submitを実行するように設定しています。

        ```json:JSON
        ["--class","org.apache.spark.examples.DFSReadWriteTest","dbfs:/FileStore/libraries/spark_examples_2_12_3_1_1.jar","/dbfs/databricks-datasets/README.md","/FileStore/examples/output/"]
        ```

        > **重要!**
        > **spark-submit**タスクにはいくつかの制限があります。
        > - 新規のクラスターでのみspark-submitを実行できます。
        > - Spark-submitではクラスターのオートスケーリングはサポートしていません。オートスケーリングに関しては、[クラスターのオートスケーリング](https://docs.databricks.com/clusters/configure.html#autoscaling)を参照ください。
        > - Spark-submitでは[Databricks Utilities](https://docs.databricks.com/dev-tools/databricks-utils.html)をサポートしていません。Databricksユーティリティを使用する際には、JARタスクを利用してください。
       
    - **Python**: **Path**テキストボックスには、DBFSあるいはクラウドストレージ上のPythonスクリプトのURIを入力します。例えば、`dbfs:/FileStore/myscript.py`となります。
    - **Pipeline**: **Pipeline**ドロップダウンで、既存の[Delta Live Tables](https://qiita.com/taka_yayoi/items/7fe8ed2c2f95fd53cc3d)パイプラインを選択します。

    - **Python Wheel**: **Package name**テキストボックスで、インポートするパッケージを入力します。例えば、`myWheel-1.0-py2.py3-none-any.whl`のようなものです。**Entry Point**テキストボックスには、wheelを開始する際に呼び出す関数を指定します。タスクの実行に必要なライブラリを追加するには**Dependent Libraries**の下の**Add**をクリックします。 

1. タスクを実行するクラスターを設定します。**Cluster**ドロップダウンで**New Job Cluster**か**Existing All-Purpose Cluster**を選択します。タスクを実行するクラスターの設定、選択に関しては[クラスター設定](#クラスター設定のティップス)を参照ください。
    - **New Job Cluster:** **Cluster**ドロップダウンの**Edit**をクリックして、[クラスター設定](https://docs.databricks.com/clusters/configure.html#cluster-configurations)を完了します。
    - **Existing All-Purpose Cluster:** **Cluster**ドロップダウンにある既存のクラスターを選択します。新たなページでクラスターを表示したい場合には、クラスター名、説明の右にある![](https://docs.databricks.com/_images/external-link.png)アイコンをクリックします。

    タスクを実行するためのクラスターの選択と設定に関しては、[クラスター設定のティップス](#クラスター設定のティップス)を参照ください。

1. タスクにパラメーターを指定することができます。どのようなフォーマットでパラメーターを指定するのかはタスクのタイプに依存します。
    - **ノートブック**: **Add**をクリックし、タスクに渡すパラメーターをキーバリュー形式で指定します。手動でタスクを実行する際、[パラメータを指定してジョブを実行](#パラメータを指定してジョブを実行)でパラメーターを上書き、追加することができます。パラメーターはパラメーターのキーで指定される[ノートブックウィジェット](https://docs.databricks.com/notebooks/widgets.html)の値を設定します。パラメーターの値の一部として、限定的な動的な値のセットを渡すためには[タスクパラメーター変数](#パラメーター変数)を使用します。
    - **JAR**: パラメーターを指定するためにJSON配列の文字列を使用します。これらの文字列は、メインクラスのmainメソッドの引数として渡されます。[JARジョブのパラメーター設定](#jarジョブのパラメーター設定)を参照ください。
    - **Spark Submit**: パラメーターはJSON配列の文字列として指定します。[Apache Spark spark\-submit](https://spark.apache.org/docs/latest/submitting-applications.html)の書式に則り、JARパスの後のパラメーターはメインクラスのmainメソッドに引き渡されます。
    - **Python**: パラメーターを指定するためにJSON配列の文字列を使用します。これらの文字列はPythonの[argparse](https://docs.python.org/3/library/argparse.html)モジュールを用いてパースされます。
    - **Python Wheel**: **Parameters**ドロップダウンで、JSONフォーマットの配列としてパラメーターを指定するには __*args (string array)__ を選択し、パラメーターごとのキーバリューを指定するには __**kwargs (key-value pairs)__ を選択します。

1. **Dependent Libraries**、**Retry Policy**、**Timeouts**などの追加のオプションにアクセスするには、**Advanced Options**をクリックしてください。[タスクの編集](#タスクの編集)を参照ください。
1. **Create**をクリックします。
1. ジョブのスケジュールを設定する場合には、**Edit Schedule**をクリックします。[ジョブのスケジュール](#ジョブのスケジュール)を参照ください。
1. 同じジョブに対して複数の同時実行を行う際には、**Setting**タブをクリックし、**Maximum Concurrent Runs**に新たな値を入力します。[最大同時実行数](#最大同時実行数)を参照ください。
1. ジョブのイベントに対してメールのアラートを設定することができます。**Setting**タブをクリックし、**Alerts**セクションの**Add**をクリックします。[アラート](#ジョブアラート)を参照ください。
1. ジョブに対するアクセス権を設定する場合には**Setting**をクリックします。[ジョブのアクセスコントロール](#ジョブに対するアクセスコントロール)を参照ください。

他のタスクを追加するには、作成したタスクの下にある![](https://docs.databricks.com/_images/add-task.png)をクリックします。事前のタスクで**New Job Cluster**を設定した場合には、クラスター共有のオプションが提供されます。タスクを作成、編集する際に個別のクラスターを設定することもできます。タスクを実行するためのクラスターの選択と設定に関しては、[クラスター設定のティップス](#クラスター設定のティップス)をご覧ください。

# ジョブの実行

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Jobs**をクリックします。
1. ジョブを選択し、**Run**タブをクリックします。ジョブを即時実行するか後でジョブを実行するようにスケジュールすることができます。

## ジョブの即時実行

ジョブを即時実行するには、ジョブ名の隣の![](https://docs.databricks.com/_images/run-now-button.png)をクリックします。

:::note info
**ティップス**
**Run Now**をクリックすることでノートブックタスクのテストランを実行することができます。ノートブックに変更を加える必要がある場合には、ノートブックを編集後に再度**Run Now**をクリックすることで、最新バージョンのノートブックが実行されます。
:::

## 異なるパラメータを指定してジョブを実行

異なるパラメーター、既存のパラメーターの異なる値でジョブを再実行するために、**Run Now with Different Parameters**を利用できます。

1. **Run Now**の隣の![](https://docs.databricks.com/_images/down-caret-blue.png)をクリックし、**Run Now with Different Parameters**を選択するか、[アクティブなラン](#ジョブ実行の参照)の一覧で**Run Now with Different Parameters**をクリックします。タスクの種類に応じて新たなパラメーターを入力します。  
    - **ノートブック**: キーバリューのペア、あるいはJSONオブジェクトとしてパラメーターを入力します。[widgets](https://docs.databricks.com/notebooks/widgets.html)の値を設定するためにこのダイアログを利用できます。
    - **JAR**あるいは**spark-submit**: パラメーターのリストあるいはJSONドキュメントを指定します。実行の際にはデフォルトパラメーターと指定されたパラメーターがマージされます。キーを削除した場合には、デフォルトのパラメーターが使用されます。実行に際して[タスクパラメーター変数](#パラメーター変数)を追加することができます。

1. **Run**をクリックします。

## 成功しなかったジョブ実行のリペア

成功しなかったタスクと、全ての依存するタスクのサブセットのみを実行することで、失敗したあるいはキャンセルされたマルチタスクジョブをリペアすることができます。成功したタスクとそれらに依存するタスクは再実行されないので、この機能は成功しなかったジョブ実行からのリカバリーに必要な時間とリソースを削減します。

ジョブ実行をリペアする前にジョブやタスクの設定を変更することができます。例えば、ノートブックのパスやクラスター設定を変更した場合、更新されたノートブックやクラスター設定を用いてタスクが再実行されます。

**Task run details**ページで[全てのタスク実行の履歴](#タスク実行履歴の参照)を参照することができます。

:::note info
**注意**
- 1つ以上のタスクがジョブクラスターを共有している場合、リペアの実行では新規のジョブクラスターが作成されます。例えば、オリジナルの実行ではジョブクラスター`my_job_cluster`が使用された場合、リペアの実行では新規のジョブクラスター`my_job_cluster_v1`を使用するので、最初の実行といかなるリペアの実行によって使用されたクラスタとクラスター設定を容易に比較することができます。`my_job_cluster_v1`の設定は、現状の`my_job_cluster`と同じものになります。
- 2つ以上のタスクをオーケストレートするジョブでのみリペアはサポートされます。
- **Runs**タブに表示される**Duration**の値には、最初の実行の開始から最後のリペアが完了するまでの時間が含まれます。例えば、実行が2回失敗し、3回目の実行で成功した際には、durationには3つ全ての実行の時間が含まれます。
:::

失敗したジョブをリペアするには、以下を実施します。

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Jobs**をクリックします。
1. **Name**カラムのジョブ名をクリックします。**Runs**タブにはアクティブなランと失敗を含む完了したランが表示されます。
1. **Completed Runs (past 60 days)**テーブルの**Start time**カラムにある失敗したランのリンクをクリックします。**Job run details**ページが表示されます。
1. **Repair run**をクリックします。**Repair job run**ダイアログが表示され、再実行される失敗した全てのタスクと依存する全てのタスクが一覧表示されます。
1. リペアするタスクのパラメータの追加、修正を行うには、**Repair job run**ダイアログにパラメータを入力します。**Repair job run**ダイアログに入力された値は既存の値を上書きします。後続のリペアの実行の際には、**Repair job run**ダイアログのキーバリューをクリアすることで、パラメーターをオリジナルの値に戻すことができます。
1. **Repair job run**ダイアログ**Repair run**をクリックします。

## タスク実行履歴の参照

成功、失敗を含むタスクの実行履歴を参照するには、

1. **Job run details**ページのタスクをクリックします。**Task run details**ページが表示されます。
1. run historyドロップダウンからタスクを選択します。


## ジョブのスケジュール

ジョブのスケジュールを定義するには以下を実行します：

1. **Job details**パネルの**Edit Schedule**をクリックして、**Schedule Type**を**Scheduled**を選択します。
1. 期間、開始時刻、タイムゾーンを指定します[Quartz Cron Syntax](http://www.quartz-scheduler.org/documentation/quartz-2.3.0/tutorials/crontrigger.html)に従ってスケジュールを表示、編集する場合には**Show Cron Syntax**のチェックボックスを選択します。

    > **注意**
    > - Databricksは、cron表現の秒の設定に関係なく、ジョブのスケジュールによって実行される後続の処理に最低10秒のインターバルを設けます。
    > - セービングタイムのあるタイムゾーンあるいはUTC時間を選択できます。セービングタイムのあるタイムゾーンを選択した場合には、[セービングタイムの開始および終了時](https://www.quartz-scheduler.org/documentation/2.3.1-SNAPSHOT/faq.html#questions-about-jobs)に時間単位のジョブがスキップされたり、実行されないように見える現象が起きます。
    > - ジョブスケジューラーは、Sparkのバッチインタフェース同様、低レーテンシーのジョブ向きではありません。ネットワークやクラウドの問題によって、ジョブ実行が最大数分遅れる可能性があります。このような場合、スケジュールされたジョブはサービスが利用可能になるや否や実行されます。

1. **Save**をクリックします。

## ジョブの一時停止、再開

以下のいずれを実行してジョブを一時停止します。

- **Job details**パネルの**Pause**をクリックします。
- **Job details**パネルの**Edit schedule**をクリックし、**Schedule Type**を **Manual (Paused)** に設定します。

一時停止したジョブスケジュールを再開するには、**Schedule Type**で**Scheduled**を選択します。

# ジョブの参照

サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)をクリックするとジョブの一覧が表示されます。ジョブのページでは、定義されたジョブ、クラスター定義、スケジュール、前回の実行結果が表示されます。
![](https://docs.databricks.com/_images/job-list.png)

ジョブの一覧では以下の方法でフィルタリングできます。

- キーワード指定
- 所有しているジョブのみを表示
- アクセス可能なジョブを表示。このフィルターを利用するには[Jobs access control](https://docs.databricks.com/security/access-control/jobs-acl.html)が有効になっている必要があります。

カラムのヘッダーをクリックすることでジョブ一覧を並び替える(降順、昇順)ことができます。デフォルトの並び替えはジョブ名の昇順です。

# ジョブ実行の参照

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Jobs**をクリックします。
1. ジョブのページで、**Name**列のジョブ名をクリックします。**Runs**タブにはアクティブなジョブ実行と完了したジョブが表示されます。
![](https://docs.databricks.com/_images/job-details.png)

マトリクスビューに切り替えるには**Matrix**をクリックします。マトリクスビューでは個々のタスクを含むジョブの実行履歴が表示されます。

マトリクスビューの**Job Runs**行には、実行に要した合計時間と実行の状態が表示されます。開始時間、期間、ステータスを含む実行の詳細を参照するには、**Job Runs**行のバーの上にマウスカーソルを移動します。

**Tasks**行のそれぞれのセルには、タスクとそれに対応するタスクのステータスが表示されます。開始時間、期間、クラスター、ステータスを含む個々のタスクの詳細を参照するにはタスクのセルの上にマウスカーソルを移動します。

ジョブの実行とタスク実行のバーは実行ステータスを表す色が割り当てられます。成功した実行は緑、失敗した実行は赤、スキップされた実行はピンクになります。それぞれのジョブ実行、タスク実行のバーの高さは実行時間の可視化となります。
![](https://docs.databricks.com/_images/job-details-matrix.png)

Databricksではジョブ実行履歴を60日間保持します。ジョブの実行結果を保存したい場合には、有効期限が切れる前に実行結果をエクスポートすることをお勧めします。詳細は[ジョブ実行結果のエクスポート](#ジョブ実行結果のエクスポート)を参照ください。

# ジョブ実行詳細の参照

ジョブ実行詳細ページには、ジョブにおけるそれぞれのタスクの実行成否の情報が含まれます。ジョブ実行詳細にアクセスするには、ジョブの**Runs**タブをクリックします。

**Runs**タブでジョブ実行詳細を参照するには以下の操作を行います：

1. **Completed Runs (past 60 days)**テーブルの**Run**列の**View Details**リンクをクリックします。
1. 以下を参照するためにタスクをクリックします：
    - タスクを実行したクラスター
    - タスクのSpark UI
    - タスクのログ
    - タスクのメトリクス

ジョブの**Runs**タブに戻るには**Job ID**の値をクリックします。**Job run ID**の値をクリックすると、ジョブ実行詳細画面に戻ります。

# 最近のジョブ実行の参照

Apache AirflowやAzure Data Factoryによって起動されたジョブを含む、ワークスペースでアクセスできる全てのジョブの最近の実行結果、稼働中のジョブの一覧を参照することができます。最近のジョブ実行の一覧を表示するには以下を実行します。

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Jobs**をクリックします。ジョブの一覧が表示されます。
1. **Job runs**タブをクリックします。ジョブ実行の一覧が表示されます。

**Job runs**リストには以下が表示されます。

- 実行の開始時刻
- 実行に関連づけられているジョブの名前
- ジョブを実行するユーザー名
- ジョブ実行がスケジュールによるものか、APIリクエストか、手動によって起動されたのか
- 現在実行中のジョブの経過時間、あるいは、完了した実行の総実行時間
- 実行のステータス。`Pending`、`Running`、`Skipped`、`Succeeded`、`Failed`、`Terminating`、`Terminated`、`Internal Error`、`Timed Out`、`Canceled`、`Canceling`、`Waiting for Retry`のいずれかとなります。

[ジョブ実行詳細](#ジョブ実行詳細の参照)を参照するには、実行の`Start time`カラムをクリックします。ジョブの操作を参照するには、**Job**カラムのジョブ名をクリックします。

# ジョブ実行結果のエクスポート

全てのジョブタイプでノートブックの実行結果と実行ログをエクスポートすることができます。

## ノートブック実行結果のエクスポート

実行結果をエクスポートすることでジョブ実行結果を保存することができます。ノートブックジョブの実行においては、結果が埋め込まれたノートブックを[エクスポート](https://qiita.com/taka_yayoi/items/c306161906d6d34e8bd5#%E3%83%8E%E3%83%BC%E3%83%88%E3%83%96%E3%83%83%E3%82%AF%E3%81%AE%E3%82%A8%E3%82%AF%E3%82%B9%E3%83%9D%E3%83%BC%E3%83%88)でき、あとでDatabricksワークスペースに[インポート](https://qiita.com/taka_yayoi/items/c306161906d6d34e8bd5#%E3%83%8E%E3%83%BC%E3%83%88%E3%83%96%E3%83%83%E3%82%AF%E3%81%AE%E3%82%A4%E3%83%B3%E3%83%9D%E3%83%BC%E3%83%88)することが可能です。

単一のタスクのジョブの実行結果ノートブックをエクスポートするには以下を実行します。

1. ジョブ詳細ページで、**Completed Runs (past 60 days)**テーブルの**Run**列のジョブ実行名の**View Details**リンクをクリックします。
1. **Export to HTML**をクリックします。

複数のタスクのジョブの実行結果ノートブックをエクスポートするには以下を実行します。

1. ジョブ詳細ページで、**Completed Runs (past 60 days)**テーブルの**Run**列のジョブ実行名の**View Details**リンクをクリックします。
1. エクスポートするノートブックタスクをクリックします。
1. **Export to HTML**をクリックします。

## ジョブ実行ログのエクスポート

ジョブ実行のログをエクスポートすることも可能です。このプロセスを自動化するために、Job API経由でログをDBFSやS3に自動で出力するようにジョブを設定できます。Jobs APIにおける[新規ジョブ作成](https://docs.databricks.com/dev-tools/api/latest/jobs.html)オペレーション(`POST /jobs/create`)に引き渡すリクエストボディの`new_cluster.cluster_log_conf`オブジェクトを参照して下さい。

# ジョブの編集

いくつかのオプションはジョブに対して適用できますが、他のオプションは個々の[タスク](#タスクの編集)に対して適用されます。例えば、最大同時実行数はジョブのみに設定できますが、パラメータはそれぞれのタスクに対して定義される必要があります。

ジョブの設定を変更するには：

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Jobs**をクリックします。
1. **Name**カラムのジョブ名をクリックします。

サイドパネルに**Job details**が表示されます。[スケジュール](#ジョブのスケジュール)、クラスター設定、アラート、最大同時実行数を変更することができます。また、[ジョブアクセスコントロール](#ジョブに対するアクセスコントロール)が有効化されている場合には、ジョブのアクセス権を編集することができます。

## クラスター

クラスターに紐づけられているタスクを参照するには、サイドパネルのクラスターの上にマウスカーソルを移動します。関連づけられている全てのタスクに対するクラスターの設定を変更するには、クラスターの下の**Configure**をクリックします。関連づけられている全てのタスクに対して新規のクラスターを設定するには、クラスターの下の**Swap**をクリックします。

## 最大同時実行数

並列で実行可能な最大数です。新たなジョブを実行する際、アクティブな実行数が最大数に達している場合には、Databricksは処理をスキップします。同じジョブを同時実行したい場合にはデフォルトの1より大きい値を指定します。これは例えば、頻度の高いスケジュール処理を実行する際、連続した処理が互いに被っても問題がない場合、異なる入力パラメーターで複数処理を実行したい場合に有効です。

## ジョブアラート

ジョブの開始、完了、失敗時に通知を行う一つ以上のメールアドレスを追加することができます。

1. **Edit alerts**をクリックします。
1. **Add**をクリックします。
1. 通知先のメールアドレスを入力し、アドレスに通知したいアラートタイプのチェックボックスを選択します。
1. 別のメールアドレスを追加するには**Add**をクリックします。
1. スキップされたジョブ実行に対するアラートを受け取りたくない場合にはチェックボックスをクリックします。
1. **Confirm**をクリックします。

![](https://docs.databricks.com/_images/job-alerts.png)

以下の通知ツールとメールの通知をインテグレーションすることができます。

- [PagerDuty](https://www.pagerduty.com/docs/guides/email-integration-guide)
- [Slack](https://get.slack.help/hc/articles/206819278-Send-emails-to-Slack)
- [Amazon SESとSNS](https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-setting-up.html)

## ジョブに対するアクセスコントロール

ジョブのアクセス制御によって、ジョブのオーナー、管理者はジョブに対してきめ細かいアクセス制御を行えます。ジョブのアクセス制御によって、ジョブのオーナーは他のどのユーザー、グループがジョブの結果を参照できるのかを制御できます。また、誰がジョブの管理(即時実行、ジョブのキャンセルの権限)をできるのかも指定できます。

詳細は[Jobs access control](https://docs.databricks.com/security/access-control/jobs-acl.html)を参照ください。

# タスクの編集

タスク設定を設定するには、

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Jobs**をクリックします。
1. **Name**カラムのジョブ名をクリックします。
1. **Tasks**タブをクリックします。

## タスクの依存関係

**Depends on**ドロップダウンを用いてジョブ内のタスクの実行順序を定義することができます。ジョブの1つ以上のタスクをこのフィールドに設定することができます。
![](https://docs.databricks.com/_images/task-dependencies.png)

:::note info
**注意**
タスクが一つのジョブの場合、**Depends on**は表示されません。
:::

タスクの依存関係を設定することで、ジョブスケジューラにおける実行順序を表現する標準的な方法である有向非巡回グラフ(DAG)を作成します。例えば、4つのタスクから構成される以下のジョブを考えてみます。
![](https://docs.databricks.com/_images/task-dependencies-diagram.png)

- Task1はルートタスクであり他のタスクには依存しない
- Task2とTask3は、Task1の完了に依存する
- 最後に、Task4はTask2とTask3の完了に依存する

Databricksは後段のタスクを実行する前に、最大限の並列性で前段のタスクを実行します。以下の図では、上のタスクの処理順序を示しています。
![](https://docs.databricks.com/_images/task-dependencies-flow.png)

## タスク設定オプション

それぞれのタスクには以下の設定を行えます。

### クラスター

タスクを実行するクラスターを設定するには、**Cluster**ドロップダウンをクリックします。共有されているジョブクラスターを編集することはできますが、他のタスクで使用されている共有クラスターを削除することはできません。

タスクを実行するクラスターの設定、選択に関しては[クラスター設定のティップス](#クラスター設定)を参照してください。

### 依存ライブラリ

タスク実行前に依存ライブラリがクラスターにインストールされます。ジョブの実行前に確実にインストールされるようにすべての依存ライブラリを指定する必要があります。

依存ライブラリを追加するには、**Advanced options**をクリックし、**Add Dependent Libraries**を選択しライブラリ選択画面を開きます。ライブラリを指定するには[ライブラリの依存関係](#ライブラリの依存関係)の推奨事項に従ってください。

:::note warn
**重要！**
[すべてのクラスターに対する自動インストール設定](https://docs.databricks.com/libraries/cluster-libraries.html#install-workspace-libraries)を行なった、あるいはライブラリがインストール済みの既存の停止クラスターを選択した場合には、ライブラリのインストール完了を待たずにジョブは実行されます。ジョブが特定のライブラリを必要とする場合には、Dependent Librariesフィールドでライブラリをジョブにアタッチする必要があります。
:::

### パラメーター変数

タスクのパラメーターの一部として、ジョブのタスクにテンプレート化された変数を渡すことができます。これらの変数はジョブ実行時に適切な値で置換されます。ランIDやジョブ開始時刻などジョブ実行に関するコンテキストを渡すためにタスクパラメーター変数を利用できます。

ジョブの実行の際、二重の中括弧で囲まれたタスクパラメーター変数は置換され、値の一部としてオプション文字列に追加されます。例えば、ジョブID6のジョブに対して、`MyJobId`を`my-job-6`で置換するには、以下のタスクパラメーターを追加します。

```json
{
  "MyJobID": "my-job-{{job_id}}"
}
```

二重の中括弧の中身は評価されませんので、この中で計算や関数を使用できません。括弧内の空白は除外されないので`{{  job_id  }}`は評価されません。

以下のタスクパラメーター変数がサポートされています。

| 変数| 説明 | 例 |
|:--|:--|:--|
| {{job_id}}  | ジョブに割り当てられた一意の識別子  | 281  |
| {{run_id}}  |ジョブ実行に割り当てられた一意の識別子  | 4785  |
| {{start_date}}  | ジョブが開始した日付。書式はyyyy-MM-dd。UTCタイムゾーン  |  2021-02-15 |
| {{start_time}}  | クラスターが作成され実行可能状態となった、ジョブ実行の開始時点のタイムスタンプ。書式は`System.currentTimeMillis()`で返されるUNIXエポック以来のミリ秒。UTCタイムゾーン  | 1551622063030  |
| {{task_retry_count}}  |最初の実行で失敗した場合のリトライ回数。最初の実行は0で以降リトライのたびに増える   |   |
| {{parent_run_id}}  |複数のタスクを持つジョブの実行に割り当てられる一位の識別子   |  3447835 |
| {{task_key}}  |複数タスクのジョブの一部であるタスクに割り当てられるタスクの一意な名前   | “clean_raw_data”  |

これらの変数は[ジョブの作成](#ジョブの作成)、[タスクの編集](#タスクの編集)、[パラメータを指定してジョブを実行](#パラメータを指定してジョブを実行)の際に設定できます。

### タイムアウト

タスクの終了に要する時間の最大値。この時間内にタスクが完了しなかった場合には、ステータスが**Timed Out**になります。

### リトライ

実行に失敗したジョブを何回リトライするのかのポリシーとなります。デフォルトでは失敗したタスクのリトライを行いません。タスクのリトライを設定するには、**Advanced options**をクリックし、**Edit Retry Policy**を選択します。

:::note info
**注意**
**Timeout**と**Retries**を両方設定した場合、それぞれのリトライにタイムアウトが適用されます。
:::

# ジョブのクローン

既存のジョブ設定をクローンすることで容易に新たなジョブを作成できます。ジョブをクローンすることで、ジョブID以外は全く同一のコピーを作成できます。ジョブのページでジョブの名前の隣にある **More …** をクリックし、ドロップダウンメニューから**Clone**を選択します。

# タスクのクローン

既存タスクをクローンすることでクイックに新規のタスクを作成することができます。

1. ジョブページで**Tasks**タブをクリックします。
1. クローンするタスクを選択します。
1. ![](https://docs.databricks.com/_images/jobs-vertical-ellipsis.png)をクリックし**Clone task**を選択します。

# ジョブの削除

ジョブページでジョブ名をクリック、ジョブ名の隣の**More …**をクリックし、ドロップダウンメニューから**Delete**を選択します。

# タスクの削除

1. **Tasks**タブをクリックします。
1. 削除するタスクを選択します。
1. ![](https://docs.databricks.com/_images/jobs-vertical-ellipsis.png)をクリックし**Remove task**を選択します。

# タスクパスのコピー

タスクのパス、例えばノートブックパスをコピーするには以下を実行します。

1. **Tasks**タブをクリックします。
1. コピーするパスを含むタスクを選択します。
1. クリップボードにコピーしたいタスクの隣にある![](https://docs.databricks.com/_images/copy-icon1.png)をクリックします。

# ベストプラクティス

## クラスター設定のティップス

クラスター設定はジョブを本格運用する際に重要です。以下では、ジョブクラスターをどのように選択し設定するのかに関して一般的なガイダンス、特定のジョブ、タスクにおける推奨事項を説明します。

### 共有ジョブクラスターの活用

複数タスクをオーケストレーションするジョブにおいて使用リソースを最適化するには、共有ジョブクラスターを使用します。共有ジョブクラスターを使うことで、同じジョブの複数のタスクでクラスターを再利用することができます。ジョブの一部である全てのタスクを実行する際に一つのジョブクラスターを使用することができますし、特定のワークロードに対して最適化された複数のジョブクラスターを使用することもできます。

1. タスクを作成する際に、**New Job Clusters**を選択し、[クラスター設定](https://qiita.com/taka_yayoi/items/8d951b660cd87c6c5f18)を完了します。
1. ジョブにタスクを追加する際に上の新規クラスターを選択するか、新たなジョブクラスターを作成します。**New Job Clusters**を選択して設定したクラスターは、ジョブの他のタスクで利用することができます。

共有ジョブクラスターは単一のジョブ実行のスコープとなっており、他のジョブや同じジョブの他の実行で使用することはできません。

共有ジョブクラスター設定でライブラリを宣言することはできません。タスクの設定で依存ライブラリを追加する必要があります。

### ジョブに適したクラスタータイプの選択

- **New Job Clusters**は、ジョブやタスク実行専用のクラスターです。共有上部クラスターは、クラスターを使用する最初のタスクによって作成、起動され、クラスターを使用する最後のタスクが完了すると停止されます。アイドル状態の際にはクラスターは停止されませんが、クラスターを使用する全てのタスクが完了すると停止されます。共有ジョブクラスターが失敗、あるいは全てのタスクが完了する前に停止された場合、新規クラスターが作成されます。単一のタスクをスコープとしたクラスターが作成され、タスクが開始する際に起動され、タスクが完了すると停止されます。プロダクション環境では、新規の共有クラスター、タスクスコープのクラスターを作成し、それぞれのジョブやタスクが完全に分離された環境で実行されるようにすることをお勧めします。
- 新規のジョブクラスターでジョブを実行する際、このジョブはデータエンジニアリング(タスク)ワークロードとみなされ、ジョブワークロードとして課金されます。既存のall-purposeクラスターでジョブを実行した際には、このジョブはデータアナリティクス(オールパーパス)ワークロードとみなされ、オールパーパスワークロードとして課金されます。
- 既存の停止クラスターを選択し、ジョブのオーナーが**Can Restart**[権限](https://docs.databricks.com/security/access-control/cluster-acl.html)を持っているのであれば、ジョブのスケジュール実行の際にはDatabricksがクラスターを起動します。
- 一定期間で[ダッシュボード](https://docs.databricks.com/notebooks/dashboards.html)を更新するようなタスクにおいては、既存のオールパーパスクラスターは好適と言えます。

### クラスター起動時間削減のためにプールを活用

新規のジョブクラスターの起動時間を短縮するためには、[プール](https://qiita.com/taka_yayoi/items/919acd7ef9decf0f61e6)を作成してジョブクラスターがプールを利用するように設定します。

### 自動AZ

自動アベイラビリティゾーン(Auto-AZ)を活用するには、[Clusters API](https://docs.databricks.com/dev-tools/api/latest/clusters.html#clusterawsattributes)を使用して、`awsattributes.zone_id = "auto"`を設定してAuto-AZを有効化する必要があります。[Availability zones](https://docs.databricks.com/clusters/configure.html#availability-zones)を参照ください。

## ノートブックジョブのティップス

ノートブックのセルのアウトプットの合計(すべてのノートブックのセルのアウトプットを結合したもの)には20MBのサイズ制限があります。加えて、それぞれのセルのアウトプットサイズには8MBの制限があります。トータルのアウトプットが20MBを超えた場合、個々のセルのアウトプットが8MBを超えた場合にはジョブ実行はキャンセルされ失敗とみなされます。

制限を超えそうなセルを見つける場合には、オールパーパスクラスターでノートブックを実行し、[notebook autosave technique](https://kb.databricks.com/notebooks/notebook-autosave.html)を活用します。

## ストリーミングタスク

Sparkのストリーミングジョブで1より大きい最大同時実行数を設定してはいけません。また、ストリーミングジョブでは、cronのスケジュールで`"* * * * * *"`(毎秒)、あるいは毎分を設定する必要があります。さらに、ストリーミングジョブでリトライを有効化してはいけません。

ストリーミングジョブは継続的に実行されるので、常にジョブの最後のタスクとして設定されるべきです。

## JARジョブ

JARジョブを実行する際には以下に注意してください。

### 出力サイズの制限

:::note info
**注意**
Databricksランタイム6.3以降で利用できます。
:::

標準出力に吐き出されるログ出力などのジョブのアウトプットは20MBのサイズ制限があります。これを超えた場合、ジョブの実行はキャンセルされ失敗とみなされます。

この制限を超えないようにするには、Spark設定の`spark.databricks.driver.disableScalaOutput`を`true`にすることで、ドライバーからDatabricksに返却される標準出力を抑制します。デフォルトでは`false`に設定されています。このフラグはScala JARジョブとScalaノートブックのセルの出力を制御します。フラグが有効化されるとSparkはジョブの実行結果をクライアントに返却しません。このフラグはクラスターのログファイルに書き込まれるデータには影響を与えません。この設定はノートブックの結果を表示しなくしてしまうため、JARジョブのジョブクラスターにおいてのみ設定することをお勧めします。

### 共有`SparkContext`の活用

Databricksはマネージドサービスであるため、あなたのApache Sparkジョブの実行を適切に行うためにはコードの変更が必要になるケースがあります。JARのジョブプログラムは、`SparkContext`を取得するために共有の`SparkContext` APIを使用する必要があります。Databricksが`SparkContext`を初期化するので、プログラムによる`new SparkContext()`は失敗します。`SparkContext`を取得するためにはDatabricksが作成する共有`SparkContext`を使用してください。

```Scala:Scala
val goodSparkContext = SparkContext.getOrCreate()
val goodSparkSession = SparkSession.builder().getOrCreate()
```

共有`SparkContext`を使用する際に避けるべきメソッドがいくつかあります。

- `SparkContext.stop()`を呼び出さないでください。
- `Main`プログラムの最後で`System.exit(0)`や`sc.stop()`を呼び出さないでください。予期しない動作を引き起こします。

### ジョブのクリーンアップのために`try-finally`を活用

以下の二つの部分から構成されるJARを考えてみます。」

- ジョブの主要な箇所を含む`jobBody()`
- `jobBody()`の正常終了後、例外が呼び出された際に呼び出される`jobCleanup()`

例として、`jobBody()`がテーブルを作成し、`jobCleanup()`が作成されたテーブルを削除するものとします。

```Scala:Scala
try {
  jobBody()
} finally {
  jobCleanup()
}
```

`sys.addShutdownHook(jobCleanup)`や以下のようなコードでクリーンアップをすべきではありません。

```Scala:Scala
val cleanupThread = new Thread { override def run = jobCleanup() }
Runtime.getRuntime.addShutdownHook(cleanupThread)
```

DatabricksによってSparkのコンテナーのライフタイムが管理されるため、shutdownのフックは正常に実行されません。

### JARジョブのパラメーター設定

JSON文字列の配列でJARジョブのパラメーターを指定できます。詳細は、[SparkJarTask](https://docs.databricks.com/dev-tools/api/latest/jobs.html#jobssparkjartask)を参照してください。パラメーターにアクセスするには、`main`関数に渡される`String`の配列を参照します。

## ライブラリの依存関係

Sparkのドライバーには上書き不可の依存ライブラリが存在します。このライブラリはあなたが使用する競合ライブラリよりも高い優先度を持ちます。

ドライバーの依存ライブラリの完全なリストを取得するには、同じSparkのバージョン(あるいは検証したいドライバーのクラスター)のクラスターにアタッチされたノートブックで以下のコマンドを実行します。

```bash
%sh
ls /databricks/jars
```

### ライブラリ依存関係の管理

ジョブのJARを作成する際にライブラリの依存関係に対応する良い方法としては、SparkとHadoopを`前提の(provided)`依存関係として一覧にすることです。Mavenにおいては、以下のコードでSpark and/or Hadoopを前提の依存関係として追加します。

```XML
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.3.0</version>
  <scope>provided</scope>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-core</artifactId>
  <version>1.2.1</version>
  <scope>provided</scope>
</dependency>
```

`sbt`では以下のコードでSparkとHadoopの依存関係を追加します。

```Scala:Scala
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.3.0" % "provided"
libraryDependencies += "org.apache.hadoop" %% "hadoop-core" % "1.2.1" % "provided"
```

:::note
**ティップス**
実行環境のバージョンに基づいて依存関係に対する適切なScalaのバージョンを指定してください。
:::

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)