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

要求¶

要复制使用存储集成的任何外部暂存区，必须将复制或故障转移组配置为复制 STORAGE INTEGRATIONS。否则，将在没有关联的存储集成的情况下复制外部暂存区。

可以使用 ALTER REPLICATION GROUP 或 ALTER FAILOVER GROUP 语句修改现有组的这些属性。

如果在 ALTER 语句的 OBJECT_TYPES 列表中添加 INTEGRATIONS，请在列表中包含所有其他现有对象，避免在目标账户中删除这些对象。将 STORAGE INTEGRATIONS 添加到 ALLOWED_INTEGRATION_TYPES 列表中时，同样适用。

例如：

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

该示例的第二部分在 目标 账户中完成复制和故障转移过程：

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

该示例的第二部分在 目标 账户中完成复制和故障转移过程：

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

该示例的第二部分在 目标 账户中完成复制和故障转移过程：

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 系统函数生成 IAM 策略，授予新的 SQS 队列订阅 SNS 主题的权限。当您从源账户复制故障转移组时，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

迁移到 Amazon Simple Notification Service (SNS)¶

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

• 自动刷新 Amazon S3 的目录表

• 针对 Amazon S3 自动执行 Snowpipe

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