Snowflake

Foundry を Snowflake に接続して、Snowflake と Foundry の間でデータを読み取り、同期します。

Snowflake を Foundry の Iceberg カタログに接続することもできます。これにより、Snowflake は Apache Iceberg REST カタログ統合を通じて、Foundry が管理する Iceberg テーブルを直接読み書きできます。セットアップ手順については、Foundry の Iceberg カタログを Snowflake に接続するを参照してください。

サポートされている機能

セットアップ

1. Data Connection アプリケーションを開き、画面右上の + 新規ソースを選択します。
2. 利用可能なコネクターの種類から Snowflake を選択します。
3. 追加の設定プロンプトに従い、以下のセクションの情報を使用してコネクターのセットアップを続けます。

Foundry でのコネクターのセットアップについて詳しくはこちら。

接続情報

* ウェアハウス情報は、Foundry データセットの同期では任意ですが、仮想テーブルの登録では必須です。

認証

Snowflake では、次の方法で認証できます。

すべての認証オプションで、指定したユーザーとロールが、対象のデータベースとスキーマに対する使用権限、および対象のテーブルに対する選択権限を持っていることを確認してください。

仮想テーブルを登録する場合は、ユーザーとそのロールにウェアハウスの使用権限も必要です。

ネットワーク

Foundry ワーカーで実行される接続では、Data Connection アプリケーションでソースをセットアップする際に、適切なエグレスポリシーを適用する必要があります。

許可リストに追加する Snowflake アカウントのホスト名とポート番号を特定するには、Snowflake コンソールで次のコマンドを実行できます。少なくとも SNOWFLAKE_DEPLOYMENT と STAGE のエントリーが Foundry にエグレスポリシーとして追加されていることを確認してください。

Copied!
1
2
3
4
SELECT t.VALUE:type::VARCHAR as type,
       t.VALUE:host::VARCHAR as host,
       t.VALUE:port as port
FROM TABLE(FLATTEN(input => PARSE_JSON(SYSTEM$ALLOWLIST()))) AS t;

許可リストに追加するホスト名とポート番号の特定について詳しくは、Snowflake の公式ドキュメント ↗を参照してください。

S3 でホストされる Snowflake インスタンス

Snowflake インスタンスが内部 S3 ステージのトラフィックを VPCE ↗ 経由でルーティングするように設定されている場合は、カスタム VPCE ドメインを使用しないように Snowflake JDBC ドライバーを手動で設定する必要があります。そうしないと、ドライバーはカスタム VPCE ドメイン（Foundry の VPC からはアクセスできません）にルーティングされ、<bucketname>.bucket.vpce-<vpceid>.s3.<region>.vpce.amazonaws.com 形式の URL への接続に失敗します。

この設定は、インスタンスの接続情報で JDBC 接続プロパティを追加し、キーを S3_STAGE_VPCE_DNS_NAME、値の項目を空欄（null に設定することに相当）にすることで、手動で行えます。 これにより、S3 ステージのトラフィックは AWS S3 Gateway Endpoint（<bucketname>.bucket.s3.<region>.vpce.amazonaws.com）経由でルーティングされます。このエンドポイントはプライベート接続を維持するため、トラフィックがパブリックインターネット経由でルーティングされることはありません。

詳しくは、PrivateLink エグレスのドキュメント ↗を参照してください。

Azure でホストされる Snowflake インスタンス

Foundry の Snowflake コネクターで使用される Snowflake JDBC ドライバーは、データを取得する際に、基盤となる「内部ステージ」のストレージバケットに直接接続を試みる場合があります。Azure でホストされる Snowflake では、Azure でホストされる Foundry エンロールメントが Azure サービスエンドポイント経由でトラフィックをルーティングするため、以下の手順に従って、Foundry から基盤となるステージバケットへのネットワーク接続を明示的に許可リストに追加する必要があります。

Snowflake ウェアハウスに関する必要な情報を収集する

Foundry へのネットワーク接続を確立するには、Azure でホストされる Snowflake ウェアハウスに関する次の情報が必要です。

• システム許可リストのドメインの完全なリスト。
• Snowflake から取得した Azure ストレージアカウント識別子。

システム許可リストのドメイン

SYSTEM$ALLOWLIST コマンドを使用して、接続を成功させるために必要となる可能性のあるドメインの完全なリストを取得します。

• 注：これは、上記でエグレスポリシーを定義するために使用したものと同じコマンドであり、Data Connection のネットワークパネルのコールアウトで説明されています。
• このリストには、Snowflake ウェアハウスのステージとして使用される Azure ストレージバケットのドメインが含まれます。

Azure ストレージアカウント識別子

SYSTEM$ALLOWLIST コマンドで返された Azure ストレージバケットについては、ストレージアカウント識別子も取得する必要があります。

