Delta Live Tablesで更新ファイルを取り込んでテーブルの洗い替えを行う

ニッチなネタですが。

# シナリオ

日次で特定のS3パスにファイルが到着します。

>.../2023/07/24: 7/23時点の全件データ断面
.../2023/07/25: 7/24時点の全件データ断面

断面データを用いてターゲットテーブルを洗い替えたいものとします。

# Delta Live Tablesとは

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

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

# 方針

- 日次でフォルダが作成されていくので、自動で未処理のファイルのみを取り込みたい。
- ターゲットテーブルに追記するのではなく、一旦レコードをDELETEしてINSERTしたい。

実は、2つ目が結構苦戦しました。Delta Live Tablesは基本追記処理となります(DELETE込みのUPSERTもあります)。

追記ではなく、洗い替えをする手段として**フルリフレッシュ**という機能があります。これはこれで便利なのですが、データソース自体もリフレッシュされてしまうので、結局のところ前日のファイルも読み込まれてしまう訳でして。

# 案1: 日毎のフォルダを特定してパイプラインをアップデート

```sql
CREATE
OR REFRESH STREAMING LIVE TABLE customer_bronze 
COMMENT "ブロンズレイヤ：データファイルをすべてString型でロードする" 
AS
SELECT
  *
FROM
  cloud_files(
    -- ディレクトリをパラメータ化
    "dbfs:/tmp/DLT_Example/customer_${mypipeline.startDate}",
    "csv",
    map(
      "delimiter",
      ",",
      "header",
      "true",
      "cloudFiles.schemaLocation",
      "dbfs:/tmp/DLT_Example/customer/schema_spend"
    )
  );
```

これはこれで動くのですが、毎回パラメータに日付を指定しなくてはなりません。

# 案2: Auto Loaderを使いつつもパイプラインアップデート前にテーブルをリセットする

結局こちらで行きました。ただ、注意点があります。デフォルトでは、Delta Live Tablesはパイプラインがアップデートされるたびに入力データに基づいてテーブルの結果を再計算するので、ソースデータから削除されたレコードが再ロードされないように抑制する必要があります。

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

https://qiita.com/taka_yayoi/items/8b98f37333c0ed621b32#%E6%89%8B%E5%8B%95%E3%81%AE%E5%89%8A%E9%99%A4%E3%82%84%E6%9B%B4%E6%96%B0%E3%82%92%E4%BF%9D%E6%8C%81%E3%81%99%E3%82%8B

> Delta Live Tablesを用いることで、テーブルから手動でレコードを削除、更新することができ、後段のテーブルの再計算を行うためにリフレッシュすることができます。
>
> デフォルトでは、Delta Live Tablesはパイプラインがアップデートされるたびに入力データに基づいてテーブルの結果を再計算するので、ソースデータから削除されたレコードが再ロードされないようにする必要があります。テーブルプロパティ`pipelines.reset.allowed`を`false`に設定することで、テーブルのリフレッシュを防ぎつつも、テーブルに対するインクリメンタルな書き込みやテーブルへの新規データの流入は妨げません。

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

```sql
CREATE
OR REFRESH LIVE TABLE complete_silver COMMENT "取り込まれたデータで洗い替えされるテーブル" AS
SELECT
  *
FROM
  live.complete_bronze;
```

このノートブックを用いてDLTパイプラインを作成します。ここではテーブルの配信先としてUnity Catalogのカタログ`takaakiyayoi_catalog`、スキーマ`dlt`を設定しています。
![Screenshot 2023-07-26 at 19.34.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cdc915e6-a238-0e45-666d-6492dce94879.png)

これで`complete_bronze`のリフレッシュ後の再ロードが抑制されるようになり、テーブルのレコード削除操作が維持されるようにます。

## テーブル削除処理の実装

あとは別途削除処理を実装します。別途Databricks SQLのクエリーとして`table_reset`を作成します。サイドメニューから**クエリー**にアクセスし、以下のクエリーを記述します。カタログ名とスキーマ名はDelta Live Tablesの定義と揃えてください。

```sql:table_reset
DELETE FROM takaakiyayoi_catalog.dlt.complete_bronze;
DELETE FROM takaakiyayoi_catalog.dlt.complete_silver;
```

![Screenshot 2023-07-26 at 19.37.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/383383bc-862e-1227-2e12-a66ca441b732.png)

こちらもクエリーを実行して動作確認します。

## ジョブを構築

最後に、クエリーとDLTパイプラインをジョブとしてまとめます。サイドメニューの**ワークフロー**にアクセスして**ジョブを作成**をクリックします。

