暂存区、管道和加载历史记录复制

This topic provides information about replication support for data pipeline objects and related metadata, including stages, storage integrations, pipes, and load history. You can replicate these objects to configure failover for ingest and ETL pipelines across regions and across cloud platforms.

Before you get started, we recommend that you have familiarity with Snowflake support for replication and failover/failback. For more information, see Introduction to replication and failover across multiple accounts.

要求

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.

例如：

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

复制和暂存区

本节介绍了 Snowflake 当前针对不同类型暂存区支持的复制功能级别。

内部暂存区的复制

下表描述了每种类型的内部暂存区的复制工作方式。

外部暂存区的复制

下表描述了每种类型的外部暂存区的复制工作方式。

注意事项

复制和管道

本节介绍了当前针对不同类型管道支持的复制功能级别。

Snowflake 支持复制以下内容：

• 管道对象，包括自动引入和从外部暂存区加载数据的 REST 端点管道。
• 管道级参数。
• 对管道对象授予的权限。

辅助数据库中的管道

Pipes in a secondary database are in a READ_ONLY execution state and receive notifications but do not load data until you promote the secondary database to serve as the primary. After you promote a secondary database, the pipes will transition to a FAILING_OVER execution state. Once failover is complete, the pipes should be in the RUNNING execution state and begin to load any data that is available since the last refresh time (that is, the last time that the former primary database was updated).

自动引入管道的复制

在发生故障转移时，复制的自动引入管道将成为新的主管道，并且可以执行以下操作：

• Load any data that has not yet been loaded. This includes any data that is new since the newly promoted primary database was last refreshed.

• 当暂存区有新文件要加载时，继续接收通知，并从这些文件加载数据。

  Note

  To receive notifications, you must configure a secondary auto-ingest pipe in a target account prior to failover. For more information, see Configure notifications for secondary auto-ingest pipes.

REST 端点管道的复制¶

For pipes that use the Snowpipe REST API to load data, Snowflake replicates the pipes and their load history metadata to each target account that you specify. There are no additional configuration steps you need to take on the target accounts. For a detailed list of load history metadata, see Load metadata.

若要在发生故障转移时继续加载数据，请通过新提升的源账户调用 REST API。

注意事项

示例 1：复制命名内部暂存区¶

该示例演示了内部暂存区的复制工作方式。具体而言，该示例显示了目录表如何在复制前后成为暂存区元数据的单一事实来源。

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.

   Copy codeExpand code block

   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;

   Show lessSee more

   Scroll to top

2. 创建故障转移组：

   Copy codeExpand code block

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

   Show lessSee more

   Scroll to top

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.

   Copy codeExpand code block

   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;

   Show lessSee more

   Scroll to top

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.

   Copy codeExpand code block

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

   Show lessSee more

   Scroll to top

示例 2：复制外部暂存区和存储集成¶

此示例提供了一个示例工作流，用于将外部暂存区和存储集成复制到目标账户。

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.

   Copy codeExpand code block

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

   Show lessSee more

   Scroll to top

2. Create an external stage in database my_database_2 using storage integration my_storage_int.

   Copy codeExpand code block

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

   Show lessSee more

   Scroll to top

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

   Copy codeExpand code block

   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;

   Show lessSee more

   Scroll to top

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

1. 创建故障转移组，并将其用作源账户中故障转移组的副本，然后刷新。

   Copy codeExpand code block

   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;

   Show lessSee more

   Scroll to top

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.

示例 3：复制自动引入管道¶

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.

该示例假定您已完成以下任务：

• 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.

   Copy codeExpand code block

   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');

   Show lessSee more

   Scroll to top

2. Refresh the pipe with an ALTER PIPE statement to load data from the stage from the last 7 days.

   Copy codeExpand code block

   ALTER PIPE mypipe REFRESH;

   Show lessSee more

   Scroll to top

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

   Copy codeExpand code block

   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;

   Show lessSee more

   Scroll to top

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

1. 创建故障转移组，并将其用作源账户中故障转移组的副本。

   Copy codeExpand code block

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

   Show lessSee more

   Scroll to top

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.

   Copy codeExpand code block

   DESC INTEGRATION my_s3_storage_int;

   Show lessSee more

   Scroll to top

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.

   Copy codeExpand code block

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

   Show lessSee more

   Scroll to top

   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. 刷新新故障转移组中的对象。

   Copy codeExpand code block

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;

   Show lessSee more

   Scroll to top

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

   Copy codeExpand code block

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;

   Show lessSee more

   Scroll to top

   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.

要验证复制的管道是否正常工作，请从管道的 COPY 语句中查询表。

SELECT * FROM mytable;

迁移到 Amazon Simple Notification Service (SNS)¶

本节介绍如何从直接将 Amazon S3 事件通知发送到 Amazon Simple Queue Service (SQS) 队列，迁移到使用 Amazon Simple Notification Service (SNS) 主题，此迁移适用于以下情况：

• Refresh directory tables automatically for Amazon S3
• Automating Snowpipe for Amazon S3

复制目录表或管道时，Snowflake 会在目标账户中创建一个新的 SQS 队列以处理自动化。您可以配置单个 SNS 主题，将事件通知从 S3 桶传送到多个账户的所有 SQS 队列。通过将 S3 事件通知广播到每个 SQS 队列，可以降低故障转移后丢失通知和数据的风险。

先决条件

要进行迁移，您必须满足以下条件：

• 您已为 S3 桶设置了一个或多个事件通知。有关说明，请参阅适用于您用例的主题：

  • Refreshing Directory Tables Automatically for Amazon S3: Creating a New S3 Event Notification
  • Creating a New S3 Event Notification to Automate Snowpipe

• 您已在目标账户中创建了复制组或故障转移组，该账户包括一个带有目录表或管道的暂存区。

迁移到 SNS 主题¶

1. Create an SNS topic in your AWS account. For instructions, see Creating an Amazon SNS topic (https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html) in the AWS SNS documentation.

2. Subscribe your target destinations (for example, other SQS queues or AWS Lambda workloads) for your S3 event notification(s) to your SNS topic. SNS publishes event notifications for your bucket to all subscribers to the topic. For instructions, see the AWS SNS documentation (https://docs.aws.amazon.com/sns/latest/dg/sns-create-subscribe-endpoint-to-topic.html).

3. 使用以下权限更新主题的访问策略：

   • Allow the Snowflake IAM user to subscribe the SQS queue that is in your target account to your topic.
   • 允许 Amazon S3 将事件通知从桶发布到 SNS 主题。

   For instructions, see Step 1: Subscribe the Snowflake SQS Queue to the SNS Topic.

4. In your target Snowflake account, call the SYSTEM$CONVERT_PIPES_SQS_TO_SNS function. The function subscribes the SQS queue in your target account to your SNS topic without interrupting metadata synchronization or ingestion work.

指定 S3 桶名称和 SNS 主题 ARN。

SELECT SYSTEM$CONVERT_PIPES_SQS_TO_SNS('s3_mybucket', 'arn:aws:sns:us-west-2:001234567890:MySNSTopic')

5. Update your S3 event notifications to use your SNS topic as a destination. For instructions, see the Amazon S3 User Guide (https://docs.aws.amazon.com/AmazonS3/latest/userguide/enable-event-notifications.html).

完成这些步骤后，SQS 队列会自动解除与您的 S3 事件通知的绑定。使用指定 S3 桶的所有目录表和管道都将开始使用 SNS 作为通知的来源。