Amazon S3用Snowpipeの自動化¶

ステップ1: S3バケットのアクセス許可を構成する¶

ステップ2： IAM AWS ロールを作成する¶

AWS 管理コンソールでSnowflakeのアクセス許可を設定するには、次の手順を実行します。

1. AWS 管理コンソールにログインします。

2. ホームダッシュボードから、 Identity & Access Management （IAM）を選択します。

   

3. 左側のナビゲーションペインから Roles を選択します。

4. Create role を選択します。

   

5. 信頼できるエンティティタイプとして Another AWS account を選択します。

6. Account ID フィールドに、自分の AWS アカウント ID を一時的に入力します。後で、信頼関係を変更し、Snowflakeへのアクセスを許可します。

7. Require external ID オプションを選択します。外部 ID は、Snowflakeなどのサードパーティに AWS リソース（S3バケットなど）へのアクセスを許可するために使用されます。

   0000 などのプレースホルダー ID を入力します。後のステップで、 IAM ロールの信頼関係を変更し、ストレージ統合の外部 ID を指定します。

8. Next を選択します。

9. ステップ1: S3バケットのアクセス許可を構成する （このトピック内）で作成したポリシーを選択します。

10. Next を選択します。

    

11. ロールの名前と説明を入力し、 Create role を選択します。

    これで、バケットの IAM ポリシー、および IAM ロールを作成し、ロールにポリシーを添付しました。

12. ロールの概要ページにある Role ARN 値を見つけて記録します。次のステップでは、このロールを参照するSnowflake統合を作成します。

    

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

CREATE STORAGE INTEGRATION コマンドを使用してストレージ統合を作成します。ストレージ統合は、S3クラウドストレージ用に生成されたIDおよびアクセス管理（IAM）ユーザーと、許可またはブロックされたストレージロケーション（バケットなど）のオプションセットを保存するSnowflakeオブジェクトです。組織のクラウドプロバイダー管理者は、生成されたユーザーにストレージの場所に対するアクセス許可を付与します。このオプションにより、ユーザーはステージの作成時またはデータのロード時に、認証情報の提供を回避できます。

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

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

条件:

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

• iam_role は、 ステップ2: AWS で IAM ロールを作成する （このトピック内）で作成したロールのAmazonリソースネーム（ARN）です。

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

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

次の例では、アカウント内のすべてのバケットへのアクセスを許可し、定義された sensitivedata フォルダーへのアクセスをブロックする統合を作成します。

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

CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::001234567890:role/myrole'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket1/mypath1/sensitivedata/', 's3://mybucket2/mypath2/sensitivedata/');

Copy

ステップ4：Snowflakeアカウントの AWS IAM ユーザーを取得する¶

1. Snowflakeアカウント用に自動的に作成された IAM ユーザーの ARN を取得するには、 DESCRIBE INTEGRATION を使用します。

   DESC INTEGRATION <integration_name>;
   

   Copy

   条件:

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

   例:

   DESC INTEGRATION s3_int;
   
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
   | property                  | property_type | property_value                                                                 | property_default |
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------|
   | ENABLED                   | Boolean       | true                                                                           | false            |
   | STORAGE_ALLOWED_LOCATIONS | List          | s3://mybucket1/mypath1/,s3://mybucket2/mypath2/                                | []               |
   | STORAGE_BLOCKED_LOCATIONS | List          | s3://mybucket1/mypath1/sensitivedata/,s3://mybucket2/mypath2/sensitivedata/    | []               |
   | STORAGE_AWS_IAM_USER_ARN  | String        | arn:aws:iam::123456789001:user/abc1-b-self1234                                 |                  |
   | STORAGE_AWS_ROLE_ARN      | String        | arn:aws:iam::001234567890:role/myrole                                          |                  |
   | STORAGE_AWS_EXTERNAL_ID   | String        | MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=                               |                  |
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
   

   Copy

2. 次のプロパティの値を記録します。

   プロパティ	説明
   STORAGE_AWS_IAM_USER_ARN	Snowflakeアカウント用に作成された AWS IAM ユーザー。例えば、 arn:aws:iam::123456789001:user/abc1-b-self1234。Snowflakeは、Snowflakeアカウント全体用に単一の IAM ユーザーをプロビジョニングします。アカウント内のすべてのS3ストレージ統合では、その IAM ユーザーが使用されます。
   STORAGE_AWS_EXTERNAL_ID	Snowflakeが AWS との信頼関係を確立するために使用する外部 ID。ストレージ統合の作成時に外部 ID （STORAGE_AWS_EXTERNAL_ID）を指定しなかった場合、Snowflakeは ID を生成して使用します。

   これらの値は次のセクションで提供します。

ステップ5：バケットオブジェクトにアクセスするために IAM ユーザー権限を付与する¶

