Google Colab の Spark にて Apache Iceberg の CoW と MoR の動作検証

## 概要

Google Colab の Spark にて Apache Iceberg の**Copy-On-Write (CoW)** と **Merge-On-Read (MoR)** の動作検証結果を共有します。Iceberg ではテーブルの行レベル更新や削除の処理方式として **Copy-On-Write** と **Merge-On-Read** の 2 つをサポートしています。それぞれ次のような特徴があります。MoR 方式を選択した場合は定期的なコンパクション（Compaction）が必要になり、運用コストが上がる点に注意が必要です。

- **Copy-On-Write (CoW)**  
  更新・削除が発生すると、該当のデータファイル全体をコピーして書き換え、新しいファイルを作成する方式。  
  - **メリット**: 読み取り時は常に最新のファイルのみを参照すればよく、シンプルで高速  
  - **デメリット**: 書き込み時にファイルの再書き出しが必要になるため、更新負荷が高い

- **Merge-On-Read (MoR)**  
  既存のデータファイルはそのまま残し、更新・削除データを新規ファイル（および削除ファイル）として書き込む方式。  
  - **メリット**: 書き込み時は既存ファイルを直接書き換えないため高速  
  - **デメリット**: 読み取り時にマージ処理が必要となり、ファイル数が増えるほどパフォーマンスに影響する可能性がある

## 検証結果

### Spark のバージョンを確認

```bash
!pyspark --version
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9a0acea6-ad83-4efc-abbd-78f9dcffa8b2.png)

### SparkSession を定義

Spark のバージョンに応じて`spark.jars.packages`で指定するライブラリのバージョンを変更する必要あり。

```python
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("IcebergExample")
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.0")
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.spark_catalog.type", "hadoop")
    .config("spark.sql.catalog.spark_catalog.warehouse", "/iceberg")
    .getOrCreate()
)
spark
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/40d88dfb-888f-4df3-b133-71bfaffbaa2a.png)

### Namespace を作成

```python
spark.sql("CREATE NAMESPACE IF NOT EXISTS  iceberf_gc ;")
```

```python
spark.sql("SHOW NAMESPACES").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/747ae6b6-0cd1-48ff-a9a0-41e5094f37a8.png)

### テーブルを作成

```python
spark.sql("""
CREATE OR REPLACE TABLE iceberf_gc.sample_cow (
  id INT,
  name STRING
)
USING iceberg
TBLPROPERTIES (
  'format-version' = '2',
  'write.delete.mode' = 'copy-on-write',
  'write.update.mode' = 'copy-on-write',
  'write.merge.mode' = 'copy-on-write'
);
""")


spark.sql("""
-- MoR モードのテーブル作成
CREATE OR REPLACE TABLE iceberf_gc.sample_mor (
  id INT,
  name STRING
)
USING iceberg
TBLPROPERTIES (
  'format-version' = '2',
  'write.delete.mode' = 'merge-on-read',
  'write.update.mode' = 'merge-on-read',
  'write.merge.mode' = 'merge-on-read'
);
""")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/564c58e6-9a0d-4e9a-ada3-5072735bf959.png)


### データの挿入とデータの削除を実施

```python
spark.sql("""
INSERT INTO iceberf_gc.sample_cow VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie');
""")

spark.sql("""
INSERT INTO iceberf_gc.sample_mor VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie');
""")
```

```python
spark.sql("""
-- CoW テーブルでの DELETE
DELETE FROM iceberf_gc.sample_cow WHERE id = 2;
""")

spark.sql("""
-- MoR テーブルでの DELETE
DELETE FROM iceberf_gc.sample_mor WHERE id = 2;
""")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/87432b55-6666-42b7-8a0b-d4b7ec3d4664.png)


### 2 つのテーブルの files を確認

```python
display(spark.table("iceberf_gc.sample_cow.files").toPandas())
```

```python
display(spark.table("iceberf_gc.sample_mor.files").toPandas())
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f74a5b5-bb01-4173-8666-655a951b9dd1.png)

### 2 つのテーブルの data ディレクトリを確認

```bash
!ls /iceberg/iceberf_gc/sample_cow/data
```

```bash
!ls /iceberg/iceberf_gc/sample_mor/data
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/05669ad1-4154-41e9-9b7d-175b7e2b61fe.png)