1. ジョブ名、タスク名を入力し、**種類**は**SQL**、**SQLタスク**は**クエリー**、**SQLクエリー**には[テーブル削除処理の実装](#テーブル削除処理の実装)で作成したクエリーを選択します。SQLウェアハウスを選択して**タスクを作成**をクリックします。
![Screenshot 2023-07-26 at 19.38.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2b3586a7-6b02-ffa2-9a12-23d9d9930ba7.png)
1. 一つ目のタスクが選択されている状態で、**タスクを追加**をクリックします。
1. タスク名を入力し、**種類**は**Delta Live Tablesパイプライン**を選択し、パイプラインで[DLTパイプラインの作成](#DLTパイプラインの作成)で作成したパイプランを選択します。**Delta Live Tablesパイプラインで完全なリフレッシュをトリガーする**にチェックが入っていることを確認してください。
![Screenshot 2023-07-26 at 19.40.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c05495f2-26b3-9131-8cd2-c47bcd4b48f0.png)

# 動作確認

## 1日目

データを準備します。ノートブックで以下を実行します。

```py
data_path = "/tmp/takaaki.yayoi@databricks.com/dlt/landing"
dbutils.fs.rm(data_path, True)
```

```py
data_07_06 = [{"Date": "2023/07/06", "Category": 'A', "ID": 1, "Value": 121.44, "Truth": True},
        {"Date": "2023/07/06", "Category": 'B', "ID": 2, "Value": 300.01, "Truth": False},
        {"Date": "2023/07/06", "Category": 'C', "ID": 3, "Value": 10.99, "Truth": None},
        {"Date": "2023/07/06", "Category": 'D', "ID": 4, "Value": 33.87, "Truth": True}
        ]

df_07_06 = spark.createDataFrame(data_07_06)
display(df_07_06)

data_07_07 = [{"Date": "2023/07/07", "Category": 'A', "ID": 1, "Value": 121.44, "Truth": True},
        {"Date": "2023/07/07", "Category": 'B', "ID": 2, "Value": 300.01, "Truth": False},
        #{"Date": "2023/07/07", "Category": 'C', "ID": 3, "Value": 10.99, "Truth": None},
        {"Date": "2023/07/07", "Category": 'D', "ID": 4, "Value": 33.87, "Truth": True}
        ]

df_07_07 = spark.createDataFrame(data_07_07)
display(df_07_07)

data_07_08 = [#{"Date": "2023/07/08", "Category": 'A', "ID": 1, "Value": 121.44, "Truth": True},
        #{"Date": "2023/07/08", "Category": 'B', "ID": 2, "Value": 300.01, "Truth": False},
        {"Date": "2023/07/08", "Category": 'C', "ID": 3, "Value": 10.99, "Truth": True},
        {"Date": "2023/07/08", "Category": 'E', "ID": 5, "Value": 33.87, "Truth": True}
        ]

df_07_08 = spark.createDataFrame(data_07_08)
display(df_07_08)
```

最初のデータを保存します。

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

Delta Live Tablesのパイプラインを**開始**してテーブルを作成します。動作確認も兼ねています。
![Screenshot 2023-07-26 at 22.03.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c6697d01-daca-cdb1-67ed-b9a710c61188.png)

処理が完了したらテーブルを確認します。7/6のデータのみが反映されています。
![Screenshot 2023-07-26 at 22.04.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4d9f3ead-e47e-7ab3-07f3-f44ab9d6747b.png)

## 2日目

ノートブックで2日目のデータを保存します。

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

2回目以降はテーブルのリセットが必要になるので、ジョブの画面に移動して、**今すぐ実行**をクリックします。
![Screenshot 2023-07-26 at 21.48.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6fe478ee-7a8a-d21f-608f-500593656fc2.png)

7/7のデータに洗い替えされます。
![Screenshot 2023-07-26 at 22.10.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c9d61bf8-122d-989a-2acc-eb87425b06ec.png)

## 3日目

ノートブックで3日目のデータを保存します。

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

ジョブの画面に移動して、**今すぐ実行**をクリックします。
![Screenshot 2023-07-26 at 21.48.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6fe478ee-7a8a-d21f-608f-500593656fc2.png)

7/8のデータに洗い替えされます。
![Screenshot 2023-07-26 at 22.11.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/043b9879-7378-84bd-7ed0-a52e94f1a99d.png)

あとは、上のジョブを日次実行するようにスケジュールすれば、毎日更新ファイルを取り込んでテーブルを洗い替えるようになります。

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

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


### Databricks無料トライアル

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