次の手順では、S3バケットを使用してデータをロードおよびアンロードできるように、 AWS 管理コンソールでSnowflakeの IAM アクセス許可を構成する方法を説明します。

1. AWS 管理コンソールにログインします。

2. Identity & Access Management （IAM）を選択します。

   

3. 左側のナビゲーションペインから Roles を選択します。

4. ステップ2: IAM AWS ロールを作成する （このトピック内）で作成したロールを選択します。

5. Trust relationships タブを選択します。

6. Edit trust relationship を選択します。

7. ステップ4：Snowflakeアカウントの AWS IAM ユーザーを取得する （このトピック内）で記録した DESC STORAGE INTEGRATION 出力値でポリシードキュメントを変更します。

   IAM ロールのポリシードキュメント

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "",
         "Effect": "Allow",
         "Principal": {
           "AWS": "<snowflake_user_arn>"
         },
         "Action": "sts:AssumeRole",
         "Condition": {
           "StringEquals": {
             "sts:ExternalId": "<snowflake_external_id>"
           }
         }
       }
     ]
   }
   

   Copy

   条件:

   • snowflake_user_arn は、記録した STORAGE_AWS_IAM_USER_ARN 値です。

   • snowflake_external_id は、記録した STORAGE_AWS_EXTERNAL_ID 値です。

     この例では、 snowflake_external_id の値は MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq= です。

     注釈

     セキュリティ上の理由から、外部 ID を指定せずに新しいストレージ統合を作成した場合（または CREATE OR REPLACE STORAGE INTEGRATION 構文を使用して既存のストレージ統合を再作成した場合）、新しい統合には 異なる 外部 ID が含まれるため、信頼ポリシーを更新しない限り信頼関係を解決できません。

8. Update Trust Policy ボタンを選択します。変更が保存されます。

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

CREATE STAGE コマンドを使用して、S3バケットを参照する外部ステージを作成します。Snowpipeはステージからデータファイルをフェッチし、ターゲットテーブルにロードする前に一時的にキューに入れます。または、既存の外部ステージを使用できます。

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

USE SCHEMA snowpipe_db.public;

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

Copy

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

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

次の例では、ユーザーセッションのアクティブスキーマに mypipe という名前のパイプを作成します。パイプは、 mystage ステージでステージングされたファイルから mytable テーブルにデータをロードします。

create pipe snowpipe_db.public.mypipe auto_ingest=true as
  copy into snowpipe_db.public.mytable
  from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

Copy

AUTO_INGEST=true パラメーターは、新しいデータをロードする準備ができたときに、S3バケットから SQS キューに送信されたイベント通知を読み取ることを指定します。

ステップ3: セキュリティを構成する¶

Snowpipeを使用して連続的なデータロードを実行するユーザーごとに、データロードのオブジェクト（ターゲットデータベース、スキーマ、テーブル、ステージオブジェクト、パイプ）に十分なアクセス制御権限を付与します。

Snowpipeを使用するには、次の権限を持つロールが必要です。

GRANT <権限> コマンドを使用して、ロールに権限を付与します。

たとえば、 snowpipe_db.public データベースオブジェクトのセットと mypipe という名前のパイプにアクセスできる snowpipe_role という名前のロールを作成します。次に、ロールをユーザーに付与します。

-- Create a role to contain the Snowpipe privileges
use role securityadmin;

create or replace role snowpipe_role;

-- Grant the required privileges on the database objects
grant usage on database snowpipe_db to role snowpipe_role;

grant usage on schema snowpipe_db.public to role snowpipe_role;

grant insert, select on snowpipe_db.public.mytable to role snowpipe_role;

grant usage on stage snowpipe_db.public.mystage to role snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = true;

-- Grant the OWNERSHIP privilege on the pipe object
grant ownership on pipe snowpipe_db.public.mypipe to role snowpipe_role;

-- Grant the role to a user
grant role snowpipe_role to user jsmith;

-- Set the role as the default role for the user
alter user jsmith set default_role = snowpipe_role;

-- Resume the pipe
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = false;

Copy

ステップ4: イベント通知を構成する¶

S3バケットのイベント通知を構成して、新しいデータをロードできるようになったときにSnowpipeに通知します。自動インジェスト機能は、S3からSnowpipeにイベント通知を配信するために SQS キューに依存しています。

使いやすくするために、Snowpipe SQS キューはSnowflakeによって作成および管理されます。SHOW PIPES コマンド出力には、 SQS キューのAmazonリソースネーム（ARN）が表示されます。

