Snowflake Apache Iceberg™ table を Delta Lake 形式のディレクトリから作成する機能を試してみた

## 概要

Snowflakeで、Apache Iceberg™テーブルをDelta Lake形式のディレクトリから作成できる機能を検証したため、その結果と検証コードを共有します。ここで注意すべき点は、Apache Iceberg形式のディレクトリ構成に変更されるわけではなく、Snowflakeが内部でスキームの解釈をおこなっているようです。本記事では、 Google Colab の Spark から Delta Lake 形式のデータを用意しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9e6da5aa-d149-415d-8bba-43b0f031bbd5.png)

出所：[Snowflakeで Apache Iceberg™ テーブルを作成する | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/tables-iceberg-create#delta-files-in-object-storage)

Delta LakeはAuto Refreshに対応しています。2025年3月9日時点では、日本語ドキュメントにこの点の記載はありませんが、英語ドキュメントにはDelta Lakeがサポート対象であると明記されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5d5d2726-45c9-442d-9fba-078f72906345.png)

出所：[Apache Iceberg™ テーブルの自動リフレッシュ | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/tables-iceberg-auto-refresh)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/711d63a0-4ea8-4152-9bdf-5457877843f7.png)

出所：[Automatically refresh Apache Iceberg™ tables | Snowflake Documentation](https://docs.snowflake.com/en/user-guide/tables-iceberg-auto-refresh)
本記事は「SnowflakeにおけるApache Iceberg機能の完全ガイド」シリーズの一部です。  
Apache Iceberg機能の全体像やSnowflake上での活用方法について詳しく知りたい方は、以下の記事をご覧ください。

- [SnowflakeにおけるApache Icebergの機能完全ガイド：Apache Iceberg on Snowflake 活用の第一歩 #Spark - Qiita](https://qiita.com/manabian/items/7ba563b614e8ae016cca)

## 環境準備

### Snowflake Catalog環境の準備

以下の記事で紹介している環境構築手順をあらかじめ実施してください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5a727271-1090-4694-bf8c-b2a493fd0e56.png)

出所：[Snowflake Catalog における Iceberg テーブルの基本的な操作手順 #Spark - Qiita](https://qiita.com/manabian/items/69761bde791d0f014246)

## Delta LakeからIcebergテーブルを作成

### Google Colab（Spark）でDelta Lake形式のデータを準備

まず、Azureストレージのアカウント名とキーを変数にセットします。

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/703e4082-e789-4d33-b7df-5c0cff5919cf.png)

SparkSessionを定義します。Delta Lakeを利用できるようにパッケージを追加しています。

```python
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
        .appName("DeltaLakeExample")
        .config("spark.jars.packages", "")
        .config(
            "spark.jars.packages",
            "io.delta:delta-spark_2.12:3.2.1,"
            "org.apache.hadoop:hadoop-azure:3.2.0"
        )
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .config("spark.sql.warehouse.dir", "/delta")
        .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/01a07587-f401-4add-9f85-9a4c812f87ef.png)

次に、Delta Lakeの外部テーブルをAzure Storage上に作成します。

```python
spark.sql(f"""
CREATE OR REPLACE TABLE first_table (
  id INT,
  name STRING
)
USING delta
LOCATION 'wasbs://snowflake@{azure_storage_account_name}.blob.core.windows.net/deltalake/first_table'
""")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/15474d99-6d3b-419d-b6f4-05536f9d43a5.png)

続いてテーブルにデータを上書き書き込みします。

```python
schema = """
id INT,
name STRING
"""
data = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]

df = spark.createDataFrame(data, schema)
display(df.toPandas())

# Delta テーブルに上書き書き込み
df.write.mode("overwrite").format("delta").saveAsTable("first_table")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bccb1611-80d4-4e21-858c-6785fe0a2554.png)

### Snowflakeでオブジェクトを作成

データベースを作成します。

```sql
CREATE OR REPLACE DATABASE delta_to_iceberg_db;
USE DATABASE delta_to_iceberg_db;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4838d204-c21c-4eee-bc99-7ed52f3bad88.png)

オブジェクトストレージ（Delta Lake形式）のカタログ統合を作成します。

```sql
CREATE OR REPLACE CATALOG INTEGRATION delta_catalog_integration
  CATALOG_SOURCE = OBJECT_STORE
  TABLE_FORMAT = DELTA
  ENABLED = TRUE;
```

```sql
DESCRIBE CATALOG INTEGRATION delta_catalog_integration;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e48d398c-57be-4956-b816-a9e7299427ed.png)

続いて、外部Volumeを作成し、その認証設定を行います。

```sql
-- 各値を変数としてセット
SET azure_tenant_id = '095ca098-c723-478f-xxxx-xxxxx';
SET azure_storage_account_name = 'snowflakeicebergqiita';
SET storage_provider = 'AZURE';

-- 連結した URL は変数に事前に設定
SET full_storage_uri = 'azure://' || $azure_storage_account_name || '.blob.core.windows.net/snowflake/deltalake/';

SELECT
    $storage_provider AS  STORAGE_PROVIDER
    ,$full_storage_uri AS STORAGE_BASE_URL
    ,$azure_tenant_id AS AZURE_TENANT_ID;
```

```sql
-- 外部 Volume を作成
CREATE OR REPLACE EXTERNAL VOLUME delta_to_iceberg_volume
    STORAGE_LOCATIONS = (
        (
          NAME = 'delta-to-iceberg-volume'
          STORAGE_PROVIDER = 'AZURE'
          STORAGE_BASE_URL = 'azure://snowflakeicebergqiita.blob.core.windows.net/snowflake/deltalake/'
          AZURE_TENANT_ID = '095ca098-c723-478f-xxxx-xxxxx'
        )
    )
    ALLOW_WRITES = true;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d34a3e8e-b4fb-40f8-8a28-ef83a45da2b7.png)

