Databricks ( Spark )にて変更データフィード（Change Data Feed、CDF） 適用前の時点に対するクエリエラーへの対応方法

## 概要

Databricks ( Spark )にて変更データフィード（Change Data Feed、CDF） 用前のテーブルバージョンに対して、CDF 機能を用いてデータフレームを作成する際のエラーへの対応方法を共有します。

CDF 有効化前の時点でのデータフレームを作成しようとすると次のようなエラーが発生します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d9d61dc0-ef5f-2236-8940-26176bf74515.png)

```python
df = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 0)
    .table(f"{db_name}.{tbl_name}")
)
df.display()
```

> AnalysisException: Error getting change data for range [0 , 4] as change data was not
recorded for version [0]. If you've enabled change data feed on this table,
use `DESCRIBE HISTORY` to see when it was first enabled.
Otherwise, to start recording change data, use `ALTER TABLE table_name SET TBLPROPERTIES
(delta.enableChangeDataFeed=true)`.

ドキュメントに下記のような記載があるため、このような動作は仕様のようです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ea1c7437-4d1f-2a52-08c4-f995b7503094.png)

引用元：[バッチ クエリの変更を読み取る - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/delta/delta-change-data-feed#read-changes-in-batch-queries)

本エラーが発生しないようにする方法の１つである、テーブルにて保持させた更新日の監査列に基づき、CDF が有効化されたバージョン以降のバージョンを取得する方法を紹介します。データ書き込み後に CDF を有効にした場合には、本記事で紹介する方法では同様に回避策にはなりません。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1110567f-b6d5-d3e9-c52e-3676e6626b6a.png)


CDF を有効化した以降の時点、かつ、Delta Log の最終処理以降の時点に対するデータに基づき処理を行う場合には、`spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled` プロパティによりエラーを回避できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ab7c236e-fd6a-c990-9915-afb2eebe49a9.png)

> IllegalArgumentException: Provided Start version(1000) for reading change data is invalid. Start version cannot be greater than the latest version of the table(4).

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1d36a60e-12fe-f799-031a-dfa9f49391a5.png)

引用元：[Azure Databricks で Delta Lake 変更データ フィードを使用する - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/delta/delta-change-data-feed#read-changes-in-streaming-queries)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e62418de-1a4c-9267-2cfc-035625ac9cc3.png)

## 対応方法
### 1. 事前準備

```python
from pyspark.sql.functions import current_timestamp

# データを読み込み
filepath = "dbfs:/databricks-datasets/tpch/data-001/nation/"
 
schema = """
  N_NATIONKEY  integer
  ,N_NAME       string
  ,N_REGIONKEY  integer 
  ,N_COMMENT    string
"""
 
df = (
    spark.read.format("csv")
    .schema(schema)
    .option("inferSchema", "True")
    .option("sep", "|")
    .load(filepath)
)
 
# 更新日列の監査列を付与
df = df.withColumn(
    "_update_timestamp",
    current_timestamp(),
)
 
df.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/208885ea-96d6-2fd0-b335-d3c533be8730.png)


```python
db_namedb_name = "_qiita"
tbl_name = "nation"

# データベースを作成
sql = f"""
CREATE SCHEMA IF NOT EXISTS {db_name}
"""
spark.sql(sql)
 
# テーブル作成
sql = f"""
DROP TABLE IF EXISTS {db_name}.{tbl_name}
"""
spark.sql(sql)
 
sql = f"""
CREATE TABLE IF NOT EXISTS {db_name}.{tbl_name}
(
  N_NATIONKEY  integer,
  N_NAME       string,
  N_REGIONKEY  integer, 
  N_COMMENT    string,
  _update_timestamp timestamp
)
USING delta
"""
spark.sql(sql)

# 変更データフィードを有効化
sql = f"""
ALTER TABLE {db_name}.{tbl_name}
  SET TBLPROPERTIES (
    delta.enableChangeDataFeed = true
  )
"""
spark.sql(sql)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3761ff48-9d31-a40a-d02e-aa08c60074b8.png)


```python
spark.sql(
    f"""
SELECT
  *
EXCEPT(
  userName
)
FROM (
  DESC HISTORY {db_name}.{tbl_name}
)
"""
).display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6d2ca8aa-60db-91bb-998f-2e441a5cbb49.png)


```python
for i in range(3):
    df.write.mode("append").saveAsTable(f"{db_name}.{tbl_name}")

spark.sql(
    f"""
SELECT
  *
EXCEPT(
  userName
)
FROM (
  DESC HISTORY {db_name}.{tbl_name}
)
"""
).display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/61a90e69-ea00-10f3-5968-f5a566debd39.png)


### 2. CDF 機能に基づきデータフレームを作成

```python
df = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 1)
    .table(f"{db_name}.{tbl_name}")
)
df.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0cc815a5-a5e6-be36-257e-18c652dbd8c6.png)


### 3. エラーを再現

```python
tgt_version  = 0
 
df = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", tgt_version)
    .table(f"{db_name}.{tbl_name}")
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6096687f-ac97-a97d-f622-5ad10b5fb4db.png)


### 4. エラー回避策

```python
min_ts = (
    spark.table(f"{db_name}.{tbl_name}")
    .selectExpr("MIN(_update_timestamp) as min")
    .first()[0]
)
print(min_ts)
 
sql = f"""
select
  *
  FROM (
    DESC HISTORY {db_name}.{tbl_name}
  )
  WHERE
    timestamp >= '{min_ts}'
  ORDER BY
    version ASC
"""

tgt_version  = 0

if tgt_version >= min_delta_version:
    starting_version = tgt_version
else:
    starting_version = min_delta_version

df = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", starting_version)
    .table(f"{db_name}.{tbl_name}")
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f2a55de6-289f-2409-9753-b8ac0c3bdc73.png)

### 5. 本手順で作成したリソースを削除

```python
spark.sql(f'DROP DATABASE {db_name} CASCADE')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dad99cf1-b1dc-6c42-9d67-ef192635a5e7.png)