Databricksベストプラクティス：ノートブックのモジュール化

以前こちらの記事を翻訳しましたが、2021年のことです。

https://qiita.com/taka_yayoi/items/ba0294e20a19cdb1fe10#1-pip%E3%83%9E%E3%82%B8%E3%83%83%E3%82%AF%E3%82%B3%E3%83%9E%E3%83%B3%E3%83%89-python%E3%83%91%E3%83%83%E3%82%B1%E3%83%BC%E3%82%B8%E3%81%AE%E3%82%A4%E3%83%B3%E3%82%B9%E3%83%88%E3%83%BC%E3%83%AB%E5%8F%8A%E3%81%B3python%E7%92%B0%E5%A2%83%E3%81%AE%E7%AE%A1%E7%90%86

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

当時はPythonのカスタムモジュールのインポートが一筋縄ではいきませんでした。Wheelを作成したり、Reposを使ったりしなくてはなりませんでした。今では、ワークスペースのどこでも[ワークスペースファイル](https://docs.databricks.com/ja/files/workspace.html)が使えるので、簡単にカスタムモジュールを利用できます。

ここでは、マニュアルに記載されている[手順](https://docs.databricks.com/ja/files/workspace-modules.html)にそって、こちらのリポジトリにあるノートブックをウォークスルーします。その過程でその他のDatabricksノートブックのTipsもご紹介します。

https://github.com/databricks/notebook-best-practices/tree/main

# モジュール化しない場合

`notebooks/covid_eda_raw.py`のノートブックではCOVID-19の入院者数データに対する分析をこなっていますが、モジュール化は行っていません。

以下のコマンドでは、クラスターのローカルストレージの`/tmp`にCSVファイルがダウンロードされます。

```sh
!wget -q https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/hospitalizations/covid-hospitalizations.csv -O /tmp/covid-hospitalizations.csv
```

:::note info
**ポイント**
Databricksでファイルを取り扱う際には、**どこに**保存するのかを意識することが重要です。詳細はこちらの記事をご覧ください。
:::

https://qiita.com/taka_yayoi/items/075c6b3aeafac54c8ac4


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


## 変換処理

```py
import pandas as pd

# /tmpから読み込み、JBNのサブセットを取得、ピボットを行い、欠損地を0に置換
df = pd.read_csv("/tmp/covid-hospitalizations.csv")
df = df[df.iso_code == 'JPN']\
     .pivot_table(values='value', columns='indicator', index='date')\
     .fillna(0)

df
```
![Screenshot 2024-05-07 at 16.12.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4c0075c0-7ce8-08ec-6311-453ec1f18cad.png)

## 可視化 

```py
df.plot(figsize=(13,6), grid=True).legend(loc='upper left')
```
![download (1).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/54f500c5-5a08-2c2a-9cde-9a3165744c25.png)

## Delta Lakeに保存

現在のスキーマにはカラム名にスペースが入っており、Delta Lakeと互換性がありません。データをテーブルとして保存するために、スペースをアンダースコアに置換します。また、日付のインデックスを自身のカラムとして追加し、他の方がこのテーブルをクエリーした際に利用できるようにします。

```py
import pyspark.pandas as ps

clean_cols = df.columns.str.replace(' ', '_')

# pandas on Sparkデータフレームの作成
psdf = ps.from_pandas(df)

psdf.columns = clean_cols
psdf['date'] = psdf.index

# 最新のデータを毎回上書きして、Delta tableに書き込み
psdf.to_table(name='users.takaaki_yayoi.dev_covid_analysis', mode='overwrite')
```

:::note info
**ポイント**
ラップトップのメモリーやCPUを使うJuypter Notebookと違い、Databricksの計算資源はクラウド上の**揮発性**のクラスターです。つまり、クラスターが停止すると永続化していないデータは消えてしまいます。後でアクセスする可能性があるデータはテーブルやファイルとして永続化するようにしましょう。
:::

永続化したテーブルはサイドメニューの**カタログ**からアクセスできるカタログエクスプローラで確認できます。
![Screenshot 2024-05-07 at 16.43.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a8acc257-0caf-ffe8-5227-caeff396c039.png)


## テーブルの参照

テーブルとして永続化すると、SQLでテーブルにアクセスできるようになります。

```sql
%sql
SELECT * FROM users.takaaki_yayoi.dev_covid_analysis
```
![Screenshot 2024-05-07 at 16.14.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a5a4c8b2-594f-de15-f0f8-e28767a7a28e.png)

:::note info
**ポイント**
Databricksノートブックでは、セル単位で言語を切り替えることができます。上の例では`%sql`マジックコマンドを指定して、SQLを実行しています。詳細は[こちら](https://docs.databricks.com/ja/notebooks/notebooks-code.html#mix-languages)をご覧ください。
:::

# ノートブックのモジュール化

ロジックを再利用するケースにおいてモジュール化は有効です。また、ノートブックの肥大化を避けることができ、コーディングやデバッグの効率が向上します。

`notebooks/covid_eda_modular.py`ではモジュール化を行っています。`covid_analysis/transforms.py`にモジュール化した関数が格納されています。

```py:covid_analysis/transforms.py
import pandas as pd


# 国コードでフィルタリング
def filter_country(pdf, country="USA"):
    pdf = pdf[pdf.iso_code == country]
    return pdf


# indicatorでピボットし、欠損値を補完
def pivot_and_clean(pdf, fillna):
    pdf["value"] = pd.to_numeric(pdf["value"])
    pdf = pdf.pivot_table(
        values="value", columns="indicator", index="date"
    ).fillna(fillna)
    return pdf


# Deltaテーブルと互換性のあるカラム名を作成
def clean_spark_cols(pdf):
    pdf.columns = pdf.columns.str.replace(" ", "_")
    return pdf


# インデックスをカラムに変換 (pandas API on Sparkでも動作します)
def index_to_col(df, colname):
    df[colname] = df.index
    return df
```

`notebooks/covid_eda_modular.py`ではモジュールを読み込んで利用します。

## セットアップ 

こちらを実行することで、モジュールで更新があった場合にそれらが反映されます。

```py
%load_ext autoreload
%autoreload 2
```

また、モジュールを読み込む際にはパスに注意してください。

```py
import sys
import os
print(sys.path)
```
```
[..., 
'/Workspace/Users/takaaki.yayoi@databricks.com/20240507_notebook_best_practice/notebook-best-practices/notebooks']
```

上の例では、ノートブックのカレントディレクトリはパスに含まれていますが、モジュールファイルのパスは含まれていません。以下を実行して親のディレクトリをパスに追加します。

```py
# 親フォルダにあるパッケージをインポートできるようにパスを追加
sys.path.append(os.path.abspath('..'))
```

```py
print(sys.path)
```
```
[...,
'/Workspace/Users/takaaki.yayoi@databricks.com/20240507_notebook_best_practice/notebook-best-practices/notebooks', 
'/Workspace/Users/takaaki.yayoi@databricks.com/20240507_notebook_best_practice/notebook-best-practices']
```

なお、Reposで作業している場合には、Reposフォルダー(Gitフォルダー)のルートパスは最初から追加されています。詳細は[こちら](https://docs.databricks.com/ja/libraries/index.html#precedence)をご覧ください。

## データの取得と変換

```py
data_path = 'https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/hospitalizations/covid-hospitalizations.csv'
print(f'Data path: {data_path}')
```
```
Data path: https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/hospitalizations/covid-hospitalizations.csv
```

```py
from covid_analysis.transforms import *
import pandas as pd

df = pd.read_csv(data_path)
df = filter_country(df, country='JPN')
df = pivot_and_clean(df, fillna=0)  
df = clean_spark_cols(df)
df = index_to_col(df, colname='date')
# PandasをSparkデータフレームに変換
df = spark.createDataFrame(df)

display(df)
```
![Screenshot 2024-05-07 at 16.29.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/37e4c9b8-39d8-3957-5567-7b311ecd8f42.png)

## Delta Lakeに保存

現在のスキーマにはカラム名にスペースが入っており、Delta Lakeと互換性がありません。データをテーブルとして保存するために、スペースをアンダースコアに置換します。また、日付のインデックスを自身のカラムとして追加し、他の方がこのテーブルをクエリーした際に利用できるようにします。

```py
# Delta Lakeに書き込み
df.write.mode('overwrite').saveAsTable('users.takaaki_yayoi.covid_stats')
```

## 可視化

```py
# Databricksのビジュアライゼーションとデータプロファイリングの活用
display(spark.table('users.takaaki_yayoi.covid_stats'))
```

:::note info
**ポイント**
Databricksノートブックには可視化の機能がビルトインされています。データフレームを表示した状態で **+ > 可視化** を選択します。
![Screenshot 2024-05-07 at 16.30.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8672de5f-3b9e-fe25-ab75-841c4c7759e4.png)
プレビューを確認しながら、可視化の設定を行います。
![Screenshot 2024-05-07 at 16.32.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/96f82f4c-e394-458c-2a39-e8ac1672002a.png)
これで簡単に可視化を行うことができます。
![Screenshot 2024-05-07 at 16.32.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e54ea3e5-15d0-88f4-88da-821bad946774.png)
:::

:::note info
**ポイント**
さらに、 **+ > データプロファイル**をクリックすると、データフレームの統計情報や傾向をクイックに確認することができます。
![Screenshot 2024-05-07 at 16.34.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/acf86561-d52d-1c5f-06de-e30910466f07.png)
:::

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

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