Databricks にて replaceWhere による選択的上書き時にソースデータフレームの制約によるエラーへの対応方法

## 概要

Databricks にて replaceWhere による選択的上書き時にソースデータフレームの制約による以下のようなエラーへの対応方法を共有します。選択的上書きとは、replaceWhere にて指定した条件（例：`created_date > '2024-04-01'`）のデータを上書きする機能なのですが、ソースのデータフレームにおいてその条件を満たさない場合（例：`created_date`が`'2024-04-01'`のデータが含まれる場合）にはエラーとなるようです。エラーの再現方法と対応方法を共有します

```text
AnalysisException: Data written out does not match replaceWhere 'created_date > '2024-04-01''.
CHECK constraint EXPRESSION(('created_date > 2024-04-01)) (created_date > '2024-04-01') violated by row with values:
 - created_date : 19814
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e825330a-cb21-362a-656e-6701b4e5a749.png)

本仕様についてはドキュメントにて下記のように説明されています。

> すべての行が述語に一致することを検証し、overwrite セマンティクスを使用してアトミックな置換を実行します。 操作内の値が制約の範囲外にある場合、この操作は既定でエラーが発生して失敗します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/72a20ca4-1b4f-42b5-fb80-1f7922148023.png)

引用元：[Delta Lake を使用してデータを選択的に上書きする - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/delta/selective-overwrite)

本記事では、下記の対応方法を紹介します。

1. フィルタリングにより対応する方法
1. Spark conf により制約チェックをスキップする方法

## 事前準備

### 1.  ライブラリの呼び出しや変数を定義

```python
import datetime

schema_name = "replace_test_02"
table_name = "table_01"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/91e85979-6aba-8ddb-d213-9cfb403b50cf.png)

### 2. スキーマとテーブルを作成

```python
spark.sql(f"create schema if not exists {schema_name};")
```

```python
schema = """
ID int,
NAME string,
created_date date
"""

spark.sql(
    f"""
    CREATE OR REPLACE TABLE {schema_name}.{table_name}
    (
        {schema}
    )
    """
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/de5ce513-3378-85e2-1bf8-e9fa7237e327.png)


## エラーの再現

### 1. 書き込み用のデータフレーム（`created_date`が`'2024-04-01'`のデータが含む）を作成

```python
base_data = [
    {
        "ID": 1,
        "NAME": "aaa",
        "created_date": datetime.date(2024, 4, 1),
    },
    {
        "ID": 2,
        "NAME": "bbb",
        "created_date": datetime.date(2024, 4, 2),
    },
    {
        "ID": 3,
        "NAME": "ccc",
        "created_date": datetime.date(2024, 4, 3),
    },
]
base_df = spark.createDataFrame(base_data, schema)
base_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c3d2a856-93ca-be84-41d2-654ade1da7d0.png)

### 2. `created_date > '2024-04-01'`の条件で選択上書きを実施するとエラーとなることを確認

```python
replace_where_cond = "created_date > '2024-04-01'"

(
    base_df.write.mode("overwrite")
    .option("replaceWhere", replace_where_cond)
    .saveAsTable(f"{schema_name}{table_name}")
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e825330a-cb21-362a-656e-6701b4e5a749.png)


## エラーへの対応方法

### 1. フィルタリングにより対応する方法

#### 1. `created_date > '2024-04-01'`の条件でソースデータフレームをフィルタリング

```python
filtered_df = base_df.where(replace_where_cond)
filtered_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b7d5a941-be67-399c-bce2-a0c47879a965.png)

#### 2. 選択上書きが正常終了することを確認

```python
(
    filtered_df.write.mode("overwrite")
    .option("replaceWhere", replace_where_cond)
    .saveAsTable(f"{schema_name}{table_name}")
)

table_df = spark.table(f"{schema_name}{table_name}")
table_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/044c5cd0-73c4-f78d-f52a-ec7a20daadb6.png)


### 2. Spark conf により制約チェックをスキップする方法

#### 1. テーブルのデータを TRUNCATE 

```python
spark.sql(f"TRUNCATE TABLE {schema_name}{table_name}")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ab0fe713-71e2-e932-2c2f-2603f528031b.png)


#### 2. 制約チェックを実施するように Spark の設定を変更

```python
spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/74e4d296-d060-4f74-a413-4801d548ff19.png)

#### 3. 選択上書きが正常終了することを確認

```python
(
    base_df.write.mode("overwrite")
    .option("replaceWhere", replace_where_cond)
    .saveAsTable(f"{schema_name}{table_name}")
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b94efd41-c1d5-5414-51fe-38d0011946be.png)