• Snowflake Standard Edition または Enterprise Edition を使用している場合は、Snowflake サポートにチケットを提出し、ストレージアカウント識別子を要求する必要があります。
• Snowflake Business Critical Edition ↗ を使用している場合は、次の手順でストレージアカウント識別子を取得できます。
  • アカウントの ENABLE_INTERNAL_STAGES_PRIVATELINK ↗ パラメーターを TRUE に設定します。
  • 次に、SYSTEM$GET_PRIVATELINK_CONFIG() ↗ 関数を呼び出します。この関数は、Azure ストレージアカウントのリソース識別子を含む privatelink-internal-stage というフィールドを返します。
    • PrivateLink 経由で接続しない場合でも、ストレージアカウントのリソース識別子を取得して指定する必要があります。

Azure Storage アカウントの完全なリソース識別子は、次の形式になります。

/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Storage/storageAccounts/{storageAccountName}

Azure コンソールで Azure Storage アカウントのリソース ID を直接確認する方法について詳しくは、Azure ドキュメント ↗を参照してください。VNET ルールとストレージアカウント識別子を使用してアカウント間のネットワークトラフィックを制限することは、Microsoft が公開しているベストプラクティス ↗に沿った方法です。Azure コンピューティングインスタンス内から、Azure でホストされる Snowflake ウェアハウスへのすべての接続で使用することを推奨します。

Foundry から Snowflake ウェアハウスに関連付けられた Azure ストレージアカウントへのアウトバウンドトラフィックを許可する

Snowflake ウェアハウスに関する必要な情報を収集したら、Foundry が Snowflake データにアクセスできるようにするために必要なポリシーを作成できます。

1. Azure ストレージの内部ステージ用に標準のエグレスポリシーを作成し、Snowflake ソースにアタッチします。

   • ストレージバケットのドメインだけでなく、SYSTEM$ALLOWLIST コマンドで返されたすべての項目に対してポリシーを追加する必要があります。

2. ストレージアカウントのリソース識別子を貼り付けて、Azure ストレージポリシーを作成します。

Data Connection の Snowflake ソースに戻り、ソースを探索して同期を実行できることを確認します。

Iceberg テーブル（仮想テーブルのみ）

このドキュメントの仮想テーブルセクションでは、Snowflake Horizon Catalog に登録された Iceberg テーブルの統合について詳しく説明しています。この機能には、Iceberg テーブルが保存されている外部ボリュームへのネットワーク接続が必要です。Foundry は Snowflake のコンピュートウェアハウスを使用してテーブルをクエリするのではなく、保存場所に直接読み書きするため、外部ボリュームごとにネットワークエグレスポリシーを作成する必要があります。

外部ボリュームおよびテーブルの保存場所を特定する方法の詳細については、Snowflake の公式ドキュメント ↗を参照してください。

仮想テーブル

このセクションでは、Snowflake ソースで仮想テーブルを使用する場合について詳しく説明します。このセクションは、Foundry データセットへの同期には適用されません。

次のテーブルは、Snowflake でサポートされている仮想テーブルの機能を示しています。

Snowflake テーブルを入力または出力として使用できる、サポート対象の Foundry ワークフローの詳細については、仮想テーブルのドキュメントを参照してください。

[1] インクリメンタルパイプラインは、テーブルのみでサポートされています。ビューはバージョン管理されないため、インクリメンタルパイプラインでは利用できません。Snowflake 仮想テーブルに基づくパイプラインでインクリメンタル処理を有効にするには、適切な保持期間で Change Tracking ↗ と Time Travel ↗ が有効になっていることを確認してください。この機能は CHANGES ↗ に依存します。Python トランスフォームの current および added 読み取りモードがサポートされています。これらのモードでは、METADATA$ACTION 列に基づいて、変更フィードの該当する行が公開されます。METADATA$ACTION、METADATA$ISUPDATE、METADATA$ROW_ID 列は Python トランスフォームで利用可能になります。

ソース資格情報の権限

すべての機能を利用するには、ソース接続に設定された資格情報に次の権限を付与してください。必要な継承モデルに応じて、データベース、スキーマ、またはテーブルのいずれかにこれらの権限を適用してください。

Iceberg テーブルを使用する場合は、テーブルが保存されている外部ボリュームに対する USAGE 権限が必要です。

また、指定する資格情報には、ソース設定で指定したウェアハウスに対する使用権限が必要です。

Snowflake のアクセス制御権限の詳細については、Snowflake の公式ドキュメント ↗を参照してください。

ソース設定の要件

仮想テーブルを使用する際は、次のソース設定の要件に留意してください。

• Foundry ワーカーソースを使用する必要があります。仮想テーブルは Agent ワーカー接続の使用に対応していません。
• このドキュメントのネットワークセクションで説明されているとおりに、双方向接続と許可リスト登録が設定されていることを確認してください。
• Code Repositories で仮想テーブルを使用する場合、必要な追加のソース設定の詳細については、仮想テーブルのドキュメントを参照してください。
• 接続情報にウェアハウスを指定する必要があります。
• 指定する資格情報には、ウェアハウスに対する使用権限が必要です。

