Stage, pipe, and load history replication¶

Requirements¶

To replicate any external stages that use a storage integration, you must configure your replication or failover group to replicate STORAGE INTEGRATIONS. Otherwise, external stages are replicated without the associated storage integration.

You can use an ALTER REPLICATION GROUP or ALTER FAILOVER GROUP statement to modify these properties for an existing group.

If you add INTEGRATIONS to the OBJECT_TYPES list in your ALTER statement, include any other existing objects in the list to avoid dropping those objects in the target account. The same applies if you add STORAGE INTEGRATIONS to the ALLOWED_INTEGRATION_TYPES list.

For example:

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

Replication and stages¶

This section describes the current level of replication functionality that Snowflake supports for different types of stages.

Replication and pipes¶

This section describes the current level of replication functionality supported for different types of pipes.

Snowflake supports replication for the following:

• Pipe objects, including auto-ingest and REST endpoint pipes that load data from external stages.

• Pipe-level parameters.

• Privilege grants on pipe objects.

Example 1: Replicate a named internal stage¶

This example demonstrates how replication works for internal stages. In particular, the example shows how the directory table is the single source of truth for stage metadata before and after replication.

The first part of the example completes the following tasks in a source account.

1. Create an internal stage named my_int_stage with a directory table enabled to replicate the files on the stage. Then copy data from a table named my_table into files on the stage.

   Note

   The example refreshes the directory table after loading file1 and file2 onto the stage to synchronize the table metadata with the latest set of files in the stage definition for the directory tables. However, no refresh operation occurs after loading 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 a failover group:

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

   Copy

The second part of the example completes the replication and failover process in a target account:

1. Create a failover group as a replica of the failover group in the source account, refresh the objects in the new failover group, and promote the target account to serve as the source account.

   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. Next, refresh the directory table on the replicated stage and copy all of the files tracked by the directory table on my_stage into a table named my_table .

   Note

   The COPY INTO statement loads file1 and file2 into the table, but not file3. This is because the directory table was not refreshed after adding file3 in the source account.

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

   Copy

Example 2: Replicate an external stage and storage integration¶

This example provides a sample workflow for replicating an external stage and storage integration to a target account.

The example assumes that you have already completed the following: Configured secure access to your Amazon S3 bucket.

The first part of the example completes the following tasks in a source account.

1. Create a storage integration for an Amazon S3 bucket in database my_database_2.

   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. Create an external stage in database my_database_2 using storage integration my_storage_int.

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

   Copy

3. Create a failover group and include database my_database_2 and storage integration objects.

   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

The second part of the example completes the replication and failover process in a target account:

1. Create a failover group as a replica of the failover group in the source account and refresh.

   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. After you replicate the storage integration to the target account, you must take additional steps to update your cloud provider permissions to grant the replication integration access to your cloud storage. For more information, see Configure cloud storage access for secondary storage integrations.

Example 3: Replicate an auto-ingest pipe¶

This example provides a sample workflow for replicating a pipe that uses an Amazon Simple Notification Service (SNS) topic with Amazon Simple Queue Service (SQS) to automate Snowpipe.

The example assumes that you have already completed the following tasks:

• Created and configured a storage integration for Amazon S3. For example purposes, we use a storage integration named my_s3_storage_int.

• Created an Amazon SNS topic and subscription, and subscribed the Snowflake SQS queue to your SNS topic.

• Created an external stage that references your storage integration. For example purposes, we use a stage named my_s3_stage. For instructions, see CREATE STAGE.

Start with the following tasks in a source account.

1. Use the CREATE PIPE command to create a pipe with auto-ingest enabled that loads data from the external stage into a table named 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. Refresh the pipe with an ALTER PIPE statement to load data from the stage from the last 7 days.

   ALTER PIPE mypipe REFRESH;
   

   Copy

3. Finally, use CREATE FAILOVER GROUP to create a failover group that allows replication of storage integrations.

   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

The second part of the example completes the replication and failover process in a target account:

1. Create a failover group as a replica of the failover group in the source account.

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

   Copy

2. Execute a DESCRIBE INTEGRATION statement to retrieve the ARN for the AWS IAM User for your Snowflake account on the secondary deployment.

   Use the ARN to grant the IAM user permissions to access your S3 bucket. See Step 5: Grant the IAM User Permissions to Access Bucket Objects.

   DESC INTEGRATION my_s3_storage_int;
   

   Copy

3. Call the SYSTEM$GET_AWS_SNS_IAM_POLICY system function to generate an IAM policy that grants the new SQS queue permission to subscribe to your SNS topic. Snowflake created the new SQS queue in your target account when you replicated the failover group from your source account.

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

   Copy

   topic_arn is the Amazon Resource Name (ARN) of the SNS topic that you created for the original pipe in your source account.

   Then, Subscribe the new Amazon SQS queue to your SNS topic.

4. Refresh the objects in your new failover group.

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;
   

   Copy

5. Finally, promote the target account to serve as the source account with the ALTER FAILOVER GROUP command.

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;
   

   Copy

   The mypipe pipe will begin to load any data that was made available since the last time the failover group was refreshed in the source account.

   To verify that the replicated pipe is working, query the table from the pipe’s COPY statement.

   SELECT * FROM mytable;
   

   Copy

Migrate to Amazon Simple Notification Service (SNS)¶

This section covers how to migrate from sending Amazon S3 event notifications directly to an Amazon Simple Queue Service (SQS) queue to using an Amazon Simple Notification Service (SNS) topic for the following scenarios:

• Refreshing directory tables automatically for Amazon S3

• Automating Snowpipe for Amazon S3

When you replicate a directory table or pipe, Snowflake creates a new SQS queue in your target account to handle automation. You can configure a single SNS topic to deliver event notifications from your S3 bucket to all SQS queues across multiple accounts. By broadcasting your S3 event notification(s) to every SQS queue, you reduce the risk of losing notifications and data after failover.