```sql
DESC EXTERNAL VOLUME delta_to_iceberg_volume;
```

```sql
DESC EXTERNAL VOLUME delta_to_iceberg_volume;
SELECT 
    PARSE_JSON($4):AZURE_MULTI_TENANT_APP_NAME::string AS AZURE_MULTI_TENANT_APP_NAME,
    PARSE_JSON($4):AZURE_CONSENT_URL::string AS AZURE_CONSENT_URL
FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()))
WHERE $2 = 'STORAGE_LOCATION_1';
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3b676edd-4797-4ce4-a1b3-1029fb6add84.png)

### 外部Volumeへのアクセス権付与

`AZURE_CONSENT_URL`のURLをAzureにログイン中のブラウザで開き、Snowflakeからのアクセス許可を与えます。

```sql
DESC EXTERNAL VOLUME my_azure_sf_volume;
SELECT 
    PARSE_JSON($4):AZURE_MULTI_TENANT_APP_NAME::string AS AZURE_MULTI_TENANT_APP_NAME,
    PARSE_JSON($4):AZURE_CONSENT_URL::string AS AZURE_CONSENT_URL
FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()))
WHERE $2 = 'STORAGE_LOCATION_1';
```

`承諾`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0ac04b60-8a5b-47c8-b66c-a3a24be08125.png)

Azure Storage上で`AZURE_MULTI_TENANT_APP_NAME`のアンダーバーより前の文字列（例：`ps27tlsnowflakepacint`）を検索し、適切なロールを付与します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d31403fe-9c9d-4067-81cd-887d17b1dacb.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f6f351c-726f-43e9-a4e7-0a34b31dcc2a.png)

なお、本記事ではAzure向けの手順を示していますが、Snowflakeのドキュメントにはクラウドベンダーごとに手順が用意されています。

- [Amazon S3用の外部ボリュームを構成する | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/tables-iceberg-configure-external-volume-s3)  
- [Google Cloud Storageの外部ボリュームを構成する | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/tables-iceberg-configure-external-volume-gcs)  
- [Azureの外部ボリュームを構成する | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/tables-iceberg-configure-external-volume-azure)  

### テーブルの作成

Delta Lake形式のオブジェクトストレージからIcebergテーブルを作成します。

```sql
CREATE OR REPLACE ICEBERG TABLE first_table
  CATALOG = delta_catalog_integration
  EXTERNAL_VOLUME = delta_to_iceberg_volume
  BASE_LOCATION = 'first_table/'
  AUTO_REFRESH = FALSE;
```

```sql
SHOW TABLES;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/59f30f87-43c1-4b48-b7a7-2cabcd17a52c.png)

クエリを発行してみます。

```sql
SELECT
    *
    FROM
        first_table
    ORDER BY
        ID;
```

一方、Icebergテーブルの最新metadata fileを参照するシステム関数を呼び出すとエラーが発生します。

```sql
SELECT SYSTEM$GET_ICEBERG_TABLE_INFORMATION('first_table');
```

> Failed to generate Iceberg metadata. Reason: Unknown error producing metadata.json file

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c2c33278-6524-416b-83eb-05043d2e340d.png)

## 手動リフレッシュの実施

#### Google Colab（Spark）でデータを追加

Google Colab（Spark）を使い、`id`列が`4`のデータをDelta Lakeテーブルに追加します。

```python
schema = """
id INT,
name STRING
"""
data = [(4, "David")]

df = spark.createDataFrame(data, schema)
display(df.toPandas())

df.write.mode("append").format("delta").saveAsTable("first_table")
```

```python
display(spark.table("first_table").orderBy("id").toPandas())
```

```python
display(spark.sql("DESC HISTORY first_table").toPandas())
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7c3d80ab-a23a-4c43-8e23-c20a54fd3e98.png)

#### Snowflakeで手動リフレッシュの実施

Snowflakeでクエリを実行すると、まだ`id`列が`4`のデータは取得できません。

```sql
SELECT
    *
    FROM
        first_table
    ORDER BY
        ID;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/623d2fd6-4cf0-4fd1-91cd-0c4990facf3b.png)

手動リフレッシュを実行すると、テーブルが最新化され、`id`列が`4`のデータも取得できます。

```sql
ALTER ICEBERG TABLE first_table REFRESH;
```

```sql
SELECT
    *
    FROM
        first_table
    ORDER BY
        ID;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6191938c-be20-40ea-8774-f7139366e1b2.png)

## 自動リフレッシュの設定

### Snowflakeでテーブルの`AUTO_REFRESH`を`TRUE`に設定

```sql
ALTER ICEBERG TABLE first_table SET
  AUTO_REFRESH = TRUE;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c7893ed7-b761-4101-8e0b-1dede7efc902.png)

### Google Colab（Spark）でデータを追加

Google Colab（Spark）にて、`id`列が`5`のデータをDelta Lakeテーブルに追加します。

```python
schema = """
id INT,
name STRING
"""
data = [(5, "Edwin")]

df = spark.createDataFrame(data, schema)
display(df.toPandas())

df.write.mode("append").format("delta").saveAsTable("first_table")
```

```python
display(spark.table("first_table").orderBy("id").toPandas())
```

```python
display(spark.sql("DESC HISTORY first_table").toPandas())
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/45ba4dea-41ef-446d-a24c-5be4112bef15.png)

### テーブルが自動で最新化されることを確認

手動リフレッシュをしなくても`id`列が`5`のデータを取得でき、テーブルが自動で最新化されていることが確認できます。

```sql
SELECT
    *
    FROM
        first_table
    ORDER BY
        ID;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eb155015-adf5-4d89-acdd-cb6277e2a7f7.png)