Snowflake から 各クラウドの Delta Lake データを読み取る

## はじめに

Snowflake には Delta Direct と呼ばれる、Delta Lake テーブルを Iceberg テーブルとして読み取る機能が存在します。

参考：https://medium.com/snowflake/integrating-delta-tables-with-snowflake-performance-beyond-external-tables-52b54d31251c

当記事では相互運用性の調査で検証した内容を記録します。

## 基本的な方式

https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table-delta

を参考に検証していきます。

ポイントは以下です。

1. 各クラウドのデータレイクを外部ボリュームとして構成
1. Delta Lake 用のオブジェクトストレージ内のファイルのカタログ統合を作成
1. Delta Lake を基にした Iceberg テーブルを作成


## OneLake 上の Delta Lake を Snowflake on AWS から読み取る

### セットアップ

1. Fabric リージョンを確認

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/b6c1628c-8b04-4803-9db6-616266481d5e.png)


1. レイクハウスにサンプルデータ読み込み
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/fcf253e2-0723-4dd8-9a93-9393b43f3746.png)

1. Files 領域のパスでテーブルファイルにアクセスできるようにショートカット
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/390004e0-92d0-43bf-a9bb-79fa230306f4.png)
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/37af1123-f007-433f-b803-7076af174dec.png)

    :::note info
    
    Snowflake の OneLake に対する外部ボリュームは Files 領域のみが対象となるため
    Tables を対象にすると Iceberg テーブル作成時にコケます
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/e02ededd-8253-4991-9ed8-a607d95c8dc2.png)

1. AWS Snowflake であること確認
    ```sql:sql
    SELECT CURRENT_REGION();
    ```
    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/5443f846-3bc5-4f4b-8b50-605b421579d1.png)


1.  OneLake を対象にした外部ボリュームを作成

    ```sql:sql
    
    CREATE EXTERNAL VOLUME ext_vol_onelake_sand
      STORAGE_LOCATIONS =
        (
          (
            NAME = 'ext_vol_onelake_sand_lh_opentableformat_for_volume'
            STORAGE_PROVIDER = 'AZURE'
            STORAGE_BASE_URL = 'azure://onelake.dfs.fabric.microsoft.com/<ワークスペース名>/<レイクハウス名>.Lakehouse/Files/<パス>'
            AZURE_TENANT_ID = '<テナントID>'
          )
        );
    
    ```
    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/3f2fcac9-9018-48ab-ba12-a7d6cde5c21e.png)

