Delta Live Tablesを用いたインクリメンタルなデータの取り込み

こちらでは、ダミーデータを使ってDelta Live Tables(DLT)のインクリメンタルなデータの取り込みをデモします。

# Delta Live Tablesとは

データパイプラインの変換処理を全て記述することなしに、パイプラインの各ステップ(テーブル)を**宣言**することで、Delta Live Tablesはテーブル間の処理を補完するので、シンプルなコードで複雑なパイプラインを構築することができます。SQLとPythonをサポートしているので、得意な言語でデータパイプラインを構築できます。また、イベントログ、エラー時のリトライ、自動テスト、オートスケーリングなどがサポートされています。処理の背後ではSparkの構造化ストリーミングやDelta Lakeを活用しています。

https://www.databricks.com/jp/product/delta-live-tables

# 前提条件

- SQLを使用します。
- あるパスにParquetが逐次配置され、新規に到着したファイルのみを処理します。これが増分のみを処理する[インクリメンタルなデータ取り込み](https://qiita.com/taka_yayoi/items/ba4b60024ec209e82a14)です。
- 取り込むデータのカラムに空白が含まれているので、[カラムマッピング](https://qiita.com/taka_yayoi/items/50c18bc9e48a31b22951)の機能を活用します。
- 取り込んだデータは[Untiy Catalog](https://qiita.com/taka_yayoi/items/15aede468bdca58ec6a3)で管理します。

# 注意事項

Delta Live Tablesのテーブルの保存先をUnity Catalogにする際、いくつかの制限事項が存在します。

- Unity Catalog配下に作成されたDelta Live Tablesのテーブルは、共有モードのUnity Catalog対応クラスター、あるいはSQLウェアハウスからのみクエリーすることができます。ユーザー割り当てモード、分離なしモードの共有クラスターからクエリーすることはできません。
- 共有モードのクラスターでは機械学習ランタイムはサポートされていません。

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

# データの準備

```py
# データ保存先パス
data_path = "/tmp/takaaki.yayoi@databricks.com/dlt/landing"
# ランディングゾーンを初期化
dbutils.fs.rm(data_path, True)
```

データを読み込みます。このデータのカラム名には空白が含まれています。

```py
df = spark.read.csv("/databricks-datasets/wine-quality/winequality-white.csv", sep=";", header=True)
display(df)
```
![Screenshot 2023-07-13 at 16.57.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/684b7b79-5217-3ec6-605e-3195085b3674.png)

データを保存します。

```py
save_path = f"{data_path}/2023/07/06/"
df.write.format("parquet").option("header", "true").mode("overwrite").save(save_path)
```

```py
display(dbutils.fs.ls(f"{data_path}/2023/07"))
```
![Screenshot 2023-07-13 at 16.58.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/afaa016b-95be-6d4a-1f91-bfa7375b4e4b.png)

# DLTパイプラインの定義

SQLノートブックを作成して以下の内容を記述します。`TBLPROPERTIES`を指定してカラムマッピングを有効化し、処理時刻のカラムを追加しています。

```sql:dlt_incremental_load_sql
CREATE
OR REFRESH STREAMING TABLE bronze TBLPROPERTIES (
  'delta.minReaderVersion' = '2',
  'delta.minWriterVersion' = '5',
  'delta.columnMapping.mode' = 'name'
) COMMENT "Parquetをそのまま保持するブロンズテーブル" AS
SELECT
  *,
  current_timestamp() as processed -- 処理時刻
FROM
  cloud_files(
    "/tmp/takaaki.yayoi@databricks.com/dlt/landing/",
    "parquet"
  )
```

# パイプラインの実行

1. サイドメニューの**Delta Live Tables**をクリックします。
![Screenshot 2023-07-13 at 16.59.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d154dbe7-c3ed-118d-5492-a6cb368b6a2b.png)
1. **パイプラインを作成**ボタンをクリックします。
1. パイプライン名を入力します、**ソースコード**で、上で作成したノートブックを選択します。
1. **配信先**でUnity Catalogを選択し、カタログを選択、ターゲットスキーマでスキーマ(データベース)を選択します。ここにテーブルが作成されます。
![Screenshot 2023-07-13 at 15.55.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6456a210-ace7-7b66-b110-575982f0575c.png)
1. **作成**をクリックします。
1. **開始**をクリックします。
1. 処理が完了すると、テーブルがグリーンになります。
![Screenshot 2023-07-13 at 17.04.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/dc2d709f-2f21-7611-2a50-6494f33c4f39.png)

# テーブルの確認

1. サイドメニューから**SQLエディタ**にアクセスし、稼働中のSQLウェアハウスを選択します。
1. 以下のクエリーを実行します。
    ```sql
    SELECT * FROM takaakiyayoi_catalog.dlt.bronze ORDER BY processed DESC
    ```
1. 結果を確認することができます。
![Screenshot 2023-07-13 at 17.07.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5dacfedb-adf3-b778-921e-c7cae97376db.png)
1. 件数も確認します。
![Screenshot 2023-07-13 at 17.07.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/65ce1886-1c36-40b2-9640-8aa781050e45.png)

なお、再度パイプラインを実行しても、元のデータに増分がないので処理は行われません。
![Screenshot 2023-07-13 at 17.08.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d364cd1a-6891-f2e9-17c8-e3072e76d864.png)

# データの追加

同じデータをパスを変えて保存します。

```py
save_path = f"{data_path}/2023/07/07/"
df.write.format("parquet").option("header", "true").mode("overwrite").save(save_path)
```

```py
display(dbutils.fs.ls(f"{data_path}/2023/07"))
```
![Screenshot 2023-07-13 at 17.10.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0a33f6c5-8fc5-01b4-c216-618d7c03b02e.png)

# 再度パイプラインの実行

Delta Live Tablesのパイプラインに移動し、**開始**をクリックします。今回は増分のデータが処理されます。

![Screenshot 2023-07-13 at 17.12.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/af9f807f-de51-6178-955a-05bd7c21487d.png)

件数も2倍になっています。
![Screenshot 2023-07-13 at 17.12.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8019b5cc-0b5b-d1a4-fe2f-b05cd7f3d053.png)

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

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


### Databricks無料トライアル

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