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

はじめに

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

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

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

基本的な方式

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

ポイントは以下です。

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

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

セットアップ

1. Fabric リージョンを確認

   

2. レイクハウスにサンプルデータ読み込み
   

3. Files 領域のパスでテーブルファイルにアクセスできるようにショートカット
   
   

   Snowflake の OneLake に対する外部ボリュームは Files 領域のみが対象となるため
   Tables を対象にすると Iceberg テーブル作成時にコケます
   

4. AWS Snowflake であること確認

   sql

   SELECT CURRENT_REGION();
   

   

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

   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>'
         )
       );
   
   

   

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

   

   viewer を利用する場合は、ファイルへのアクセス権をもたないので、対象のレイクハウスの ReadAll などの追加のファイルアクセス権限を付与します
   

   疎通を確認

   sql

   
   SELECT SYSTEM$VERIFY_EXTERNAL_VOLUME('ext_vol_onelake_sand');
   
   

   

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

   sql

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

   

8. Iceberg テーブル作成と定義確認

   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/';
   
   

   

変更反映の確認

1. 変更前の件数を確認

   

2. Fabric 側で 1000 件 append

   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)
   

   

3. snowflake 側で変更を確認
   

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

セットアップ

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

   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)
   
   
   
   
   
   

   

2. IAM　ロールの準備

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

   ポリシー例：

   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": "*"
           }
       ]
   
   

3. Azure Snowflake であることを確認

   sql

   SELECT CURRENT_REGION();
   

   

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

   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 文字列>'
            )
         );
   

5. IAM ロールを編集
   

6. 疎通確認

   sql

   
   SELECT SYSTEM$VERIFY_EXTERNAL_VOLUME('ext_vol_aws_localdevdemo');
   
   

   

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

   sql

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

   

8. Iceberg テーブル作成と定義確認

   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/';
   
   

   

変更反映の確認

1. Fabric 側で 1000 件 append

   pyspark

   
   df.write.mode("append").format("delta").save(output_path)
   
   result_df = spark.read.format("delta").load(output_path)
   result_df.count()
   

   

2. snowflake 側で変更を確認