Databricks 上で Apache Iceberg テーブルを CLONE する機能の検証結果

## 概要
Databricks 上で Apache Iceberg テーブルを CLONE する機能を検証した結果を共有します。Google Colab で作成したテーブルに対して以下の操作（INSERT、DELETE、UPDATE、MERGE、TRUNCATE TABLE）を行い、そのたびに Databricks で CLONE を実施し期待どおりに動くかを確認しました。  
今回はストレージを Azure Blob Storage とし、Google Colab から Iceberg テーブルを作成し、その後 Databricks から何度も CLONE を行う手順を紹介します。

Iceberg テーブルを CLONE する場合には制約があることに注意してください。特に Merge-On-Read (MoR) により書き込まれている場合には動作しないようです。

> 更新、削除、またはマージが発生した Iceberg の読み取り時マージ テーブルを複製することはできません。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6c274ef8-d7c6-4515-9955-a375cce6209e.png)

CoW と MoR については以下の記事を参照してください。

- [Google Colab の Spark にて Apache Iceberg の CoW と MoR の動作検証 #iceberg  Qiita](https://qiita.com/manabian/items/b78b1361b53411458a94)


## 事前準備

### Google Colab にて SparkSession の定義と Namespace の作成

```python
# Azure Storage に関る情報をセット
azure_storage_account_name = "snowflakeicebergqiita"
azure_storage_account_key = "kKRn6nruDHEhUF0AQzM1PFlXtXs9V0BJQSrn6Z8GvJOsb0JHflV9Fn=="

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("iceberg_lab")
    .config(
        "spark.jars.packages",
        'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,'
        'org.apache.iceberg:iceberg-azure-bundle:1.5.2,'
        'org.apache.hadoop:hadoop-azure:3.4.1'
    )
    .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", f"wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone")
    .config(f'spark.hadoop.fs.azure.account.key.{azure_storage_account_name}.blob.core.windows.net', azure_storage_account_key)
    .getOrCreate()
)
spark
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/96a1bb7f-6e34-4f16-94a9-29fac2df4390.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/de778a3a-478f-4f5b-be06-7f98045675d0.png)

### Databricks にてデータベースを作成

```sql
%sql
CREATE CATALOG iceberg_clone_test;
```

### Databricks にて

```python
# Azure Storage に関る情報をセット
azure_storage_account_name = "snowflakeicebergqiita"
azure_storage_account_key = "kKRn6nruDHEhUF0AQzM1PFlXtXs9V0BJQSrn6Z8GvJOsb0JHflV9Fn=="
```

```python
spark.conf.set(f"fs.azure.account.key.{azure_storage_account_name}.blob.core.windows.net", azure_storage_account_key)
spark.conf.set(f"fs.azure.account.key.{azure_storage_account_name}.dfs.core.windows.net", azure_storage_account_key)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/166c12e6-a0ef-46f9-ab0f-c131a393bd99.png)

本ステップにて Blob Storage エンドポイントと dfs エンドポイントに対する認証設定を実施している背景については下記の記事で紹介しています。

- [Databricks にて Azure Storage 上の Apache Iceberg を CLONE する際の Invalid configuration value detected エラーへの暫定対応方法 #iceberg - Qiita](https://qiita.com/manabian/items/0101664ab2b33cde59aa)

## 検証コードと結果

#### Google Colab にてテーブルを作成
```python
spark.sql(f"""
CREATE OR REPLACE TABLE default.first_table(
  id    INT,
  name  STRING,
  status STRING
)
USING iceberg;
""")
```

```python
spark.sql(f"SHOW TABLES;").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b20a26eb-aa30-4286-9727-1a3c7bb22dd7.png)

#### Databricks にて CLONE の実施

```python
retund_df = spark.sql(f"""
CREATE OR REPLACE TABLE iceberg_clone_test.default.test_table_01
  CLONE iceberg.`wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone/default/first_table`;
""")
retund_df
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9fa1cfaa-826c-4d40-a618-ad56f9cbe902.png)

#### CLONE の結果確認
```sql
%sql
DESC iceberg_clone_test.default.test_table_01;
```
```sql
%sql
DESC HISTORY iceberg_clone_test.default.test_table_01;
```
```sql
%sql
SELECT * FROM iceberg_clone_test.default.test_table_01 ORDER BY id;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/befe656b-db1d-4c74-9e79-7a73dd5f134c.png)


### INSERT 後の CLONE
#### Google Colab にてデータを INSERT
```python
spark.sql("""
INSERT INTO
    default.first_table (id, name, status)
VALUES
    (1, 'Alice', 'active'),
    (2, 'Bob', 'active'),
    (3, 'Charlie', 'active'),
    (4, 'Daisy', 'active'),
    (5, 'Ethan', 'active');
""").show()
```
```python
spark.sql("""
SELECT * FROM default.first_table ORDER BY id;
""").show()
```

#### Databricks にて CLONE
```python
retund_df = spark.sql(f"""
CREATE OR REPLACE TABLE iceberg_clone_test.default.test_table_01
  CLONE iceberg.`wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone/default/first_table`;
""")
retund_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/db3c492e-565e-420b-9f8f-e6ebe36dd7a2.png)

