ステージ、パイプ、ロード履歴の複製¶

要件¶

データパイプラインオブジェクトを複製する前に、 ETL の複製を有効にする必要があります。ETL 複製のサポートは 2024_01 BCR バンドル の一部です。次のステートメントを実行すると、2024_01 BCRバンドルを 有効 にすることができます。

SELECT SYSTEM$ENABLE_BEHAVIOR_CHANGE_BUNDLE('2024_01');

ストレージ統合を使用する外部ステージを複製するには、 STORAGE INTEGRATIONS を複製するように複製またはフェールオーバーグループを構成する必要もあります。そうでなければ、外部ステージは関連するストレージの統合なしに複製されます。

ALTER REPLICATION GROUP または ALTER FAILOVER GROUP ステートメントを使用して、既存のグループのこれらのプロパティを変更できます。

例:

ALTER FAILOVER GROUP my_failover_group SET
  OBJECT_TYPES = ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS;

複製とステージ¶

このセクションでは、Snowflakeがさまざまなタイプのステージに対して現在サポートしている複製機能のレベルについて説明します。

複製とパイプ¶

このセクションでは、さまざまなタイプのパイプに対して現在サポートしている複製機能のレベルについて説明します。

Snowflakeは、次の複製をサポートしています。

• 外部ステージからデータをロードする自動インジェストおよび REST エンドポイントパイプを含むパイプオブジェクト。

• パイプレベルのパラメーター。

• パイプオブジェクトへの権限の付与。

例1: 名前付き内部ステージの複製¶

この例は、内部ステージの複製がどのように機能するかを示しています。特にこの例では、複製の前後で、ディレクトリテーブルがテーブルメタデータの唯一の信頼できる情報源であることを示しています。

例の最初の部分では、 ソース アカウントで次のタスクを完了します。

1. ステージのファイルを複製するために、ディレクトリテーブルを有効にした my_int_stage という名前の内部ステージを作成します。次に、 my_table という名前のテーブルからステージのファイルにデータをコピーします。

   注釈

   この例では、 file1 と file2 をステージにロードした後にディレクトリテーブルをリフレッシュして、テーブルのメタデータをディレクトリテーブルのステージ定義にある最新のファイルセットと同期させています。ただし、 file3 をロードした後にリフレッシュ操作は発生しません。

   CREATE OR REPLACE STAGE my_stage
     DIRECTORY = (ENABLE = TRUE);
   
   COPY INTO @my_stage/folder1/file1 from my_table;
   COPY INTO @my_stage/folder2/file2 from my_table;
   ALTER STAGE my_stage REFRESH;
   
   COPY INTO @my_stage/folder3/file3 from my_table;
   

   Copy

2. フェールオーバーグループを作成する

   CREATE FAILOVER GROUP my_stage_failover_group
     OBJECT_TYPES = DATABASES
     ALLOWED_DATABASES = my_database_1
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

この例の2番目の部分は、 ターゲット アカウントでの複製とフェールオーバー処理を完了します。

1. ソースアカウントのフェールオーバーグループのレプリカとしてフェールオーバーグループを作成し、新しいフェールオーバーグループのオブジェクトをリフレッシュし、ターゲットアカウントをソースアカウントとして機能するように昇格させます。

   CREATE FAILOVER GROUP my_stage_failover_group
     AS REPLICA OF myorg.my_account_1.my_stage_failover_group;
   
   ALTER FAILOVER GROUP my_stage_failover_group REFRESH;
   
   ALTER FAILOVER GROUP my_stage_failover_group PRIMARY;
   

   Copy

2. 次に、複製されたステージのディレクトリテーブルをリフレッシュし、 my_stage 上のディレクトリテーブルが追跡しているすべてのファイルを my_table という名前のテーブルにコピーします。

   注釈

   COPY INTO ステートメントは file1 と file2 をテーブルにロードしますが、 file3 はロードしません。これは、ソースアカウントに file3 を追加した後、ディレクトリテーブルがリフレッシュされなかったためです。

   ALTER STAGE my_stage REFRESH;
   
   COPY INTO my_table FROM @my_stage;
   

   Copy

例2: 外部ステージとストレージの統合を複製する¶

この例は、外部ステージとストレージ統合をターゲットアカウントに複製するためのサンプルワークフローを提供します。

この例は、既に以下を完了していることを前提としています: Amazon S3バケットへの安全なアクセスの構成。

例の最初の部分では、 ソース アカウントで次のタスクを完了します。

1. データベース my_database_2 のAmazon S3バケットにストレージ統合を作成します。

   CREATE STORAGE INTEGRATION my_storage_int
     TYPE = external_stage
     STORAGE_PROVIDER = 's3'
     STORAGE_ALLOWED_LOCATIONS = ('s3://mybucket/path')
     STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/blockedpath')
     ENABLED = true;
   

   Copy

2. ストレージ統合 my_storage_int を使用して、データベース my_database_2 に外部ステージを作成します。

   CREATE STAGE my_ext_stage
     URL = 's3://mybucket/path'
     STORAGE_INTEGRATION = my_storage_int
   

   Copy

