Databricksレイクハウスモニタリングによるテーブル品質の監視

サーバレスが有効化されているリージョンではテーブルや機械学習モデルの監視を行えるレイクハウスモニタリングが利用できるようになっています。

https://qiita.com/taka_yayoi/items/5aaddc9aa7b06dc87f12

時系列(time series)分析の[サンプルノートブック](https://docs.databricks.com/ja/lakehouse-monitoring/create-monitor-api.html#notebook-example-time-series-profile)をウォークスルーします。時間とともにデータが変化していくテーブルを監視するタイプです。

# レイクハウスモニタリングサンプルノートブック: 時系列分析

**ユーザー要件**
- Unity Catalogにアクセスできるクラスターでのコマンド実行権限が必要です。
- 最低でも一つのカタログに対する`USE CATALOG`権限が必要です。最低でも一つのスキーマに対する`USE SCHEMA`権限が必要です。このノートブックではテーブルを`main.default`スキーマに作成します。`main.default`スキーマへの権限がない場合には、権限を持つカタログとスキーマに変更してください。

**システム要件:**
- ワークスペースでUnity Catalogが有効化されている必要があります。
- Databricks Runtime 12.2LTS以降
- シングルユーザーあるいは共有クラスター

このノートブックでは時系列モニターの作成方法を説明します。

レイクハウスモニタリングの詳細に関してはドキュメントを参照ください。([AWS](https://docs.databricks.com/ja/lakehouse-monitoring/index.html)|[Azure](https://learn.microsoft.com/ja-jp/azure/databricks/lakehouse-monitoring/))

## セットアップ
* クラスター設定の検証
* Pythonクライアントのインストール
* カタログ、スキーマ、テーブル名の定義

```py
# クラスター設定のチェック。このセルが失敗する場合、ノートブックの右上のクラスターセレクターで、Databricks Runtime 12.2 LTS　以降が稼働しているクラスターを設定、選択してください。
import os

assert float(os.environ.get("DATABRICKS_RUNTIME_VERSION", 0)) >= 12.2, "Please configure your cluster to use Databricks Runtime 12.2 LTS or above."
```

```py
%pip install "https://ml-team-public-read.s3.amazonaws.com/wheels/data-monitoring/a4050ef7-b183-47a1-a145-e614628e3146/databricks_lakehouse_monitoring-0.4.4-py3-none-any.whl"
```

```py
# 新規にインストールしたwheelを用いて環境をリセットするためにこのステップが必要です。
dbutils.library.restartPython()
```

```py
# カタログに対する`USE CATALOG`権限とスキーマに対する`USE SCHEMA`権限が必要です。
# 必要に応じて、ここでカタログとスキーマを変更して下さい。

CATALOG = "takaakiyayoi_catalog"
SCHEMA = "monitoring"
```

テーブル名にユーザー名を埋め込みます。

```py
username = spark.sql("SELECT current_user()").first()["current_user()"]
username_prefixes = username.split("@")[0].split(".")
```

```py
unique_suffix = "_".join([username_prefixes[0], username_prefixes[1][0:2]])
TABLE_NAME = f"{CATALOG}.{SCHEMA}.wine_ts_{unique_suffix}"
BASELINE_TABLE = f"{CATALOG}.{SCHEMA}.wine_ts_baseline_{unique_suffix}"
TIMESTAMP_COL = "timestamp"
```

```py
spark.sql(f"DROP TABLE IF EXISTS {TABLE_NAME}")
spark.sql(f"DROP TABLE IF EXISTS {BASELINE_TABLE}")
```

## ユーザージャーニー
1. テーブルの作成: 生のデータを読み込みプライマリテーブル(監視されるテーブル)とベーステーブル(期待する品質基準を満たす既知のデータを格納)を作成。
2. プライマリテーブルに対するモニターを作成。
3. メトリクステーブルの調査。
4. テーブルを変更してメトリクスを更新。メトリクステーブルの調査。
5. [オプション] モニターの削除。

### 1. Unity Catalogでプライマリテーブルと(オプション)ベースラインテーブルを作成
* これらのテーブルはUnity Catalog配下のDeltaテーブルであり、ノートブックを実行するユーザーが所有者である必要があります。
* 監視されるテーブルは「プライマリテーブル」とも呼ばれます。  
* ベースラインテーブルは監視されるテーブルと同じスキーマを持つ必要があります。

このサンプルでは`winequality`データセットを使用します。

```py
import pandas as pd

white_wine = pd.read_csv("/dbfs/databricks-datasets/wine-quality/winequality-white.csv", sep=";")
red_wine = pd.read_csv("/dbfs/databricks-datasets/wine-quality/winequality-red.csv", sep=";")

# カテゴリーの追加
white_wine["type"] = "white"
red_wine["type"] = "red"
data_pdf = pd.concat([white_wine, red_wine], axis=0)

# カラム名の整形
data_pdf.columns = data_pdf.columns.str.replace(" ", "_")
```

#### データを分割します。`baseline_df`はベースラインテーブルです。 `ts1_df`はプライマリテーブルであり、`ts2_df`は模擬的に未来のデータを示すものです。

```py
data_df = spark.createDataFrame(data_pdf)
baseline_df, ts1_df, ts2_df = data_df.randomSplit(weights=[0.20, 0.40, 0.40], seed=42)
```

#### 時系列データをシミュレートするために異なるタイムスタンプを作成します。

```py
from datetime import timedelta, datetime
from pyspark.sql import functions as F

# 異なるタイムスタンプを持つデータをシミュレート
timestamp_0 = datetime.now() # ベースラインデータ
timestamp_1 = (datetime.now() + timedelta(1)).timestamp() # 1日後
timestamp_2 = (datetime.now() + timedelta(2)).timestamp() 

baseline_df = baseline_df.withColumn("timestamp", F.lit(timestamp_0).cast("timestamp"))
ts1_df = ts1_df.withColumn("timestamp", F.lit(timestamp_1).cast("timestamp"))
ts2_df = ts2_df.withColumn("timestamp", F.lit(timestamp_2).cast("timestamp"))
```

```py
baseline_df.display()
```
![Screenshot 2023-12-16 at 16.52.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d6b41c9f-df1f-294d-d470-e66be6bba1ff.png)

#### Unity CatalogにベースラインのDeltaテーブルとプライマリのDeltaテーブルを作成

```py
# Unity Catalogにベースラインテーブルを書き込み

(baseline_df
  .write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema",True)
  .option("delta.enableChangeDataFeed", "true")
  .saveAsTable(f"{BASELINE_TABLE}")
)
```
```py
# Unity Catalogにプライマリテーブルを書き込み。これが監視されるテーブルとなります。このノートブックの後半で、このテーブルに未来のタイムスタンプを持つデータを追加します。 

(ts1_df
  .write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema",True)
  .option("delta.enableChangeDataFeed", "true")
  .saveAsTable(f"{TABLE_NAME}")
)
```

テーブルが作成されます。
![Screenshot 2023-12-16 at 16.52.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7f719efb-a4d6-d954-280c-6fd770facdbe.png)

行数を確認します。

```py
spark.sql(f"SELECT COUNT(*) FROM {TABLE_NAME}").display()
```
![Screenshot 2023-12-16 at 16.53.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fbbeb1eb-33d8-94a7-af72-d8169d598b34.png)

2. モニターの作成
このノートブックでは、`TimeSeries`タイプの分析を説明します。他の分析タイプに関しては、レイクハウスモニタリングのドキュメントを参照ください。([AWS](https://docs.databricks.com/ja/lakehouse-monitoring/index.html)|[Azure](https://docs.databricks.com/ja/lakehouse-monitoring/index.html))

**ビジネスやユースケース観点から除外されるべきすべてのカラムを除外するようにしてください。**

```py
import databricks.lakehouse_monitoring as lm
```

```py
# 分析するウィンドウのサイズ
GRANULARITIES = ["1 day"]                       

# データをスライスするエクスプレッション
SLICING_EXPRS = ["type='Red'"]  
```

```py
print(f"Creating monitor for {TABLE_NAME}")

info = lm.create_monitor(
  table_name=TABLE_NAME,
  profile_type=lm.TimeSeries(
    timestamp_col=TIMESTAMP_COL,
    granularities=GRANULARITIES
  ),
  slicing_exprs=SLICING_EXPRS,
  baseline_table_name=BASELINE_TABLE,
  output_schema_name=f"{CATALOG}.{SCHEMA}"
)
```

カタログエクスプローラからプライマリテーブルにアクセスし、**品質**タブをクリックするとモニターを確認できます。
![Screenshot 2023-12-16 at 16.53.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8a374dd7-2f33-faf9-be1a-fc9d8ab86f3e.png)
![Screenshot 2023-12-16 at 16.53.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ac812ecc-0df5-5e13-1e24-f6fe9f9ed039.png)

```py
import time


# モニターの作成を待ちます
while info.status == lm.MonitorStatus.PENDING:
  info = lm.get_monitor(table_name=TABLE_NAME)
  time.sleep(10)

assert(info.status == lm.MonitorStatus.ACTIVE)
```

また、**更新履歴を表示**をクリックすることでメトリクスの更新処理の状態を確認できます。
![Screenshot 2023-12-16 at 16.53.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4af614cc-c456-0c7b-7273-04cd22ae3103.png)

モニターを作成すると、自動でメトリクスの更新処理が起動します。

```py
# 作成時にメトリックの更新が自動でトリガーされます
refreshes = lm.list_refreshes(table_name=TABLE_NAME)
assert(len(refreshes) > 0)

run_info = refreshes[0]
while run_info.state in (lm.RefreshState.PENDING, lm.RefreshState.RUNNING):
  run_info = lm.get_refresh(table_name=TABLE_NAME, refresh_id=run_info.refresh_id)
  time.sleep(30)

assert(run_info.state == lm.RefreshState.SUCCESS)
```

モニターを作成するとメトリクスを確認するためのダッシュボードが自動生成されます。ダッシュボードを開くにはセルのアウトプットでハイライトされているダッシュボードのリンクをクリックします。カタログエクスプローラのUIからダッシュボードに移動することもできます。
![Screenshot 2023-12-16 at 17.24.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/271269f9-6d18-7ef8-7bd5-4af2257ed72c.png)

なお、ダッシュボードやダッシュボードで使用されるクエリーはホームディレクトリ配下の`databricks_lakehouse_monitoring`に格納されます。
![Screenshot 2023-12-16 at 16.54.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f4065d99-857c-93b4-c0cf-8fcce4d34718.png)

ダッシュボードにはデータの統計情報などが表示されます。
![Screenshot 2023-12-16 at 17.01.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fad02c5c-fc1b-07d1-f99e-a42bedba9c19.png)
![Screenshot 2023-12-16 at 17.01.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e462f1e2-4913-d198-3a1c-b55581594c93.png)

以下のメソッドでモニターの情報を表示できます。

```py
lm.get_monitor(table_name=TABLE_NAME)
```
![Screenshot 2023-12-16 at 17.26.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6248b02e-f12a-9756-b11b-219caf5b9387.png)

## 3. メトリクステーブルの調査

デフォルトではメトリクステーブルはdefaultデータベースに格納されます。

`create_monitor`の呼び出しで2つの新規テーブル作成されます: プロファイルメトリクステーブルとドリフトメトリクステーブルです。

これらの2つのテーブルは分析ジョブのアウトプットを記録します。このテーブルでは監視されるプライマリテーブルと同じ名前を使用し、サフィックス`_profile_metrics`と`_drift_metrics`を使用します。

### プロファイルメトリクステーブルのオリエンテーション

プロファイルメトリクステーブルにはサフィックス`_profile_metrics`が追加されます。テーブルに表示される統計情報の一覧に関しては、ドキュメントを参照ください([AWS](https://docs.databricks.com/ja/lakehouse-monitoring/monitor-output.html#profile-metrics-table)|[Azure](https://learn.microsoft.com/ja-jp/azure/databricks/lakehouse-monitoring/monitor-output#profile-metrics-table))。

- プライマリテーブルのすべてのカラムに対して、ベースラインテーブルとプライマリテーブルに対するサマリーの統計情報を表示します。カラム`log_type`はプライマリテーブルの統計情報を示す`INPUT`、ベースラインテーブルの統計情報を示す`BASELINE`を表示しています。プライマリテーブルのカラムはカラム`column_name`で識別できます。
- `TimeSeries`タイプの分析では、`granularity`カラムには行に対する粒度が表示されます。ベースラインテーブルの統計情報では`granularity`カラムの値は`null`になります。
- このテーブルでは、それぞれの時間ウィンドウにおけるそれぞれのスライスキーのそれぞれの値に対する統計情報とテーブル全体に対する統計情報が表示されます。テーブル全体の統計情報は`slice_key` = `slice_value` = `null`で特定することができます。
- プライマリテーブルでは、`window`カラムに当該行に対応する時間ウィンドウが表示されます。ベースラインテーブルの統計情報では、`window`カラムは`null`になります。
- いくつかの統計情報は特定のカラムではなくテーブル全体に対して計算されます。`column_name`カラムの`:table`によって、これらの統計情報を特定できます。

```py
# プロファイルメトリクステーブルを表示
profile_table = f"{TABLE_NAME}_profile_metrics"
display(spark.sql(f"SELECT * FROM {profile_table}"))
```
![Screenshot 2023-12-16 at 17.02.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/33866865-39f5-1870-a9a0-66c9e286b56d.png)

### ドリフトメトリクステーブルのオリエンテーション

ドリフトメトリクステーブルにはサフィックス`_drift_metrics`が追加されます。テーブルに表示される統計情報の一覧に関しては、ドキュメントを参照ください([AWS](https://docs.databricks.com/ja/lakehouse-monitoring/monitor-output.html#drift-metrics-table)|[Azure](https://learn.microsoft.com/ja-jp/azure/databricks/lakehouse-monitoring/monitor-output#drift-metrics-table))。

- プライマリテーブルのすべてのカラムに対して、ドリフトテーブルではテーブルの現在の値と以前の分析実行時の値、ベースラインテーブルと比較した一連のメトリクスを表示します。ベースラインテーブルに対するドリフトでは、カラム`drift_type`に`BASELINE`が表示され、以前の時間ウィンドウに対するドリフトでは、`CONSECUTIVE`が表示されます。プロファイルテーブルと同じように、プライマリテーブルのカラムはカラム`column_name`で特定できます。
  - この時点では、このモニターの最初の実行であるため、比較する以前のウィンドウがありません。このため、`drift_type`が`CONSECUTIVE`である行はありません。
- `TimeSeries`タイプの分析では、`granularity`カラムには行に対する粒度が表示されます。
- このテーブルでは、それぞれの時間ウィンドウにおけるそれぞれのスライスキーのそれぞれの値に対する統計情報とテーブル全体に対する統計情報が表示されます。テーブル全体の統計情報は`slice_key` = `slice_value` = `null`で特定することができます。
- `window`カラムには当該行に対応する時間ウィンドウが表示されます。`window_cmp`カラムには比較対象のウィンドウが表示されます。ベースラインテーブルとの比較の場合には、`window_cmp`は`null`になります。
- いくつかの統計情報は特定のカラムではなくテーブル全体に対して計算されます。`column_name`カラムの`:table`によって、これらの統計情報を特定できます。

```py
# ドリフトメトリクステーブルの表示
drift_table = f"{TABLE_NAME}_drift_metrics"
display(spark.sql(f"SELECT * FROM {drift_table}"))
```

![Screenshot 2023-12-16 at 17.02.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/028a16ac-9c71-1c2f-b7ad-23f5f660a295.png)

## 4. テーブルに新規データを追加し、メトリクスをリフレッシュ

### テーブルに新規データを追加

以下のセルではプライマリテーブルに、シミュレートした未来のデータ`ts2_df`を追加します。

```py
(ts2_df
  .write.format("delta").mode("append") 
  .option("mergeSchema",True) 
  .option("delta.enableChangeDataFeed", "true") 
  .saveAsTable(f"{TABLE_NAME}")
)
```

行数を確認します。

```py
spark.sql(f"SELECT COUNT(*) FROM {TABLE_NAME}").display()
```
![Screenshot 2023-12-16 at 17.03.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6181a471-4196-7730-6907-fd5928e2be48.png)

### メトリクスのリフレッシュ

```py
run_info = lm.run_refresh(table_name=TABLE_NAME)
while run_info.state in (lm.RefreshState.PENDING, lm.RefreshState.RUNNING):
  run_info = lm.get_refresh(table_name=TABLE_NAME, refresh_id=run_info.refresh_id)
  time.sleep(30)

assert(run_info.state == lm.RefreshState.SUCCESS)
```

2回目のリフレッシュが実行されます。
![Screenshot 2023-12-16 at 17.03.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cf6fde90-ef75-609e-abdf-e1c26892bde5.png)

リフレッシュが完了したら、変化を確認するためにモニタリングダッシュボードを開きます。
![Screenshot 2023-12-16 at 17.12.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3c4d4caa-1154-02f7-5d89-869ef3c5f7d8.png)
![Screenshot 2023-12-16 at 17.12.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bc17b16a-1887-6444-8426-9dfdf5edcb95.png)

時間ウィンドウ間の変化を確認することができます。
![Screenshot 2023-12-16 at 17.13.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/660b2daf-a148-ed04-be9e-933e9a2b67c2.png)
![Screenshot 2023-12-16 at 17.13.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c5bdbf9e-ae53-46ea-5e52-2f9f91b80f9f.png)

## [オプション] モニターの削除

モニターをクリーンアップするには以下のコードのコメントを解除してください。一つのテーブルには単一のモニターのみを追加できます。

```py
# lm.delete_monitor(table_name=TABLE_NAME)
```

テーブルの統計情報をモニタリングするためには、これまでには手組みで色々なロジックを構成しないといけませんでしたが、例クハウスモニタリングを活用いただくことで簡単にテーブルの状態を監視できるようになります。(日本リージョンで使えるようになったら)是非お試しください！

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

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


### Databricks無料トライアル

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