Databricksジョブのウォークスルー

これも一年以上の記事なので、再度ウォークスルーします。

https://qiita.com/taka_yayoi/items/10194fd3b44e9575de26

Databricksのジョブはマルチタスクジョブなので、個々の処理を**タスク**として定義し、処理フローを組み立ててジョブとして構成することができます。タスクには以下のようなものを定義できます。

- ノートブック
- Delta Live Tablesパイプライン
- SQLクエリー
- ダッシュボード
- 別のジョブ
- [if/else](https://qiita.com/taka_yayoi/items/4c333568bfd6c80cf2b7)(日本リージョンでも利用できるようになりました！)

ここではシンプルな例として、2つのノートブックタスクから構成されるジョブを構築して実行します。

# ノートブックの作成

## データ取得ノートブック

まず、COVID-19感染者データを取得してテーブルに保存するノートブックを作成します。

```py
import re

# Username を取得
user_id = spark.sql('select current_user() as user').collect()[0]['user']
# Username の英数字以外を除去し、全て小文字化。Username をファイルパスやデータベース名の一部で使用可能にするため。
username = re.sub('[^A-Za-z0-9]+', '', user_id).lower()
print(username)
```

```py
import pandas as pd

# pandasを使ってURLから直接データを取得します
pdf = pd.read_csv("https://covid19.mhlw.go.jp/public/opendata/newly_confirmed_cases_daily.csv")
# データを表示します
display(pdf)
```

```py
# 縦長に変換
# id_vars: Dateをidとして、変換後のデータフレームにそのまま残します
# var_name: variable変数の列名をPrefectureにします
# value_name: value_name変数はCaseとします
long_pdf = pdf.melt(id_vars=["Date"], var_name="Prefecture", value_name="Cases")
display(long_pdf)
```

ジョブで使用するテーブル名をジョブで使う他のタスクから参照できるように、[タスクバリュー](https://qiita.com/taka_yayoi/items/6f41209c112dccf141df)を活用します。`dbutils.jobs.taskValues.set`でテーブル名をタスクバリューとして設定します。

```py
# pandasデータフレームをSparkデータフレームに変換します
sdf = spark.createDataFrame(long_pdf)

# 他のユーザーと重複しないようにテーブル名にユーザー名を埋め込みます
table_name = f"takaakiyayoi_catalog.quickstart_schema.{username}_covid_table"

# タスクバリューとしてジョブに含める他のノートブックから参照できるようにします
dbutils.jobs.taskValues.set(key="table_name", value=table_name)

# テーブルとして保存します
sdf.write.format("delta").mode("overwrite").saveAsTable(table_name)
```

## 可視化ノートブック

テーブルのデータからグラフを作成するノートブックを作成します。`dbutils.jobs.taskValues.get`でタスクバリューとして設定されているテーブル名を取得します。

```py
import pyspark.pandas as ps

# タスクバリューからテーブル名を取得します
table_name = dbutils.jobs.taskValues.get("get_data", "table_name")
print(table_name)

sdf = spark.read.table(table_name)
psdf = ps.DataFrame(sdf)
#display(psdf)
```

```py
%matplotlib inline
```

```py
# 東京のデータを可視化
tokyo_psdf = psdf[psdf["Prefecture"]=="Tokyo"][["Date", "Cases"]].set_index("Date")
tokyo_psdf.plot.line()
```

このように、2つのノートブックを作成します。
![Screenshot 2023-11-18 at 14.43.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ca564ae2-27aa-e17d-8b27-2c99a0b72f0a.png)

# ジョブの作成

サイドメニューから**ワークフロー**を選択します。ワークフローとジョブの関係についてはこちらをご覧ください。

https://qiita.com/taka_yayoi/items/e881769270f0ec0b7d06

**ジョブを作成**をクリックします。
![Screenshot 2023-11-18 at 14.45.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ea28e060-041b-d0b1-2ae4-e3739d7323c0.png)

ジョブに名前をつけます。
![Screenshot 2023-11-18 at 14.46.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/182ed9b2-47b1-431e-b336-c5e4d7230b80.png)
![Screenshot 2023-11-18 at 14.48.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f895f772-bb54-579b-f24b-a1fa06db4ec4.png)

タスクを組み合わせてジョブを構成していきます。1つ目のタスクを作ります。名前は`get_data`にします。なお、タスク名にスペースを含めることはできません。

**種類**はノートブックで、**パス**をクリックして上で作成したノートブックを選択します。
![Screenshot 2023-11-18 at 14.49.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f3a5523e-35e1-d6dd-91f6-c63b51988b8d.png)
![Screenshot 2023-11-18 at 14.49.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/06257ae4-ad51-a33a-1755-b219b2829fd8.png)
これで、一つ目のタスクができたので**作成**をクリックします。
![Screenshot 2023-11-18 at 14.49.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4874c381-4b78-785b-4679-9b567adfd8f4.png)

タスクを追加するので、**タスクを追加**をクリックします。
![Screenshot 2023-11-18 at 14.51.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a38ebdd8-af6e-dcd7-84ec-d5995bc8b57e.png)
**ノートブック**を選択して、後続のタスクを追加します。
![Screenshot 2023-11-18 at 14.51.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d51c2c21-d96f-d8a1-6f7e-cf79dbacb1a2.png)
![Screenshot 2023-11-18 at 14.52.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8201c5aa-6766-f83b-2d7f-1f5f215205a7.png)
2つ目のタスク名は`visualize`にします。先ほどと同様に可視化のノートブックを選択します。
![Screenshot 2023-11-18 at 14.52.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/07ec1731-c8fc-ff10-6313-9e46aa5ef3ac.png)
**タスクを作成**をクリックします。
![Screenshot 2023-11-18 at 14.53.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f3781920-ca61-88ab-25c9-fae7e138b78c.png)

これで、2つのタスクから構成されるジョブが完成しました！
![Screenshot 2023-11-18 at 14.53.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5db65a84-d877-5f05-e3c8-1c18da87ef2c.png)

# ジョブにおけるクラスター

https://docs.databricks.com/ja/clusters/index.html

Databricksでプログラムの処理を実行する際には、クラスターが必要となります。ジョブにおいても同様です。ただ、ジョブの場合にはジョブクラスターを活用することができます。従来のall purposeクラスターは明示的に停止するか自動停止を活用しない限り稼働し続けますが、ジョブクラスターはジョブの完了後に自動で停止します。本格運用する際にはジョブクラスターの利用をお勧めします。単価もお安くなっています。

ただ、ジョブのデバッグを行う際にジョブクラスターを使うと、実行の都度クラスターが停止するので生産性が上がりません。この場合には、ジョブでもall purposeクラスターを使う方が合理的といえます。

ジョブの画面で**クラスター**セクションにある**切り替え**ボタンを押して、all purposeクラスターを使用することができます。繰り返しになりますが、本格稼働させる際にはジョブクラスターを選択することを忘れないようにしてください。
![Screenshot 2023-11-18 at 15.09.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1e6c5c00-d7df-d420-c944-5917756efaae.png)
![Screenshot 2023-11-18 at 15.09.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/42051778-ff1d-5e20-3f87-139edc55e838.png)
![Screenshot 2023-11-18 at 15.10.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b0fc5736-9dd5-b473-9b39-1cb120e55bcd.png)
![Screenshot 2023-11-18 at 15.10.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fcdf3eaf-06ac-05be-5406-07f5036060aa.png)

# ジョブの実行

右上の**今すぐ実行**をクリックしてジョブを実行します。
![Screenshot 2023-11-18 at 14.53.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/37390114-df9c-62aa-9129-910b784ecb06.png)


なお、スケジュール実行したい場合には**スケジュールとトリガー**の**トリガーを追加**をクリックして、スケジュールを指定します。
![Screenshot 2023-11-18 at 14.54.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/53da521b-1737-cee0-57b9-db62c19cf40e.png)

**ジョブの実行**タブに切り替えます。
![Screenshot 2023-11-18 at 14.56.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c8aaffa5-0787-f1cb-62fd-eb0947be157d.png)

リアルタイムで実行の進捗を確認できます。
![Screenshot 2023-11-18 at 14.56.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2b476ff8-47b0-683d-dbb6-e6742403bc76.png)

少しするとジョブが完了します。開始時刻のリンクをクリックします。
![Screenshot 2023-11-18 at 15.03.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5b85b2bb-b711-169c-e7c9-07e2a383a8af.png)

タスクごとの実行結果を確認します。タスク`visualize`をクリックします。
![Screenshot 2023-11-18 at 15.04.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/37a0756e-d3c9-c7e3-328a-a08e1e161c48.png)

可視化の結果を確認できます。
![Screenshot 2023-11-18 at 15.04.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c0587943-e6fa-aac2-d7ac-e4af9c4bb3f4.png)

今回は静的なデータから可視化を行っているだけですが、これが日次で更新される場合であっても、ジョブをスケジュール実行することで常に最新の可視化結果を自動で手に入れることができます。ジョブというと定期実行のためのものとのイメージがありますが、複雑な処理をオーケストレーションできることもジョブの大きなメリットです。是非ご活用ください！

### Databricksクイックスタートガイド

[Databricksクイックスタートガイド](https://www.amazon.co.jp/dp/B09V1YXFVQ/)


### Databricks無料トライアル

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