1. 権限付与と疎通確認
   https://qiita.com/ryoma-nagata/items/44c165c8f44f73e0b3b4 などを参考に MS Entra ID サービスプリンシパルを対象の Fabic テナントに追加し、権限を付与します。

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/92bb1f3a-66d5-40e7-9b2e-699d9c0754ab.png)


    viewer を利用する場合は、ファイルへのアクセス権をもたないので、対象のレイクハウスの ReadAll などの追加のファイルアクセス権限を付与します
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/9439df19-51fa-4bbc-a191-a5d609381241.png)

    
    疎通を確認

    ```sql:sql
    
    SELECT SYSTEM$VERIFY_EXTERNAL_VOLUME('ext_vol_onelake_sand');
    
    ```

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/5e363ac7-052a-425e-a1fc-5ba9de4feb54.png)


    
1. Delta Lake 用のオブジェクトストレージ内のファイルのカタログ統合を作成

    ```sql:sql
    
    CREATE OR REPLACE CATALOG INTEGRATION delta_catalog_integration
      CATALOG_SOURCE = OBJECT_STORE
      TABLE_FORMAT = DELTA
      ENABLED = TRUE;
    
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c0bce069-78ae-48f8-80a4-8661599751cf.png)

1. Iceberg テーブル作成と定義確認
    
    ```sql:sql
    CREATE OR REPLACE ICEBERG TABLE public.iceberg_onelake_fact_sale
      CATALOG = delta_catalog_integration
      EXTERNAL_VOLUME = ext_vol_onelake_sand
      AUTO_REFRESH = TRUE
      BASE_LOCATION = 'fact_sale/';
    
    ```
    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/6eb73913-2387-4d7d-98a3-a36bc3397070.png)


### 変更反映の確認

1. 変更前の件数を確認

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/6bce730d-df5f-40b0-8d77-bf8ea0e30edb.png)

1. Fabric 側で 1000 件 append

    ```python:pyspark

    df = spark.sql("SELECT * FROM LH_opentableformat.fact_sale LIMIT 1000")
    
    df.write.mode("append").saveAsTable("fact_sale")
    
    result = spark.sql("SELECT count(1) FROM LH_opentableformat.fact_sale ")
    
    display(result)
    ```

    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/24d082e0-98dc-4005-ac9a-58e1272f4543.png)

1. snowflake 側で変更を確認
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/9f561607-74d3-4113-b5b2-52cd928f2bfe.png)


## S3 上の Delta Lake を Snowflake on Azure から読み取る

### セットアップ

1. Fabric notebook で書き込みします。元のデータは前述のサンプルを使いまわします。

    ```python:pyspark
    
    df = spark.sql("SELECT * FROM LH_opentableformat.fact_sale LIMIT 1000")
    
    s3_backet_name ="localdevdemo"
    access_key_id=KEYID
    accecc_key_secret=SECRET
    
    sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", access_key_id)
    sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", accecc_key_secret)
    
    s3_route_path=f's3a://{s3_backet_name}'
    output_path = f"{s3_route_path}/for_volume/fact_sale"
    
    df.write.mode("overwrite").format("delta").save(output_path)


    
    
    
    ```
    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/8fd8e5c6-0a0c-4fc2-bd51-e5748513e210.png)

1. IAM　ロールの準備

    https://docs.snowflake.com/ja/user-guide/tables-iceberg-configure-external-volume-s3 を参考にして IAM などを設定します.Putなどは今回は不要ではあると思います。

    ポリシー例：

    ```json:json
    
    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "VisualEditor0",
                "Effect": "Allow",
                "Action": [
                    "s3:PutObject",
                    "s3:GetObject",
                    "s3:GetObjectVersion",
                    "s3:DeleteObject",
                    "s3:DeleteObjectVersion",
                    "s3:ListBucket",
                    "s3:GetBucketLocation"
                ],
                "Resource": "*"
            }
        ]
    
    ```
1. Azure Snowflake であることを確認

    ```sql:sql
    SELECT CURRENT_REGION();
    ```

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/e151e574-c679-4d47-9afd-9fbe68dd8192.png)


1. 外部ボリュームを作成し、Snowflake の ID を取得します。
    ```sql:sql

    CREATE OR REPLACE EXTERNAL VOLUME ext_vol_aws_localdevdemo
       STORAGE_LOCATIONS =
          (
             (
                NAME = 'localdevdemo_for_volume'
                STORAGE_PROVIDER = 'S3'
                STORAGE_BASE_URL = 's3://<bucket名>/<パス>/'
                STORAGE_AWS_ROLE_ARN = '<ロールのARN>'
                STORAGE_AWS_EXTERNAL_ID = '<ロール作成時に指定した ID 文字列>'
             )
          );
    ```

    ![uploading...0]()


1. IAM ロールを編集
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/64eb1eb4-c05f-4e10-9f3e-08cd6d731542.png)

1. 疎通確認
    
    ```sql:sql
    
    SELECT SYSTEM$VERIFY_EXTERNAL_VOLUME('ext_vol_aws_localdevdemo');
    
    ```

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c187604f-3fba-48ca-b2e1-7b4b3c806bb4.png)

    
1. Delta Lake 用のオブジェクトストレージ内のファイルのカタログ統合を作成

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

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c0bce069-78ae-48f8-80a4-8661599751cf.png)


1. Iceberg テーブル作成と定義確認
    
    ```sql:sql
    CREATE OR REPLACE ICEBERG TABLE public.iceberg_s3_fact_sale
      CATALOG = delta_catalog_integration
      EXTERNAL_VOLUME = ext_vol_aws_localdevdemo
      AUTO_REFRESH = TRUE
      BASE_LOCATION = 'fact_sale/';
    
    ```

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/37ee4792-6d99-4052-97bc-885cf7f1ef32.png)


### 変更反映の確認


1. Fabric 側で 1000 件 append

    ```python:pyspark

    df.write.mode("append").format("delta").save(output_path)

    result_df = spark.read.format("delta").load(output_path)
    result_df.count()
    ```

    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/d645302b-5aac-4119-8c82-060ce0efa01b.png)



1. snowflake 側で変更を確認
    
    
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/1e490b54-b97c-4308-9094-e8b5dfd99d13.png)