Databricks Jobsを使ってみる

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

[Databricksにおけるジョブ管理](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)で説明されている、Databricksジョブ機能をウォークスルーします。[Jobs CLI](https://docs.databricks.com/dev-tools/cli/jobs-cli.html)を用いてローカルマシンからジョブを起動する方法、[Jobs REST API](https://docs.databricks.com/dev-tools/api/latest/jobs.html)も説明します。

# ジョブ用ノートブック

Databricksのジョブは、複数のノートブック、[データパイプライン](https://qiita.com/taka_yayoi/items/7fe8ed2c2f95fd53cc3d)を組み合わせることができるマルチタスクジョブです。ここではシンプルにするために一つのノートブックを用いてジョブを構成します。

ノートブックはPython、R、SQLなどを用いて実装することができますが、ここではPythonノートブックを使用します。データはサンプルデータとしてDatabricks環境に格納されているflowersデータセットを使用します。パラメーターとして花の種類を受け取り、指定された花の種類のデータを抽出し、書き込みを行う処理を行います。

パラメーターをノートブックで受け取るには、Databricks[ウィジェット](https://qiita.com/taka_yayoi/items/4df1fd723bdab9a99b08)を使用します。

```py:Python
flower = dbutils.widgets.get('flower')
```

これにより、ノートブックは`flower`というパラメーターを受け取れるようになります。このパラメーターを用いてデータを絞り込みます。

```py:Python
flowers = spark.read.format("delta").load("dbfs:/databricks-datasets/flowers/delta/").filter(f"label='{flower}'")
```

読み込んだデータをDelta形式で書き込みます。

```py:Python
import uuid

# 画像データは圧縮済みなのでParquetの圧縮をオフにします
spark.conf.set("spark.sql.parquet.compression.codec", "uncompressed")
path = "/tmp/flowers/" + str(uuid.uuid4()) # ユニークなIDでDBFS上の保存場所のパスを作成します
flowers.write.format("delta").mode("overwrite").save(path)
```

書き込んだデータを確認します。

```py:Python
saved_flowers = spark.read.format("delta").load(path)
display(saved_flowers)
```

![Screen Shot 2022-02-01 at 11.27.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d2bbb66f-cc90-1a14-c72e-fecbc4bd7afb.png)

https://github.com/taka-yayoi/public_repo/blob/main/jobs_walkthrough/Databricks%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E3%82%B5%E3%83%B3%E3%83%97%E3%83%AB.py

# ジョブの設定

上述のノートブックは、ノートブック単体でも処理できますが、ジョブとして登録することで定期実行を行うことや、後述するCLIを用いてジョブを起動することが可能となります。

1. サイドメニューの**Jobs**を開きます。
1. **Create Job**をクリックします。
![Screen Shot 2022-02-01 at 11.28.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d7bd87d8-cb98-b86a-a31c-fb715580f9ee.png)
1. ジョブは複数のタスクから構成されます。ここでは一つのみのタスクを作成します。
![Screen Shot 2022-02-01 at 11.29.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5caa341f-e84d-ffe1-4d44-db2025c0c56e.png)

  - **Task name**にタスク名を入力します。
  - **Type**で**Notebook**を選択し、上で作成したノートブックを選択します。
  - **Cluster**は**New Job Cluster**のままとします。
  - **Parameters**の**Add**をクリックし、**Key**に`flower`を入力します。
  - **Create**をクリックします。

以下のようにジョブが作成されます。
![Screen Shot 2022-02-01 at 11.33.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/235ce6ff-4df5-f14d-38b4-082a12892b1f.png)

右上の**Run now**をクリックして動作確認をすることもできます。**Schedule**で、スケジュールを組むこともできます。

ジョブ名の下の**Runs**タブで、実行結果を確認することができます。
![Screen Shot 2022-02-01 at 11.34.06.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bd8423cc-2e00-330d-734d-15b460497f94.png)

# Jobs CLIの活用

上で説明した手順はDatabricks上での設定・実行手順となりますが、ローカルマシン、別システムからジョブを起動したい場合には、Jobs CLIあるいはJobs APIを活用することができます。

Jobs CLIを使用するには、ローカルマシンにDatabricks CLIをインストールします。

```sh:Bash
pip install databricks-cli
```

CLIがDatabricksにアクセスするために、パーソナルアクセストークンを生成して、設定を行います。パーソナルアクセストークンは、サイドメニューの**Settings > User Settings**を開き、**Access Tokens**で**Generate New Token**をクリックします。トークンが表示されるのでコピーしておきます。

![](https://sajpstorage.blob.core.windows.net/demo20220129-pytorch-mlflow/PAT.png)

以下のコマンドを実行すると、アクセス先のDatabricksホスト名、上で生成したトークンを指定します。

```sh:Bash
databricks configure --token
```

![Screen Shot 2022-02-01 at 11.07.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/22de7e6c-356b-6a68-07cf-4ba181b1e749.png)

[Jobs CLI](https://docs.databricks.com/dev-tools/cli/jobs-cli.html)はDatabricks CLIに含まれています。

Jobs CLIを実行する際には、上で定義したジョブのIDが必要となります。ジョブの詳細画面でジョブIDを確認することができます。
![Screen Shot 2022-02-01 at 11.41.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cc302abb-38b3-c4bd-ebae-f652535e53c1.png)

```sh:Bash
databricks jobs run-now --job-id <ジョブID> --notebook-params '{"flower": "<パラメーターの値>"}'
```

上のジョブIDを指定し、花の種類が`roses`としてパラメーターを指定してジョブを実行するには以下を実行します。

```sh:Bash
databricks jobs run-now --job-id 157442 --notebook-params '{"flower": "roses"}'
```

# Jobs APIの活用

[Jobs API 2\.1](https://docs.databricks.com/dev-tools/api/latest/jobs.html)を用いることで、REST API経由でジョブを操作することができます。

REST APIを用いる際にもパーソナルアクセストークンが必要になります。HTTPリクエストヘッダーにBearerトークンとして埋め込みます。

以下ではノートブックでパーソナルアクセストークンを指定するケースを説明するので、[シークレット](https://qiita.com/taka_yayoi/items/338ef0c5394fe4eb87c0)を活用します。

```py:Python
# トークンは機密性の高い情報なので、ノートブックに直接記載しません。事前にCLIでシークレットとして登録しておいたトークンを呼び出します
token = dbutils.secrets.get("demo-token-takaaki.yayoi", "token")
os.environ["DATABRICKS_TOKEN"] = token
```

以下のような関数を用意しておきます。

```py:Python
import requests
import json

def call_api(job_id, flower):
  # 使用しているワークスペースのホスト名を指定します
  instance_id = '<Databricksホスト名>'

  api_version = '/api/2.0'
  api_command = '/jobs/run-now'
  url = f"https://{instance_id}{api_version}{api_command}"
  headers = {'Authorization': f'Bearer {os.environ.get("DATABRICKS_TOKEN")}'}

  params = {
    "job_id": job_id,
    "notebook_params": {
      "flower": flower
    }
  }
  #print(params)

  response = requests.post(
    url = url,
    headers = headers,
    json = params
  )

  print(json.dumps(json.loads(response.text), indent = 2))
```

以下のようにジョブIDとパラメーターを指定してREST APIを呼び出すと、ジョブが起動します。

```py:Python
call_api(157442, "sunflowers")
```

![Screen Shot 2022-02-01 at 12.59.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/30d9a778-51ff-86ca-0f79-dcbf2309d6e2.png)

https://github.com/taka-yayoi/public_repo/blob/main/jobs_walkthrough/Jobs%20REST%20API%E5%91%BC%E3%81%B3%E5%87%BA%E3%81%97%E3%82%B5%E3%83%B3%E3%83%97%E3%83%AB.py

### Databricks 無料トライアル

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