Delta live tableを使って1つのjobで複数テーブルに分割する

# Delta Live Tableって？
Databricksが開発したPipelineツールです。
Notebook上に宣言的にSQL/Pythonで記述することができて

- pipelineのvisualize、Linageが自動で描ける
- ユーザ側でエラーハンドリングのコードなどは書く必要がない（Databricks側でよしなにやる）
- expectation(不正なデータを防ぐための条件が書ける。例えばageカラムのvalueが20以下のデータが入ってきたら、pipelineを停止する、エラーとして別カラムに書くなど）
- SQLがかければ、pipelineを簡単に作れる

といった感じです。
今回、1つのjobの中で、条件に応じて複数のテーブルを生成してみようと思います。

## 全体像
![Screenshot 2022-07-19 at 16.37.01.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/d24c1ba7-4446-6f73-1610-6e3d0b268c1d.jpeg)

全体像はこんな感じで、S3にあるデータセットからデータ読み込んで、必要なカラムを抽出して、最終的にCustID(顧客番号)ごとにテーブルを分割します。

### コードの説明

コードの全体像はこんな感じです。

![Screenshot 2022-07-19 at 16.48.13.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/900613c0-5f51-8d94-99f1-0c15557bae5d.jpeg)

### Brozeテーブルの生成(cmd1 / cmd2の箇所)
必要なlibraryをimportしたのち、関数(retail_raw)としてspark.readを呼び出して、csvファイルをloadしています。

使ったデータはdatabricksにあるサンプルデータセットで、onlineショップの販売データです。

![Screenshot 2022-07-19 at 16.54.49.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/6ec74023-afaf-fcb7-fba2-8f115add6213.jpeg)



```python
import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import *

csv_path = "/databricks-datasets/online_retail/data-001/data.csv"

@dlt.table(
  comment="オンラインショップの注文データ, ingested from /databricks-datasets."
)

def retail_raw():
  return (
    spark.read.format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load(csv_path)
  )

```


### siliverテーブルの生成　(cmd3)

cmd2で定義したretail_rawからデータを読み込み、関数retail_preparedを定義します。
withColumnRenamedで適度にcolumn名を変更して、最後にselectで必要なデータだけを読み出しています。


```python
@dlt.table(
  comment="分析用にテーブルの加工をします"
)
def retail_prepared():
  return (
    dlt.read("retail_raw")
      .withColumnRenamed("StockCode", "zaikoNO")
      .withColumnRenamed("CustomerID", "CustID")
      .select("InvoiceNo", "InvoiceDate", "zaikoNO", "CustID")
  )
```

### goldテーブルの生成(cmd4以降)

こんな感じで、関数(CustID14,15）を定義して、filterメソッドとつかってCustIDを分割しています。


```python
@dlt.table(
  comment="CustID14のみのテーブル"
)
def CustID14():
  return (
    dlt.read("retail_prepared")
      .filter(expr("CustID LIKE '14%'"))
      .select("CustID", "InvoiceDate")
  )

@dlt.table(
  comment="CustID15のみのテーブル"
)
def CustID15():
  return (
    dlt.read("retail_prepared")
      .filter(expr("CustID LIKE '15%'"))
      .select("CustID", "InvoiceDate")
  )

```

これでとりあえず、種となるnotebookは完了です。


## Delta live tableのjobを作る

上の手順で作成したNotebookをDelta live tableに登録します。
Workflowの画面から

- 先ほど作成したnotebookを指定
- Targetに書き込み先のDatabase名を指定

![Screenshot_2022-07-19_at_16_59_45.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/53247ea9-9c01-0576-5a8b-6440d1fd3deb.jpeg)



保存後、startボタンを押すとjobが実行され、pipelineが描画されます。

![Screenshot_2022-07-19_at_17_04_18.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/efe103f9-b627-1379-3fd1-cf596dbb5d60.jpeg)

画面右側にはそれぞれの関数(retail_rawなど）のスキーマ情報、処理件数、エラーがどのくらいあったかどうかなどが表示されます。
画面下側にあるAll/info/warning/errorの箇所は、pipelineのエラーが記録され、なにかあればそこからdebugをする形になります。
画面右上の箇所から、スケジュール実行や、設定変更が可能です。

## 書き込みの結果を見てみる
jobを実行した結果、Deltalake形式のテーブルが自動で生成されるので、確認してみます。

DBSQL(Redashベース）から見てみたいと思います。
各関数で定義した名前で、deltalakeのテーブルが生成されています。

![Screenshot 2022-07-19 at 17.13.35.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/bd8fd049-ae40-cb61-83c3-96ddd598199e.jpeg)

SQLを書いて中身を見てみたいと思います。

![Screenshot 2022-07-19 at 17.15.35.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/f8089b01-070e-8fce-f4eb-f5aede2a8580.jpeg)

![Screenshot 2022-07-19 at 17.15.19.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/33b15481-c28a-b695-752a-dbe04c658a9f.jpeg)


## 最後に

簡単ではありますが、こう言ったふうにpython/sqlでpipelineを簡単に定義できるのと、1つのjobで複数テーブルに書き込めるのはなかなか便利かなとおもうので、ぜひ興味あるひとは使ってみてください。