1. SHOW PIPES コマンドを実行します。

   SHOW PIPES;
   

   Copy

   notification_channel 列のステージには、 SQS キューの ARN があります。 ARN を便利な場所にコピーします。

   注釈

   AWS ガイドラインに従って、SnowflakeはS3バケットごとに1つ以下の SQS キューを指定します。この SQS キューは、同じ AWS アカウントの複数のバケットで共有できます。 SQS キューは、S3バケットの外部ステージをターゲットテーブルに接続するすべてのパイプの通知を調整します。データファイルがバケットにアップロードされると、ステージディレクトリパスに一致するすべてのパイプが、対応するターゲットテーブルへのファイルの1回限りのロードを実行します。

2. Amazon S3コンソールにログインします。

3. Amazon S3ドキュメント に記載されている手順を使用して、S3バケットのイベント通知を構成します。次のようにフィールドに入力します。

   • Name ：イベント通知の名前（例： Auto-ingest Snowflake）。

   • Events ： ObjectCreate (All) オプションを選択します。

   • Send to ：ドロップダウンリストから SQS Queue を選択します。

   • SQS ：ドロップダウンリストから Add SQS queue ARN を選択します。

   • SQS queue ARN ： SHOW PIPES 出力から SQS キュー名を貼り付けます。

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

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

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

SQS 通知が構成される 前に 外部ステージに存在したデータファイルのバックログをロードするには、 履歴データのロード をご参照ください。

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

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

前提条件：Amazon SNS トピックとサブスクリプションを作成する¶

1. AWS アカウントで SNS トピックを作成して、S3バケットのSnowflakeステージロケーションのすべてのメッセージを処理します。

2. S3イベント通知（例：他の SQS キューまたは AWS Lambdaワークロード）のターゲットの宛先をこのトピックにサブスクライブします。SNS バケットのイベント通知をトピックのすべてのサブスクライバーにパブリッシュします。

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

ステップ1：Snowflake SQS キューを SNS トピックに登録する¶

1. AWS 管理コンソールにログインします。

2. ホームダッシュボードから、 Simple Notification Service （SNS）を選択します。

3. 左側のナビゲーションペインから Topics を選択します。

4. S3バケットのトピックを見つけます。トピック ARN に注意します。

5. Snowflakeクライアントを使用して、 SNS トピック ARNで SYSTEM$GET_AWS_SNS_IAM_POLICY システム関数をクエリします。

   select system$get_aws_sns_iam_policy('<sns_topic_arn>');
   

   Copy

   この関数は、 SNS トピックをサブスクライブするSnowflake SQS キュー許可を付与する IAM ポリシーを返します。

   例:

   select system$get_aws_sns_iam_policy('arn:aws:sns:us-west-2:001234567890:s3_mybucket');
   
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | SYSTEM$GET_AWS_SNS_IAM_POLICY('ARN:AWS:SNS:US-WEST-2:001234567890:S3_MYBUCKET')                                                                                                                                                                   |
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | {"Version":"2012-10-17","Statement":[{"Sid":"1","Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"},"Action":["sns:Subscribe"],"Resource":["arn:aws:sns:us-west-2:001234567890:s3_mybucket"]}]}                 |
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   

   Copy

6. AWS 管理コンソールに戻ります。左側のナビゲーションペインから Topics を選択します。

7. S3バケットのトピックを選択し、 Edit ボタンをクリックします。 Edit ページが開きます。

8. Access policy - Optional をクリックして、ページのこの領域を展開します。

9. SYSTEM$GET_AWS_SNS_IAM_POLICY 関数の結果から追加された IAM ポリシーを JSON ドキュメントにマージします。

   例:

   元の IAM ポリシー（略称）：

   {
     "Version":"2008-10-17",
     "Id":"__default_policy_ID",
     "Statement":[
        {
           "Sid":"__default_statement_ID",
           "Effect":"Allow",
           "Principal":{
              "AWS":"*"
           }
           ..
        }
      ]
    }
   

   Copy

   統合された IAM ポリシー：

   {
     "Version":"2008-10-17",
     "Id":"__default_policy_ID",
     "Statement":[
        {
           "Sid":"__default_statement_ID",
           "Effect":"Allow",
           "Principal":{
              "AWS":"*"
           }
           ..
        },
        {
           "Sid":"1",
           "Effect":"Allow",
           "Principal":{
             "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
            },
            "Action":[
              "sns:Subscribe"
            ],
            "Resource":[
              "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
            ]
        }
      ]
    }
   

   Copy

