Google Cloud Storage用Snowpipeの自動化¶

ステップ1:Snowflakeでクラウドストレージ統合を作成する¶

CREATE STORAGE INTEGRATION コマンドを使用して統合を作成します。統合は、外部クラウドストレージの認証責任をSnowflakeが生成したエンティティ (つまり、クラウドストレージサービスアカウント) に委譲するSnowflakeオブジェクトです。Cloud Storageバケットにアクセスするために、Snowflakeは、データファイルを保存するバケットへのアクセス許可を付与できるサービスアカウントを作成します。

1つのストレージ統合で、複数の外部ステージ (つまり、 GCS) をサポートすることができます。ステージ定義にある URL は、 STORAGE_ALLOWED_LOCATIONS パラメーターのために指定された GCS バケット（およびオプションのパス）に合わせる必要があります。

CREATE STORAGE INTEGRATION <integration_name>
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://<bucket>/<path>/', 'gcs://<bucket>/<path>/')
  [ STORAGE_BLOCKED_LOCATIONS = ('gcs://<bucket>/<path>/', 'gcs://<bucket>/<path>/') ]

条件:

• integration_name は、新しい統合の名前です。

• bucket は、データファイルをストレージするクラウドストレージのバケット名です (例えば、 mybucket)。必須の STORAGE_ALLOWED_LOCATIONS パラメーターおよびオプションの STORAGE_BLOCKED_LOCATIONS パラメーターは、この統合を参照するステージが作成または変更されたときに、それぞれこれらのバケットへのアクセスを制限またはブロックします。

• path は、バケット内のオブジェクトを細かく制御するために使用できるオプションのパスです。

次の例では、2つのバケットとパスのいずれかを参照するために統合を使用する外部ステージを明示的に制限する統合を作成します。後のステップで、これらのバケットとパスのいずれかを参照する外部ステージを作成します。

この統合も使用する追加の外部ステージは、許可されたバケットとパスを参照できます。

CREATE STORAGE INTEGRATION gcs_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://mybucket1/path1/', 'gcs://mybucket2/path2/')
  STORAGE_BLOCKED_LOCATIONS = ('gcs://mybucket1/path1/sensitivedata/', 'gcs://mybucket2/path2/sensitivedata/');

Copy

ステップ2:Snowflakeアカウント用のCloud Storageサービスアカウントを取得する¶

DESCRIBE INTEGRATION コマンドを実行して、Snowflakeアカウント用に自動的に作成されたCloud Storageサービスアカウントの ID を取得します。

DESC STORAGE INTEGRATION <integration_name>;

条件:

• integration_name は、 ステップ1:Snowflakeでクラウドストレージ統合を作成する (このトピック内) で作成した統合の名前です。

例:

DESC STORAGE INTEGRATION gcs_int;

+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+
| property                    | property_type | property_value                                                              | property_default |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------|
| ENABLED                     | Boolean       | true                                                                        | false            |
| STORAGE_ALLOWED_LOCATIONS   | List          | gcs://mybucket1/path1/,gcs://mybucket2/path2/                               | []               |
| STORAGE_BLOCKED_LOCATIONS   | List          | gcs://mybucket1/path1/sensitivedata/,gcs://mybucket2/path2/sensitivedata/   | []               |
| STORAGE_GCP_SERVICE_ACCOUNT | String        | service-account-id@project1-123456.iam.gserviceaccount.com                  |                  |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+

Copy

出力の STORAGE_GCP_SERVICE_ACCOUNT プロパティには、Snowflake アカウントに作成されたクラウドストレージサービスアカウントが表示されます (つまり、 service-account-id@project1-123456.iam.gserviceaccount.com)。Snowflakeアカウント全体に、単一のCloud Storageサービスアカウントをプロビジョニングします。すべてのCloud Storage統合で、そのサービスアカウントを使用します。

ステップ3: バケットオブジェクトにアクセスするためのサービスアカウント権限を付与する¶

次の詳細な手順では、Cloud Storageバケットを使用してデータをロードおよびアンロードできるようにするために、 Google Cloud console でSnowflake用の IAM アクセス許可を構成する方法を説明します。

前提条件¶

このトピックの説明では、次のアイテムが作成および構成されていることを前提としています。

GCP アカウント:

• GCS バケットからイベントメッセージを受信するPub/Subトピック。詳細については、 Pub/Subトピックの作成 （このトピック内）をご参照ください。

• Pub/Subトピックからイベントメッセージを受信するサブスクリプション。詳細については、このトピックの Pub/Subサブスクリプションの作成 をご参照ください。

手順については、 Pub/Subのドキュメント をご参照ください。

Snowflake:

• データをロードするSnowflakeデータベース内のターゲットテーブル。

$ gsutil notification create -t <topic> -f json -e OBJECT_FINALIZE gs://<bucket-name>

ステップ1： Snowflakeで通知統合を作成する¶

CREATE NOTIFICATION INTEGRATION コマンドを使用して通知統合を作成します。