3. フェールオーバーグループを作成し、データベース my_database_2 とストレージ統合オブジェクトを含めます。

   CREATE FAILOVER GROUP my_external_stage_fg
     OBJECT_TYPES = databases, integrations
     ALLOWED_INTEGRATION_TYPES = storage integrations
     ALLOWED_DATABASES = my_database_2
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

この例の2番目の部分は、 ターゲット アカウントでの複製とフェールオーバー処理を完了します。

1. ソースアカウントのフェールオーバーグループのレプリカとしてフェールオーバーグループを作成し、更新します。

   CREATE FAILOVER GROUP my_external_stage_fg
     AS REPLICA OF myorg.my_account_1.my_external_stage_fg;
   
   ALTER FAILOVER GROUP my_external_stage_fg REFRESH;
   

   Copy

2. ストレージ統合をターゲットアカウントに複製した後、複製統合にクラウドストレージへのアクセスを許可するために、クラウドプロバイダーのアクセス許可を更新する追加のステップを実行する必要があります。詳細については、 セカンダリストレージ統合のためにクラウドストレージアクセスを構成する をご参照ください。

例3: 自動インジェストパイプの複製¶

この例では、 Amazon Simple Notification Service（SNS）トピックとAmazon Simple Queue Service（SQS）を使用するパイプを複製し、Snowpipe を自動化するためのサンプルワークフローを提供します。

例では、すでに次のタスクを完了していることを前提としています。

• Amazon S3のストレージ統合の作成と構成。たとえば、 my_s3_storage_int という名前のストレージ統合を使用します。

• Amazon SNS トピックとサブスクリプションを作成し、Snowflake SQS キューを SNS トピックにサブスクライブします。

• ストレージ統合を参照する外部ステージを作成します。たとえば、 my_s3_stage という名前のステージを使用します。手順については、 CREATE STAGE をご参照ください。

ソース アカウントで次のタスクを開始します。

1. CREATE PIPE コマンドを使用して、外部ステージから mytable という名前のテーブルにデータをロードする自動インジェストを有効にしたパイプを作成します。

   CREATE PIPE snowpipe_db.public.mypipe AUTO_INGEST=TRUE
    AWS_SNS_TOPIC='<topic_arn>'
    AS
      COPY INTO snowpipe_db.public.mytable
      FROM @snowpipe_db.public.my_s3_stage
      FILE_FORMAT = (TYPE = 'JSON');
   

   Copy

2. ALTER PIPE ステートメントでパイプをリフレッシュし、ステージから過去7日間のデータをロードします。

   ALTER PIPE mypipe REFRESH;
   

   Copy

3. 最後に、 CREATE FAILOVER GROUP を使用してストレージ統合の複製を許可するフェールオーバーグループを作成します。

   CREATE FAILOVER GROUP my_pipe_failover_group
     OBJECT_TYPES = DATABASES, INTEGRATIONS
     ALLOWED_INTEGRATION_TYPES = STORAGE INTEGRATIONS
     ALLOWED_DATABASES = snowpipe_db
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

この例の2番目の部分は、 ターゲット アカウントでの複製とフェールオーバー処理を完了します。

1. ソースアカウントのフェールオーバーグループのレプリカとして、フェールオーバーグループを作成します。

   CREATE FAILOVER GROUP my_pipe_failover_group
     AS REPLICA OF myorg.my_account_1.my_pipe_failover_group;
   

   Copy

2. DESCRIBE INTEGRATION ステートメントを実行して、セカンダリデプロイメントのSnowflakeアカウントの AWS IAM ユーザーの ARN を取得します。

   ARN を使用して、IAM ユーザにS3バケットへのアクセス許可を与えます。 ステップ 5: IAM ユーザーにバケットオブジェクトへのアクセス権限を付与する を参照してください。

   DESC INTEGRATION my_s3_storage_int;
   

   Copy

3. SYSTEM$GET_AWS_SNS_IAM_POLICY システム関数を呼び出し、新しい SQS キューに SNS トピックをサブスクライブする権限を与える IAM ポリシーを生成します。Snowflakeは、ソースアカウントからフェールオーバーグループを複製したときに、ターゲットアカウントに新しい SQS キューを作成しました。

   SELECT SYSTEM$GET_AWS_SNS_IAM_POLICY('<topic_arn>');
   

   Copy

   topic_arn は、ソースアカウントで元のパイプ用に作成した SNS トピックのAmazonリソースネーム（ARN）です。

   次に、 新しいAmazonの SQS キューを SNS トピック にサブスクライブします。

4. 新しいフェールオーバーグループのオブジェクトをリフレッシュします。

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;
   

   Copy

5. 最後に、 ALTER FAILOVER GROUP コマンドを使用して、ターゲットアカウントをソースアカウントとして機能するように昇格させます。

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;
   

   Copy

   mypipe パイプは、ソースアカウントでフェールオーバーグループが最後にリフレッシュされたとき以降に使用可能になったデータのロードを開始します。

   複製されたパイプが機能していることを確認するには、パイプの COPY ステートメントからテーブルにクエリを実行してください。

   SELECT * FROM mytable;
   

   Copy