Databricksのdbxを試してみる

こちらのマニュアルで説明しているdbxを実際に動かしてみます。[Databricks Connect](https://qiita.com/taka_yayoi/items/a9781c0a871a5a4fcf2e)の後継です。

https://qiita.com/taka_yayoi/items/994c701552e58d260562

# Databricksワークスペースの設定

アクセス先のDatabricksワークスペースでパーソナルアクセストークンを作成します。

1. トップバーのユーザー設定をクリックします。
![Screen Shot 2022-10-12 at 16.43.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ec1a9be2-3b3d-36c9-37cb-618741d2038d.png)
1. 新規トークンを作成をクリックして、トークンを作成しトークンをメモしておきます。
![Screen Shot 2022-10-12 at 16.44.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3a5870d4-f32e-bb25-9782-5b79bec908dd.png)

dbxからアクセスするクラスターのIDをメモしておきます。クラスターの詳細ページにアクセスして、URLからIDを取り出してメモしておきます。以下の赤枠内がクラスターIDです。詳細は[DatabricksでワークスペースID、クラスターID、ノートブックID、モデルID、ジョブIDを取得する](https://qiita.com/taka_yayoi/items/7127cabac70dd994fcba#%E3%82%AF%E3%83%A9%E3%82%B9%E3%82%BF%E3%83%BC%E3%81%AEurlid)をご覧ください。
![Screen Shot 2022-10-12 at 16.54.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2d4b05e7-a324-8144-41a3-eea6f79d972d.png)

# ローカルでの設定

ここでは、[Databricks CLI](https://docs.databricks.com/dev-tools/cli/index.html)がインストール済みであるものとして、CLIの設定ファイル`.databrickscfg`を編集してDatabricksワークスペースのホストと上記ステップで取得したパーソナルアクセストークンを設定します。

```
[DEFAULT]
host = https://<DatabricksワークスペースURL>/
token = <パーソナルアクセストークン>
jobs-api-version = 2.0
```

dbxルートディレクトリを作成します。

```bash
mkdir dbx-demo
cd dbx-demo
```

Python仮想環境を作成します。pipenvがインストールされていない場合にはインストールします。

```bash
pipenv --python <version>
```
![Screen Shot 2022-10-12 at 16.50.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/89ee0d91-b1fb-cc86-51e8-2ed5d1690aa8.png)

仮想環境を有効化します。

```
pipenv shell
```
![Screen Shot 2022-10-12 at 16.51.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/56617719-5cce-77c9-544c-c99bfeac6445.png)

`conf`ディレクトリを作成し、`deployment.yaml`ファイルを作成します。

```
mkdir conf
```

```yaml:deployment.yaml
build:
  no_build: true
environments:
  default:
    workflows:
      - name: "dbx-demo-job"
        spark_python_task:
          python_file: "file://dbx-demo-job.py"
```

`dbx-demo`配下に`dbx-demo-job.py`ファイルを作成します。

```py:dbx-demo-job.py
# For testing and debugging of local objects, run
# "pip install pyspark=X.Y.Z", where "X.Y.Z"
# matches the version of PySpark
# on your target clusters.
from pyspark.sql import SparkSession

from pyspark.sql.types import *
from datetime import date

spark = SparkSession.builder.appName("dbx-demo").getOrCreate()

# Create a DataFrame consisting of high and low temperatures
# by airport code and date.
schema = StructType([
   StructField('AirportCode', StringType(), False),
   StructField('Date', DateType(), False),
   StructField('TempHighF', IntegerType(), False),
   StructField('TempLowF', IntegerType(), False)
])

data = [
   [ 'BLI', date(2021, 4, 3), 52, 43],
   [ 'BLI', date(2021, 4, 2), 50, 38],
   [ 'BLI', date(2021, 4, 1), 52, 41],
   [ 'PDX', date(2021, 4, 3), 64, 45],
   [ 'PDX', date(2021, 4, 2), 61, 41],
   [ 'PDX', date(2021, 4, 1), 66, 39],
   [ 'SEA', date(2021, 4, 3), 57, 43],
   [ 'SEA', date(2021, 4, 2), 54, 39],
   [ 'SEA', date(2021, 4, 1), 56, 41]
]

temps = spark.createDataFrame(data, schema)

# Create a table on the cluster and then fill
# the table with the DataFrame's contents.
# If the table already exists from a previous run,
# delete it first.
spark.sql('USE default')
spark.sql('DROP TABLE IF EXISTS demo_temps_table')
temps.write.saveAsTable('demo_temps_table')

# Query the table on the cluster, returning rows
# where the airport code is not BLI and the date is later
# than 2021-04-01. Group the results and order by high
# temperature in descending order.
df_temps = spark.sql("SELECT * FROM demo_temps_table " \
   "WHERE AirportCode != 'BLI' AND Date > '2021-04-01' " \
   "GROUP BY AirportCode, Date, TempHighF, TempLowF " \
   "ORDER BY TempHighF DESC")
df_temps.show()

# Results:
#
# +-----------+----------+---------+--------+
# |AirportCode|      Date|TempHighF|TempLowF|
# +-----------+----------+---------+--------+
# |        PDX|2021-04-03|       64|      45|
# |        PDX|2021-04-02|       61|      41|
# |        SEA|2021-04-03|       57|      43|
# |        SEA|2021-04-02|       54|      39|
# +-----------+----------+---------+--------+

# Clean up by deleting the table from the cluster.
spark.sql('DROP TABLE demo_temps_table')
```

dbxの設定を行います。

```sh
dbx configure --profile DEFAULT
```
![Screen Shot 2022-10-12 at 17.55.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2f40bae8-4324-a264-0962-2a4b2657d1c3.png)

以下のコマンドを実行して、PythonスクリプトをDatabricksで実行します。`<existing-cluster-id>`が[以前のステップ](#databricksワークスペースの設定)でメモしたクラスターIDを指定してください。

```sh
dbx execute --cluster-id=<existing-cluster-id> --job=dbx-demo-job --no-rebuild --no-package
```
![Screen Shot 2022-10-12 at 17.58.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e63b2ddd-a87c-3aae-8ab8-df81a6a1e018.png)

無事に処理結果が表示されました。
![Screen Shot 2022-10-12 at 17.59.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/741a37b4-9a5f-eb1d-ee33-d9d8ea9a0058.png)


### Databricks 無料トライアル

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