##### CLONE の結果確認
```sql
%sql
DESC HISTORY iceberg_clone_test.default.test_table_01;
```
```sql
%sql
SELECT * FROM iceberg_clone_test.default.test_table_01 ORDER BY id;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5071d5c7-e54d-42e2-a41f-d71d37eba693.png)


### UPDATE 後の CLONE
#### Google Colab にて UPDATE
```python
spark.sql("""
UPDATE default.first_table
    SET status = 'inactive' WHERE id IN (2, 3);
""").show()
```
```python
spark.sql("""
SELECT * FROM default.first_table ORDER BY id;
""").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e024d6a8-aac3-4a35-a63c-512ad5d3d098.png)

#### Databricks にて CLONE
```python
retund_df = spark.sql(f"""
CREATE OR REPLACE TABLE iceberg_clone_test.default.test_table_01
  CLONE iceberg.`wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone/default/first_table`;
""")
retund_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/59b75c9c-a688-420d-8210-b179f76ec720.png)

##### CLONE の結果確認
```sql
%sql
DESC HISTORY iceberg_clone_test.default.test_table_01;
```
```sql
%sql
SELECT * FROM iceberg_clone_test.default.test_table_01 ORDER BY id;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dcafaa2a-f175-4cdc-a5c3-9bae8c1b9324.png)


### DELETE 後の CLONE
#### Google Colab にて DELETE
```python
spark.sql("""
DELETE FROM default.first_table
    WHERE id = 5;
""").show()
```

```python
spark.sql("""
SELECT * FROM default.first_table ORDER BY id;
""").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/49648f53-cadc-4ed9-b91e-c9099247c4ab.png)

#### Databricks にて CLONE
```python
retund_df = spark.sql(f"""
CREATE OR REPLACE TABLE iceberg_clone_test.default.test_table_01
  CLONE iceberg.`wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone/default/first_table`;
""")
retund_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ed64499b-165f-4ef0-a0d0-c1d392d1ba45.png)

##### CLONE の結果確認
```sql
%sql
DESC HISTORY iceberg_clone_test.default.test_table_01;
```
```sql
%sql
SELECT * FROM iceberg_clone_test.default.test_table_01 ORDER BY id;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cd4da871-db50-4ba3-b7c0-4e8ce149ecbb.png)


### MERGE 後の CLONE
#### Google Colab にて MERGE
```python
spark.sql("""
MERGE INTO default.first_table AS tgt
USING (
    SELECT 1 AS id, 'Alice_updated' AS name, 'inactive' AS status
    UNION ALL
    SELECT 6 AS id, 'Frank' AS name, 'active' AS status
) AS src
ON tgt.id = src.id
WHEN MATCHED THEN
    UPDATE SET tgt.name = src.name, tgt.status = src.status
WHEN NOT MATCHED THEN
    INSERT (id, name, status)
    VALUES (src.id, src.name, src.status);
""").show()
```

```python
spark.sql("""
SELECT * FROM default.first_table ORDER BY id;
""").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f69a5995-aa83-45e4-b039-e37c80af6cf7.png)

#### Databricks にて CLONE
```python
retund_df = spark.sql(f"""
CREATE OR REPLACE TABLE iceberg_clone_test.default.test_table_01
  CLONE iceberg.`wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone/default/first_table`;
""")
retund_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4da25565-47cb-42d2-a174-404e490a349b.png)

##### CLONE の結果確認
```sql
%sql
DESC iceberg_clone_test.default.test_table_01;
```
```sql
%sql
DESC HISTORY iceberg_clone_test.default.test_table_01;
```
```sql
%sql
SELECT * FROM iceberg_clone_test.default.test_table_01 ORDER BY id;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d125ac1c-8792-4b5c-9646-2f9ce498e8fe.png)


### TRUNCATE TABLE 後の CLONE
#### Google Colab にて TRUNCATE TABLE
```python
spark.sql("""
TRUNCATE TABLE default.first_table;
""").show()
```

```python
spark.sql("""
SELECT * FROM default.first_table ORDER BY id;
""").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8dbf0e97-e02b-4246-aa5a-f527c5459b2b.png)

#### Databricks にて CLONE
```python
retund_df = spark.sql(f"""
CREATE OR REPLACE TABLE iceberg_clone_test.default.test_table_01
  CLONE iceberg.`wasbs://external@{azure_storage_account_name}.blob.core.windows.net/iceberg_clone/default/first_table`;
""")
retund_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eb9812da-66f0-4165-bc74-8e3d4867e81a.png)

##### CLONE の結果確認
```sql
%sql
DESC HISTORY iceberg_clone_test.default.test_table_01;
```
```sql
%sql
SELECT * FROM iceberg_clone_test.default.test_table_01 ORDER BY id;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/caec27d3-5e4a-43de-87da-5f2a4b50a983.png)