Databricks の Delta Live Table にて変更データ キャプチャ (CDC) を SCD TYPE 2 として実行した際に Timestamp 型のカラムで履歴管理が可能

## 概要
Databricks の Delta Live Table （DLT）にて変更データ キャプチャ (CDC) を SCD TYPE 2 として実行した際に Timestamp 型のカラムで履歴管理が可能です。DLT では APPLY CHANGES API で SCD TYPE 2 でのデータ連携が可能なのですが、int 型でしか履歴管理できないのではないかと勘違いしていました。実際には、 APPLY CHANGES API 実行時の`sequence_by`パラメータで指定したデータ型となるようです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f336e574-5be4-3220-be77-57b01d940a11.png)

勘違いしていた背景としては、下記のドキュメントにて`__START_AT`列と`__END_AT`列が INT 型となっていたことです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c81fee58-290e-4feb-ae37-4e940a26b32c.png)

下記データの Timestamp 型の列（`update_timestamp`列）を`sequence_by`パラメータで指定したところ、 Timestamp 型での履歴管理も実施できることを確認できました。本記事では、INT 型と Timestamp 型の２つのカラムで、 SCD TYPE 2 の処理を実施する手順を紹介します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f6233046-6f2e-cc11-e190-90c339e6da73.png)


## 検証コードと実行結果

### 事前準備

#### 1. カタログとスキーマを作成

```sql
%sql
CREATE CATALOG IF NOT EXISTS manabian_test;
CREATE SCHEMA IF NOT EXISTS manabian_test.dlt_sample;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/da507dbc-536a-b9dd-95a5-809d25f2cf50.png)

### 2. ソーステーブルの作成

```sql
df = spark.sql("""
SELECT '1' AS userId, 'Isabel' AS name, 'Monterrey' AS city, 'INSERT' AS operation, 1 AS sequenceNum, CAST('2024-01-01 00:00:00' AS TIMESTAMP) AS update_timestamp
UNION ALL
SELECT '1' AS userId, 'Isabel' AS name, 'Chihuahua' AS city, 'UPDATE' AS operation, 5 AS sequenceNum, CAST('2024-05-01 00:00:00' AS TIMESTAMP) AS update_timestamp
UNION ALL
SELECT '1' AS userId, NULL AS name, NULL AS city, 'DELETE' AS operation, 6 AS sequenceNum, CAST('2024-06-01 00:00:00' AS TIMESTAMP) AS update_timestamp
UNION ALL
SELECT '2' AS userId, 'Raul' AS name, 'Oaxaca' AS city, 'INSERT' AS operation, 1 AS sequenceNum, CAST('2024-01-01 00:00:00' AS TIMESTAMP) AS update_timestamp
UNION ALL
SELECT '3' AS userId, 'Mercedes' AS name, 'Tijuana' AS city, 'INSERT' AS operation, 2 AS sequenceNum, CAST('2024-02-01 00:00:00' AS TIMESTAMP) AS update_timestamp
UNION ALL
SELECT '3' AS userId, 'Mercedes' AS name, 'Mexicali' AS city, 'UPDATE' AS operation, 5 AS sequenceNum, CAST('2024-05-01 00:00:00' AS TIMESTAMP) AS update_timestamp
UNION ALL
SELECT '3' AS userId, 'Mercedes' AS name, 'Guadalajara' AS city, 'UPDATE' AS operation, 6 AS sequenceNum, CAST('2024-06-01 00:00:00' AS TIMESTAMP) AS update_timestamp
""")

temp_table_name = "_temp_table"
df.createOrReplaceTempView(temp_table_name)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eb06e296-b6be-54d4-9c8a-34d59060ed41.png)

```sql
%sql
CREATE OR REPLACE TABLE manabian_test.dlt_sample.source_table_01
AS
SELECT
  *
  FROM
    _temp_table
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dbaaf351-97a9-7f90-c545-75e26e7d423b.png)

### DLT によるデータ実行

#### 1. DLT のコードを記述

```python
import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def source_table_01():
  return spark.readStream.table("manabian_test.dlt_sample.source_table_01")
```

```python
dlt.create_streaming_table("target_table_by_update_timestamp")
dlt.apply_changes(
  target = "target_table_by_update_timestamp",
  source = "source_table_01",
  keys = ["userId"],
  sequence_by = col("update_timestamp"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)
```

```python
dlt.create_streaming_table("target_table_by_sequenceNum")
dlt.apply_changes(
  target = "target_table_by_sequenceNum",
  source = "source_table_01",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/11ce31e9-31eb-78d0-ce11-6d605332b755.png)

#### 2. DLT を実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fce331d2-f1e3-a516-87b5-808c73076cb3.png)

### 処理結果の確認

#### 1. Timestamp 型のカラム（`update_timestamp`列）を `sequence_by`パラメータで指定した場合の処理結果

```sql
%sql
SELECT * FROM manabian_test.dlt_sample.target_table_by_update_timestamp;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/685632fc-9644-7105-a55e-57dfeb34104d.png)

#### 2. INT 型のカラム（`sequenceNum`列）を `sequence_by`パラメータで指定した場合の処理結果

```sql
%sql
SELECT * FROM manabian_test.dlt_sample.target_table_by_sequenceNum;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a64c27fe-a32d-d1e5-b8e7-b9f4b53e62cd.png)