Google Cloud Storage用Snowpipeの自動化¶

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

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

単一のストレージ統合により、複数の外部（つまり、 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 は、データファイルを保存するCloud Storageバケットの名前です（例: 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 Serviceアカウントを取得する¶

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アカウント用に作成されたCloud Storageサービスアカウントを示しています（例： service-account-id@project1-123456.iam.gserviceaccount.com）。Snowflakeアカウント全体に、単一のCloud Storageサービスアカウントをプロビジョニングします。すべてのCloud Storage統合で、そのサービスアカウントを使用します。

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

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

前提条件¶

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

GCP アカウント:

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

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

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

Snowflake:

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

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

ステップ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 <テーブル> ステートメントを定義します。

CREATE PIPE <pipe_name>
  AUTO_INGEST = true
  INTEGRATION = '<notification_integration_name>'
  AS
<copy_statement>;

条件:

<パイプ名>

パイプの識別子。パイプが作成されるスキーマに対して一意である必要があります。

識別子はアルファベットで始まる必要があり、識別子文字列全体が二重引用符で囲まれていない限り、スペースや特殊文字を含めることはできません（例： "My object"）。二重引用符で囲まれた識別子も大文字と小文字が区別されます。

INTEGRATION = '<通知統合名>'

Google Cloud Pub/Sub通知を使用して、ディレクトリテーブルのメタデータを自動的に更新するために使用される通知統合の名前。通知統合は、Snowflakeとサードパーティのクラウドメッセージキューサービス間のインターフェイスを提供するSnowflakeオブジェクトです。

copy_statement

キューファイルからSnowflakeテーブルにデータをロードするために使用される COPY INTO <テーブル> ステートメント。このステートメントは、パイプのテキスト/定義として機能し、 SHOW PIPES 出力に表示されます。

たとえば、 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がデータをロードした後のステージングされたファイルの削除 をご参照ください。