詳細については、上記の接続情報セクションを参照してください。

Iceberg 同期

このソースから Foundry の Iceberg テーブルに直接データを同期できます。Iceberg 同期は、非インクリメンタルおよびインクリメンタル（差分連携）のバッチ取り込みと複数の書き込みモードをサポートし、Foundry ワーカーのランタイムで実行されます。設定の詳細とサポートされる書き込みモードについては、Iceberg 同期のドキュメントを参照してください。

Iceberg 同期は、仮想テーブル機能とは異なります。同期では Snowflake ソースのデータを Foundry が管理する Iceberg テーブルに書き込みます。一方、仮想テーブルでは、Snowflake Horizon Catalog に保存されている Iceberg テーブルを登録し、データをコピーせずに Foundry で使用します。

計算のプッシュダウン

Foundry では、仮想テーブルを使用する場合に計算を Snowflake にプッシュダウンできます。仮想テーブルの入力には、述語プッシュダウンを標準でサポートする Snowflake Spark コネクター ↗を使用します。

同じソースに登録された Snowflake 仮想テーブルをパイプラインの入力と出力として使用すると、計算をすべて Snowflake に委譲できます。計算を Snowflake にプッシュダウンする方法の詳細については、Python のドキュメントを参照してください。Pipeline Builder で計算を Snowflake にプッシュダウンする方法については、外部パイプラインのドキュメントを参照してください。

コードで Snowflake ソースを使用する

より複雑なシナリオでは、プロコードの代替アプリケーションを使用して Snowflake ソースに接続できます。

以下の例では、外部トランスフォームで Snowflake Connector for Python ↗（snowflake-connector-python）を使用して Snowflake ソースに接続する方法を示します。認証には OIDCを使用します。これにより、固定の Snowflake 資格情報を必要とせずに、有効期間の短い OAuth トークンが提供されます。OIDC が有効なソースでセッション資格情報を使用する方法の詳細については、Python でのソースの使用のドキュメントを参照してください。

外部トランスフォームで Snowflake から読み取る

この例では、OIDC ベースの OAuth 資格情報を使用して Snowflake テーブルからデータを読み取ります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import polars as pl
from transforms.api import transform, Output, LightweightOutput
from transforms.external.systems import ResolvedSource, external_systems, Source
import snowflake.connector


@external_systems(
    snowflake_source=Source("<source_rid>")
)
@transform.using(
    output=Output("<output_dataset_rid>")
)
def read_from_snowflake(output: LightweightOutput, snowflake_source: ResolvedSource):
    conn = snowflake.connector.connect(
        authenticator="oauth",
        token=snowflake_source.get_session_credentials().get().access_token,
        account="<account_identifier>",
        warehouse="<warehouse_name>",
        database="<database_name>",
    )

    cursor = conn.cursor()
    try:
        cursor.execute("SELECT * FROM my_table LIMIT 100")
        columns = [col[0] for col in cursor.description]
        rows = cursor.fetchall()
    finally:
        cursor.close()
        conn.close()

    df = pl.DataFrame(rows, schema=columns)
    output.write_table(df)

外部トランスフォームで Snowflake に書き込む

この例では、Foundry データセットから Snowflake テーブルにデータをエクスポートします。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import polars as pl
from transforms.api import transform, Input, Output, LightweightOutput
from transforms.external.systems import ResolvedSource, external_systems, Source
import snowflake.connector


@external_systems(
    snowflake_source=Source("<source_rid>")
)
@transform.using(
    output=Output("<output_dataset_rid>"),
    source_data=Input("<input_dataset_rid>"),
)
def write_to_snowflake(output: LightweightOutput, snowflake_source: ResolvedSource, source_data):
    conn = snowflake.connector.connect(
        authenticator="oauth",
        token=snowflake_source.get_session_credentials().get().access_token,
        account="<account_identifier>",
        warehouse="<warehouse_name>",
        database="<database_name>",
    )

    cursor = conn.cursor()
    try:
        df = source_data.dataframe()
        for row in df.iter_rows(named=True):
            cursor.execute(
                "INSERT INTO my_table (col1, col2) VALUES (%s, %s)",
                (row["col1"], row["col2"]),
            )
    finally:
        cursor.close()
        conn.close()

    output.write_table(df)

データモデル

データ型が array ↗、object ↗、variant ↗ の列は、Foundry によって string 型としてパースされます。これは、ソースのデータ型が可変であるためです。

たとえば、Snowflake の配列 [ 1, 2, 3 ] は、Foundry では文字列 "[1,2,3]" として解釈されます。

詳細については、Snowflake の公式ドキュメント ↗を参照してください。