10. 追加のポリシー許可を追加して、S3がバケットのイベント通知を SNS トピックに公開できるようにします。

    例（これらの手順全体で使用される SNS トピック ARN およびS3バケットを使用）:

    {
        "Sid":"s3-event-notifier",
        "Effect":"Allow",
        "Principal":{
           "Service":"s3.amazonaws.com"
        },
        "Action":"SNS:Publish",
        "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
        "Condition":{
           "ArnLike":{
              "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
           }
        }
     }
    

    Copy

    統合された IAM ポリシー：

    {
      "Version":"2008-10-17",
      "Id":"__default_policy_ID",
      "Statement":[
         {
            "Sid":"__default_statement_ID",
            "Effect":"Allow",
            "Principal":{
               "AWS":"*"
            }
            ..
         },
         {
            "Sid":"1",
            "Effect":"Allow",
            "Principal":{
              "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
             },
             "Action":[
               "sns:Subscribe"
             ],
             "Resource":[
               "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
             ]
         },
         {
            "Sid":"s3-event-notifier",
            "Effect":"Allow",
            "Principal":{
               "Service":"s3.amazonaws.com"
            },
            "Action":"SNS:Publish",
            "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
            "Condition":{
               "ArnLike":{
                  "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
               }
            }
          }
       ]
     }
    

    Copy

11. Save changes ボタンをクリックします。

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

CREATE STAGE コマンドを使用して、S3バケットを参照する外部ステージを作成します。Snowpipeはステージからデータファイルをフェッチし、ターゲットテーブルにロードする前に一時的にキューに入れます。

または、既存の外部ステージを使用できます。

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

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

Copy

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

CREATE PIPE コマンドを使用してパイプを作成します。パイプは、Snowpipeがインジェスションキューからターゲットテーブルにデータをロードするために使用する COPY INTO <テーブル> ステートメントを定義します。COPY ステートメントで、 前提条件：Amazon SNS とサブスクリプションを作成する から SNS トピック ARN を識別します。

次の例では、ユーザーセッションのアクティブスキーマに mypipe という名前のパイプを作成します。パイプは、 mystage ステージでステージングされたファイルから mytable テーブルにデータをロードします。

create pipe snowpipe_db.public.mypipe
  auto_ingest=true
  aws_sns_topic='<sns_topic_arn>'
  as
    copy into snowpipe_db.public.mytable
    from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

Copy

条件:

AUTO_INGEST = true

新しいデータをロードする準備ができたときに、S3バケットから SQS キューに送信されたイベント通知を読み取ることを指定します。

AWS_SNS_TOPIC = '<SNSトピックARN>'

現在の例 arn:aws:sns:us-west-2:001234567890:s3_mybucket のように、S3バケットの SNS トピックの ARN を指定します。CREATE PIPE ステートメントは、指定された SNS トピックにSnowflake SQS キューをサブスクライブします。パイプは、 SNS トピックを介したイベント通知によってトリガーされたインジェストキューにのみファイルをコピーします。

パイプからいずれかのパラメーターを削除するには、現在、 CREATE OR REPLACE PIPE 構文を使用してパイプを再作成する必要があります。

ステップ4: セキュリティを構成する¶

Snowpipeを使用して連続的なデータロードを実行するユーザーごとに、データロードのオブジェクト（ターゲットデータベース、スキーマ、テーブル、ステージオブジェクト、パイプ）に十分なアクセス制御権限を付与します。

Snowpipeを使用するには、次の権限を持つロールが必要です。

GRANT <権限> コマンドを使用して、ロールに権限を付与します。

たとえば、 snowpipe_db.public データベースオブジェクトのセットと mypipe という名前のパイプにアクセスできる snowpipe_role という名前のロールを作成します。次に、ロールをユーザーに付与します。

-- Create a role to contain the Snowpipe privileges
use role securityadmin;

create or replace role snowpipe_role;

-- Grant the required privileges on the database objects
grant usage on database snowpipe_db to role snowpipe_role;

grant usage on schema snowpipe_db.public to role snowpipe_role;

grant insert, select on snowpipe_db.public.mytable to role snowpipe_role;

grant usage, read on stage snowpipe_db.public.mystage to role snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = true;

-- Grant the OWNERSHIP privilege on the pipe object
grant ownership on pipe snowpipe_db.public.mypipe to role snowpipe_role;

-- Grant the role to a user
grant role snowpipe_role to user jsmith;

-- Set the role as the default role for the user
alter user jsmith set default_role = snowpipe_role;

-- Resume the pipe
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = false;

Copy

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

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

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

SQS 通知が構成される 前に 外部ステージに存在したデータファイルのバックログをロードするには、 履歴データのロード をご参照ください。

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

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

ステップ1: Amazon SNS トピックを作成する¶

前提条件: Amazon SNS トピックとサブスクリプションを作成する （このトピック内）に従います。

ステップ2: EventBridge ルールを作成してS3バケットをサブスクライブし、 SNS トピックに通知を送信します¶

• S3バケットの Amazon EventBridge を有効にします。

• EventBridge ルールを作成し、ステップ1で作成した SNS トピックに 通知を送信 します。

ステップ3: SQS 通知を使用してSnowpipeを自動化するためにAmazon SNS を構成する¶

オプション2: SQS 通知を使用してSnowpipeを自動化するために、Amazon SNS を構成する （このトピック内）に従います。