通知統合は、Pub/Subサブスクリプションを参照します。Snowflakeは、通知統合をアカウント用に作成された GCS サービスアカウントに関連付けます。Snowflakeは、Snowflakeアカウントのすべての GCS 通知統合によって参照される単一のサービスアカウントを作成します。

CREATE NOTIFICATION INTEGRATION <integration_name>
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = '<subscription_id>';

条件:

• integration_name は、新しい統合の名前です。

• subscription_id は、 Pub/Subサブスクリプションの取得 ID で記録したサブスクリプション名です。

例:

CREATE NOTIFICATION INTEGRATION my_notification_int
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = 'projects/project-1234/subscriptions/sub2';

ステップ2:Pub/SubサブスクリプションにSnowflakeアクセスを許可する¶

1. DESCRIBE INTEGRATION コマンドを実行して、Snowflakeサービスアカウント ID を取得します。

   DESC NOTIFICATION INTEGRATION <integration_name>;
   

   Copy

   条件:

   • integration_name は、 ステップ1:Snowflakeで通知統合を作成する で作成した統合の名前です。

   例:

   DESC NOTIFICATION INTEGRATION my_notification_int;
   

   Copy

2. GCP_PUBSUB_SERVICE_ACCOUNT 列にサービスアカウント名を記録します。これは次の形式です。

   <service_account>@<project_id>.iam.gserviceaccount.com
   

   Copy

3. プロジェクトエディターとしてGoogle Cloud Platform Consoleにログインします。

4. ホームダッシュボードから、 Big Data » Pub/Sub » Subscriptions を選択します。

5. アクセス用に構成するサブスクリプションを選択します。

6. 右上隅の SHOW INFO PANEL をクリックします。サブスクリプションの情報パネルがスライドアウトします。

7. ADD PRINCIPAL ボタンをクリックします。

8. New principals フィールドで、記録したサービスアカウント名を検索します。

9. Select a role ドロップダウンから、 Pub/Sub Subscriber を選択します。

10. Save ボタンをクリックします。サービスアカウント名は、情報パネルの Pub/Sub Subscriber ロールドロップダウンに追加されます。

11. Cloud Consoleの Dashboard ページに移動し、ドロップダウンリストからプロジェクトを選択します。

12. ADD PEOPLE TO THIS PROJECT ボタンをクリックします。

13. 記録したサービスアカウント名を追加します。

14. Select a role ドロップダウンから、 Monitoring Viewer を選択します。

15. Save ボタンをクリックします。サービスアカウント名は、 Monitoring Viewer ロールドロップダウンに追加されます。

ステップ3: ステージを作成する（必要な場合）¶

CREATE STAGE コマンドを使用して、 GCS バケットを参照する外部ステージを作成します。Snowflakeは、ステージングされたデータファイルを外部テーブルメタデータに読み取ります。または、既存の外部ステージを使用できます。

次の例では、ユーザーセッションのアクティブスキーマに mystage という名前のステージを作成します。クラウドストレージ URL にはパス files が含まれています。ステージは my_storage_int という名前のストレージ統合を参照します。

USE SCHEMA mydb.public;

CREATE STAGE mystage
  URL='gcs://load/files/'
  STORAGE_INTEGRATION = my_storage_int;

Copy

ステップ4: 自動インジェストを有効にしたパイプを作成する¶

CREATE PIPE コマンドを使用してパイプを作成します。パイプは、Snowpipeがインジェスションキューからターゲットテーブルにデータをロードするために使用する COPY INTO <テーブル> ステートメントを定義します。

たとえば、 snowpipe_db.public スキーマにパイプを作成し、 mystage という外部（GCS）ステージにステージングされたファイルから、 mytable という宛先テーブルにデータをロードします。

CREATE PIPE snowpipe_db.public.mypipe
  AUTO_INGEST = true
  INTEGRATION = 'MY_NOTIFICATION_INT'
  AS
    COPY INTO snowpipe_db.public.mytable
      FROM @snowpipe_db.public.mystage/path2;

INTEGRATION パラメーターは、 ステップ1:Snowflakeでクラウドストレージ統合を作成する で作成した my_notification_int 通知統合を参照します。統合名はすべて大文字で提供する必要があります。

自動インジェストを使用したSnowpipeが構成されました。

新しいデータファイルが GCS バケットに追加されると、イベントメッセージはSnowpipeに通知して、パイプで定義されたターゲットテーブルにそれらをロードします。

ステップ5: 履歴ファイルをロードする¶

Pub/Subメッセージが構成される 前 に外部ステージに存在したデータファイルのバックログをロードするには、 ALTER PIPE ... REFRESH ステートメントを実行します。

ステップ6: ステージングされたファイルを削除する¶

データが正常にロードされ、ファイルが不要になったら、ステージングされたファイルを削除します。手順については、 Snowpipeがデータをロードした後のステージングされたファイルの削除 をご参照ください。