Delta Live TablesにおけるAuto Loaderとチェンジデータキャプチャの活用

こちらの続編です。

https://qiita.com/taka_yayoi/items/46c5e690f959893491a6

上の記事では、Delta Live Tablesでオートローダー(Auto Loader)を活用することで、新規に到着したデータのみを処理するパイプラインを簡単に構築できる様子を説明しました。

ここでは、さらに要件を追加します。

# 要件

- 取り込まれるデータにはキーが存在する。
- 取り込み済みのデータと取り込んだデータでキーによる照合を行い、一致するものがある場合には`UPDATE`、ない場合には`INSERT`を行う。いわゆるUPSERTを行う。
- 処理の順序をしているキーが存在するので、それに従ってUPSERTを行う。

# DLTのAPPLY CHANGES INTO

DLTでUPSERTを行うには`APPLY CHANGES INTO`を使用します。

https://qiita.com/taka_yayoi/items/5319b9b010ce4703e98a

```sql
APPLY CHANGES INTO LIVE.table_name
FROM source
KEYS (keys)
[WHERE condition]
[IGNORE NULL UPDATES]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2}]
```

今回使うのは以下の2つです。

- **KEYS**: ソースデータで行を一意に特定するカラム、あるいはカラムの組み合わせです。これは、ターゲットテーブルの特定のレコードに適用されるCDCイベントの特定に使用されます。この句は必須です。
- **SEQUENCE BY**: ソースデータにおけるCDCイベントの論理的順序を指定するカラム名です。Delta Live Tablesは、out-of-orderな変更イベントを取り扱うために、この順序を使用します。この句は必須です。

# データの準備

```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)
```
![Screenshot 2023-07-13 at 20.14.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/67d9b62a-9118-c3d1-04e2-2f4b8d0d062b.png)

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

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

# パイプラインの定義

ブロンズテーブルの定義は[前回](https://qiita.com/taka_yayoi/items/46c5e690f959893491a6)と同じです。

```sql
CREATE
OR REFRESH STREAMING TABLE cdc_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"
  )
```

これで新規データが毎回取り込まれるので、それに対してチェンジデータキャプチャを適用します。

```sql
CREATE OR REFRESH STREAMING TABLE cdc_silver COMMENT "変更点のみが反映されるシルバーテーブル";

APPLY CHANGES INTO LIVE.cdc_silver
FROM STREAM(live.cdc_bronze) -- ブロンズテーブルからストリーミング読み込み
KEYS (ID) -- IDでマッチング
SEQUENCE BY Date -- Date順に処理
```

# パイプラインの実行

![Screenshot 2023-07-13 at 20.18.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8a3c3480-f6a6-df19-fcf9-249206dd863c.png)

シルバーテーブルを確認します。

```sql
SELECT * FROM takaakiyayoi_catalog.dlt.cdc_silver ORDER BY ID ASC
```

7/6のデータを読み込んだ後に、7/7のデータの変更点が反映されています。
![Screenshot 2023-07-13 at 20.19.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6d9fe870-f835-c52d-2d8d-6d44594faf38.png)

# 追加データの読み込み

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

これは、ID3をUPDATEし、ID5をINSERTします。
![Screenshot 2023-07-13 at 20.20.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fdd433a6-4260-81d9-c2fc-60dc4ba056da.png)

再度DLTパイプラインを実行します。2行分のみが反映されます。
![Screenshot 2023-07-13 at 20.21.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0164a0a1-091f-64e3-5747-145717c0659a.png)

期待した通りにテーブルが更新されています。
![Screenshot 2023-07-13 at 20.22.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a00153bc-7c98-0b2a-8049-645f914530a2.png)

このような仕組みをスクラッチで構築するのは骨が折れますが、DLTを活用すればシンプルなコーディングだけで実現できます。是非お試しください！

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

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


### Databricks無料トライアル

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