Delta Live Tables(DLT)クイックスタートのウォークスルー

Databricksで高品質、高パフォーマンスなETLパイプラインを構築できるソリューションがDelta Live Tables(略してDLT)です。

こちらの記事でクイックスタートを説明していますが、一年以上に書いたものなので改めてDLTの画面をウォークスルーしてみました。

https://qiita.com/taka_yayoi/items/7fe8ed2c2f95fd53cc3d

# Delta Live Tablesとは

Delta Live Tablesは、信頼性が高く、維持が容易でテスト可能なデータ処理パイプラインを構築するためのフレームワークです。データに対する変換処理を定義すると、Delta Live Tablesはタスクのオーケストレーション、クラスター管理、モニタリング、データ品質、エラー処理を行います。そして、Delta Live Tablesはそれぞれの処理ステップであなたが定義したターゲットスキーマに基づき、どのようにデータが変換されるのかを管理します。
![Screen Shot 2022-12-13 at 10.36.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a410f1b8-614f-2269-8c89-6d075ef87bd4.png)

また、Delta Live Tablesの[エクスペクテーション](https://qiita.com/taka_yayoi/items/437817339ada5d21b424)を活用してデータ品質を維持することもできます。エクスペクテーションによって、期待されるデータ品質を定義し、エクスペクテーションに沿わないデータをどのように取り扱うのかを定義することができます。
![Screen Shot 2022-12-13 at 10.38.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7eaaafba-b5af-d62c-a924-e599ba462dc0.png)

Delta Live Tablesのパイプラインは、Databricksノートブックを使用して簡単に作成することができます。
![Screen Shot 2022-12-13 at 10.39.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d65ff312-7177-9077-71f5-103a000e6bf7.png)

なお、DLTは[メダリオンアーキテクチャ](https://qiita.com/taka_yayoi/items/f3d9028cf533e2c78b9d)を前提としていますので、こちらのコンセプトを理解しておくことをお勧めします。

> メダリオンアーキテクチャは、レイクハウスに格納されているデータの品質を示す一連のデータレイヤーを表現します。企業のデータプロダクトに対して信頼できる唯一の情報源(single source of truth)を構築するために、Databricksではマルチレイヤーのアプローチを取ることをお勧めしています。このアーキテクチャは、公立的な分析に最適化されたレイアウトでデータ格納される前の検証、変換を行う複数のレイヤーをデータが追加する際の、原子性、一貫性、分離性、耐久性を保証します。ブロンズ(生)、シルバー(検証済み)、ゴールド(拡張済み)という用語はそれぞれのレイヤーにおけるデータの品質を表現しています。

# クイックスタートノートブックのウォークスルー

本記事ではDelta Live TablesパイプラインとWikipediaクリックストリームデータを用いてデモをお見せします。Wikipediaクリックストリームデータを用いて以下のことを行います。

- 生のJSONクリックストリームデータをテーブルに読み込みます。
- 生データテーブルからレコードを読み込み、Delta Live Tablesの[エクスペクテーション](https://qiita.com/taka_yayoi/items/437817339ada5d21b424)を用いて、クレンジングされたデータを格納する新たなテーブルを作成します。
- Delta Live Tablesのクエリーを行い、クレンジングデータテーブルのレコードから新たなデータセットを作成します。

以下のノートブックを使用します。なお、[Pythonのノートブック](https://docs.databricks.com/_static/notebooks/dlt-wikipedia-python.html)でも同様のことが可能です。

https://docs.databricks.com/_static/notebooks/dlt-wikipedia-sql.html

## パイプラインの準備

上のノートブックだけでもDLTを実行することは可能ですが、追加のパラメーターを設定することで、DLTの取り回しがさらに楽になります。

### ヘルパーノートブックの作成

`./Includes/setup`というノートブックを作成し、以下のロジックを記述します。こちらを実行することで、DLTパイプラインで使用するデータベースやストレージロケーションを設定します。

```py:Python
import pyspark.sql.functions as F
import re

course = "dlt"

username = spark.sql("SELECT current_user()").collect()[0][0]
userhome = f"dbfs:/user/{username}/{course}"
database = f"""{course}_{re.sub("[^a-zA-Z0-9]", "_", username)}_db"""

print(f"""
username: {username}
userhome: {userhome}
database: {database}""")

dbutils.widgets.text("mode", "setup")
mode = dbutils.widgets.get("mode")

if mode == "reset":
    spark.sql(f"DROP DATABASE IF EXISTS {database} CASCADE")
    dbutils.fs.rm(userhome, True)
spark.sql(f"CREATE DATABASE IF NOT EXISTS {database}")
spark.sql(f"USE {database}")

if mode == "cleanup":
    spark.sql(f"DROP DATABASE IF EXISTS {database} CASCADE")
    dbutils.fs.rm(userhome, True)

storage_location = userhome + "/output"
```

### 準備用ノートブックの作成、実行

ワークスペースの現在フォルダに**DLT-main**というPythonノートブックを作成します。名前は任意です。

```
%run ./Includes/setup $mode="reset"
```

上のコマンドを実行することで、ヘルパーノートブックが実行されます。
![Screen Shot 2022-12-13 at 10.42.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/77155015-6c3f-8b4c-9633-08cd592352df.png)

以下のコマンドを実行してストレージの場所を取得しておきます。

```py:Python
storage_location.split(':')[1]
```

![Screen Shot 2022-12-13 at 10.52.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/732ad73d-111b-1468-ae61-c831aa32c7a3.png)


## パイプラインのノートブックの作成

DLTでは、パイプラインの処理を記述するのではなく、パイプラインの各ステップを宣言するアプローチをとっています。これによって、各ステップの意味が明確になることに加え、間の処理はDLTが面倒を見てくれます。**dlt-quickstart-sql**というSQLノートブックを作成し、以下のロジックを記述します。

### ブロンズテーブル

ブロンズテーブルでは生のデータを保持します。

```sql:SQL
CREATE OR REFRESH LIVE TABLE clickstream_raw
COMMENT "/databricks-datasets から取り込まれるWikipediaの生のクリックストリームデータセット(ブロンズ)"
AS SELECT * FROM json.`/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json`;
```

### シルバーテーブル

ブロンズテーブルに対してクレンジングした結果をシルバーテーブルとして保持します。ここでは[エクスペクテーション](https://qiita.com/taka_yayoi/items/437817339ada5d21b424)を用いてデータの品質を担保しています。

```sql:SQL
CREATE OR REFRESH LIVE TABLE clickstream_prepared(
  -- current_page_titleは非NULL、click_countは1以上であるという制約
  CONSTRAINT valid_current_page EXPECT (current_page_title IS NOT NULL),
  CONSTRAINT valid_count EXPECT (click_count > 0) ON VIOLATION FAIL UPDATE
)
COMMENT "分析のためにクレンジング、準備されたWikipediaクリックストリームデータ(シルバー)"
AS SELECT
  curr_title AS current_page_title,
  CAST(n AS INT) AS click_count,
  prev_title AS previous_page_title
FROM live.clickstream_raw;
```

### ゴールドテーブル

BIや機械学習など最終的にビジネスで活用するデータを格納するのがゴールドテーブルです。

```py:Python
CREATE OR REFRESH LIVE TABLE top_spark_referers
COMMENT "Apacke Sparkページにリンクしているクリック数トップページを含むテーブル(ゴールド)"
AS SELECT
  previous_page_title as referrer,
  click_count
FROM live.clickstream_prepared
WHERE current_page_title = 'Apache_Spark'
ORDER BY click_count DESC
LIMIT 10;
```

このノートブックが完成したら、いよいよDLTパイプラインの作成と実行です。

## DLTパイプラインの作成

1. 左のサイドメニューの**ワークフロー**ボタンをクリックするところからスタートします。ページの上部に**Delta Live Tables**タブが表示されます。
1. **パイプラインを作成**をクリックします。
1. **パイプライン名**と**ノートブックライブラリ**は必須の値です。ノートブックセレクターを用いて、上で作成した**dlt-quickstart-sql**のノートブックを選択してください。
![Screen Shot 2022-12-13 at 10.13.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2467a45f-beb4-b567-cb02-de8eb63bdf8d.png)
1. 上で取得した`storage_location`をパイプラインの**ストレージの場所**に貼り付けてください。
    - **ストレージの場所**は、パイプラインによって生成されるデータ、メタデータ、ログが記録される場所となります。実運用においては、セキュアな外部オブジェクトストレージを指定するようにしてください。
1. 上のセルで表示された`database`を**ターゲットスキーマ**に貼り付けてください。
    - **ターゲットスキーマ**は、パイプラインによって生成されたデータを格納するデータベースを保持します。データベースに格納することで、データに対して容易にクエリーを行えるようになります。
![Screen Shot 2022-12-13 at 10.13.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c0ddcfb6-713e-ea7a-54dd-d3f05b47bd52.png)
1. **作成**をクリックします。これでDLTパイプラインが作成されました。

## DLTパイプラインの実行

1. DLTパイプラインが表示されます。
![Screen Shot 2022-12-13 at 10.14.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fb3aa751-4681-8391-4443-4a5fa092dca8.png)
1. 可能であればパイプラインを**開発**モードで実行します。これによって、実行の合間で同じクラスターを再利用でき、ジョブ失敗時の自動リトライを無効化しますので、デバッグが容易となります。
1. **開始**をクリックしてテーブルに対する最初の更新をスタートします。Delta Live Tablesは、必要な全てのインフラストラクチャーを自動でデプロイし、全てのデータセット間の依存関係を解決します。

    :::note
    **注意**
    依存関係の解決とインフラストラクチャのデプロイのため、最初のテーブルアップデートは数分かかる場合があります。
   :::

1. 数分経つとクラスターが配備され、パイプラインのDAG(有効非巡回グラフ)が表示され、パイプラインが実行されます。
![Screen Shot 2022-12-13 at 10.17.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f51023aa-e7b2-eb0f-79d3-d563a67714e2.png)
1. すべてのテーブルに完了と表示されたらパイプラインが完了したことを意味します。エラーが生じた際には、該当テーブルが赤くなります。
![Screen Shot 2022-12-13 at 10.20.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f9fa51f4-9815-0c4f-c633-03f6c745b212.png)
1. エクスペクテーションが設定されているテーブルをクリックすると、エクスペクテーションを手寄与した結果を確認することができます。
![Screen Shot 2022-12-13 at 10.20.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2bfec258-85bb-18f4-d722-9c572d34a927.png)

## テーブルの確認

DLTの処理結果はすべてDeltaテーブルとして保持されます。**DLT-main**ノートブックに以下のような関数を作成することで、クイックにDeltaテーブルにアクセスすることができます。

```py:Python
def query_table(table_name):
    return spark.sql(f"SELECT * FROM delta.`{storage_location}/tables/{table_name}`")
```

### ブロンズテーブル
![Screen Shot 2022-12-13 at 10.27.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0ed2d6c1-e82c-a3cb-09f9-89896c0a8e64.png)

### シルバーテーブル
![Screen Shot 2022-12-13 at 10.27.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d13426ab-8eb4-0323-8eb0-bf35a75d5656.png)

### ゴールドテーブル
![Screen Shot 2022-12-13 at 10.27.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b6636931-e395-c28b-5c25-0125b3784bb7.png)

# まとめ

Delta Live Tables(DLT)は、背後でSparkの構造化ストリーミングとDelta Lakeを活用しています。DLTを用いることで、ユーザーはSparkやDeltaの詳細を意識することなしに、容易にデータパイプラインを開発、運用できるようになります。本書では触れませんでしたが、DLTの処理結果はすべてイベントログとして、これらもDeltaテーブルとして保持されますので、これらを解析、監視することも可能です。データエンジニアリングに課題を感じられているのであれば、是非DLTを御試しください。この他にも以下のような機能をサポートしています。

- [Databricks Delta Live Tablesでチェンジデータキャプチャをシンプルに \- Qiita](https://qiita.com/taka_yayoi/items/bdf72aaa087190f98f54)
- [Delta Live Tablesクックブック \- Qiita](https://qiita.com/taka_yayoi/items/184d872392ac4fb7fb38)
- [Delta Live Tablesにおけるストリームデータ処理 \- Qiita](https://qiita.com/taka_yayoi/items/62983c6d48abd843727a)
- [Delta Live TablesとApache Kafkaを用いた低レーテンシーストリーミングデータパイプライン \- Qiita](https://qiita.com/taka_yayoi/items/541681049c58f308f27c)
- [Delta Live Tablesのイベントログ \- Qiita](https://qiita.com/taka_yayoi/items/5ae5824cfab30caf9795)

### Databricks 無料トライアル

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