数据管道多地容灾能力¶

要求¶

• 版本： Snowflake Business Critical Edition（或更高版本）。

• 支持的引入方法： 此功能仅支持通过 Snowpipe（自动引入）和 COPY INTO <table> 进行基于文件的数据加载。不支持 Openflow 或 Snowpipe Streaming。

• 相同的路径结构： 为了使您的管道在故障转移后能够定位到新文件，您必须使用与主存储位置完全相同的层级结构、文件夹结构和相对路径将文件写入辅助存储位置。

注意事项¶

• 计费： 此功能会产生标准的复制费用（数据传输和计算资源），将计入您的目标账户。

• 暂存区修改停机时间： 更改现有暂存区的 RELATIVE_URL 属性将使依赖对象失效并停止引入。我们建议在设置期间创建新的暂存区以避免停机。

• 多队列通知集成 (MQNI)： 不支持在源账户和目标账户中使用相同的活动队列。这样做可能会导致通知丢失。Snowflake 不会检查是否跨账户使用了相同的队列。

• 目录表： 当前不支持在使用 MLSI 的暂存区上创建目录表。

复制行为¶

• 异步复制： Snowflake 将您的表和管道的加载历史记录在完全相同的快照中一起复制。由于它们是同步的，因此中断不会导致数据重复。如果您的辅助数据库落后四个小时，那么表数据也落后四个小时，处理四个小时的排队通知只是将表更新到最新状态。

• D双写避免数据丢失： 您的恢复点对象 (RPO) 由您的复制刷新间隔决定。为防止故障转移期间数据丢失，您的辅助云消息队列的消息保留期必须长于您的复制间隔。如果你的队列在计划复制追赶上进度之前丢弃了消息，那么这些文件在故障转移后将不会引入。

• 单写数据丢失风险： 如果您使用单写路由，则在您上次成功复制之后在主位置处理的任何文件，对于辅助位置来说都是完全未知的。在故障转移时，这些数据将暂时在您的目标账户中缺失。

1.双写 – 推荐¶

生产者应用程序同时将文件写入主云存储桶和辅助云存储桶。辅助消息队列会累积通知，但由于辅助数据库为只读，Snowflake 不会处理这些通知。

• 故障转移时会发生什么： 辅助数据库变为可写。Snowpipe 读取辅助队列，并利用复制的加载历史记录对文件进行去重。如果一次中断导致某个文件未能在主位置完成写入，辅助管道会读取来自辅助队列的通知，发现该文件在加载历史记录中缺失，然后将其引入。

• 故障恢复时会发生什么： 当主位置恢复，并且您刷新故障转移组然后执行故障恢复时，Snowpipe 会自动开始引入新文件，因为在故障恢复准备期间，加载历史记录已从辅助账户同步过来。

• 结果： 无数据丢失，无重复数据。Snowflake 会在双向自动处理数据对账。

• 所需操作： 无需额外操作，只需执行故障恢复前的标准复制同步 (ALTER FAILOVER GROUP ... REFRESH)，以确保主账户拥有最新的加载历史记录。

2.单写路由¶

生产者仅写入主云存储。发生中断时，您需要重新路由生产者，使其开始将新文件写入辅助云存储。

• 故障转移时会发生什么： 辅助账户立即开始处理路由到辅助桶的新文件。但是，滞留在受影响主位置中的传输中文件会暂时遗留在那里。

• 故障恢复时会发生什么： 当主位置恢复，并且您故障恢复到主 Snowflake 账户时，Snowpipe 会自动处理在中断发生前成功到达队列的所有文件通知。

• 结果： 无重复数据。但是，因中断而完全未能生成云通知的任何文件（或者中断持续时间超过了队列消息保留策略所允许的范围），都需要手动介入处理。

• 所需操作： 故障恢复后，将主存储桶与 Snowflake 中的 COPY_HISTORY 视图进行比较，以找出任何缺失的文件。运行 ALTER PIPE ... REFRESH 或手动执行 COPY INTO 命令，以加载这些特定的滞留文件。

第 1 步：创建多位置存储集成 (MLSI)¶

要配置多位置存储集成，您需遵循配置存储集成的标准步骤，但本部分指出了一些差异之处。

在您的源账户中，通过为 STORAGE_LOCATIONS 列表中的每个位置提供值来创建 MLSI。您可以混搭不同云提供商，以实现跨云配置。

CREATE STORAGE INTEGRATION my_mlsi
  TYPE = EXTERNAL_STAGE
  STORAGE_LOCATIONS =
  (
    (
      NAME = 'my-s3-us-west-1'
      STORAGE_PROVIDER = 'S3'
      STORAGE_BASE_URL = 's3://myBucketWest'
      STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::12345:role/myrole'
      STORAGE_AWS_EXTERNAL_ID = 'mlsi-external-id'
      ENCRYPTION = ( TYPE = 'AWS_SSE_S3' )
    ),
    (
      NAME = 'my-s3-us-east-1'
      STORAGE_PROVIDER = 'S3'
      STORAGE_BASE_URL = 's3://myBucketEast'
      STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::67890:role/myrole'
      STORAGE_AWS_EXTERNAL_ID = 'mlsi-external-id'
      ENCRYPTION = ( TYPE = 'AWS_SSE_S3' )
    )
  )
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('*')
  ACTIVE = 'my-s3-us-west-1';

其中：

• STORAGE_LOCATIONS： 为存储集成指定一个或多个存储位置（S3 桶、GCS 桶或 Azure 容器）的列表。要查看各云提供商的参数，请参阅 CREATE STORAGE INTEGRATION 参考页面中的 云提供商参数 (cloudProviderParams)。

• NAME： 字符串，用于指定存储位置的标识符（名称）。

• ENCRYPTION： 指定存储位置的加密方式。您必须在存储集成级别而非暂存级别为存储位置指定加密。仅在从加密文件加载时需要；如果存储位置和文件均未加密，则不需要。要查看各云提供商的加密选项，请参阅 CREATE STAGE 参考页面中的 ENCRYPTION。

• ACTIVE： 指定存储位置的名称，用于将其设置为当前账户中存储集成的活动位置。

  对于源账户中的活动存储位置，您必须设置访问控制并授予 Snowflake 访问您存储的权限。请按照以下主题中的说明进行操作：

  • 配置 Snowflake 存储集成以访问 Amazon S3

  • 为 Google Cloud Storage 配置集成

  • 配置 Azure 容器以加载数据

第 2 步：第 2 步：将 MLSI 与外部暂存区关联¶

我们强烈建议创建新暂存区，而不是更改现有暂存区。

使用 CREATE STAGE 命令，将您创建的多位置存储集成与一个或多个外部暂存区关联：

CREATE STAGE my_ext_stage
  RELATIVE_URL = '/my_folder/my_sub_folder/'
  STORAGE_INTEGRATION = 'my_mlsi';

其中：

• RELATIVE_URL： 从存储集成中定义的存储位置到外部暂存区位置的相对路径。为了让您的管道在故障转移后能够定位新文件，您必须使用与主位置相同的层级结构、文件夹结构和相对路径，将文件写入辅助存储位置。

• STORAGE_INTEGRATION： 您的多位置存储集成的名称。

例如：

ALTER STAGE my_ext_stage SET
  RELATIVE_URL = '/my_folder/my_sub_folder/'
  STORAGE_INTEGRATION = 'my_mlsi';

第 3 步：配置多队列通知集成 (MQNI)¶

如果您通过云消息服务使用自动数据加载，并且已为外部暂存区配置了多位置存储集成，则还必须使用多队列通知集成，以实现 Snowpipe 管道的无缝故障转移。

对于您为通知集成定义的每个队列，您必须按照以下主题中的步骤准备消息服务：

• 配置 Amazon SNS 使用 SQS 通知自动执行 Snowpipe。为 MLSI 中拥有存储位置的每个 AWS 区域创建一个 SNS 主题。

• 使用 GCS Pub/Sub 配置自动化

• 使用 Azure 事件网格配置自动化

CREATE NOTIFICATION INTEGRATION my_mqni
  ENABLED = TRUE
  TYPE = MULTI_QUEUE
  DIRECTION = INBOUND
  QUEUES = (
    (
      NAME = 'my-us-west-1'
      NOTIFICATION_PROVIDER = AWS_SNS
      AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-west'
    ),
    (
      NAME = 'my-us-east-1'
      NOTIFICATION_PROVIDER = AWS_SNS
      AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-east'
    )
  )
  ACTIVE = 'my-us-west-1';

CREATE PIPE my_pipe
  AUTO_INGEST = TRUE
  INTEGRATION = my_mqni
  AS COPY INTO my_table FROM @my_ext_stage/my_pipe/;

SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  '<new_mqni_name>',
  '<original_sns_topic_arn_or_int_name>',
  '<new_sns_topic_arn_or_int_name>'
)

SELECT SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  'my_mqni',
  'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-west',
  'arn:aws:sns:us-east-1:67890:my-snowpipe-mlsi-east'
);

SELECT SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  'my_azure_mqni',
  'my_azure_ni_1',
  'my_azure_ni_2'
);

第 4 步：将 MLSI 和 MQNI 复制到目标账户¶

1.要复制多位置存储集成和多队列通知集成，请更改现有复制组或故障转移组，在 ALLOWED_INTEGRATION_TYPES 列表中包含 STORAGE INTEGRATIONS 和 NOTIFICATION INTEGRATIONS。

例如，使用 ALTER FAILOVER GROUP 命令：

ALTER FAILOVER GROUP my_fg SET
  OBJECT_TYPES = DATABASES, ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS,
    NOTIFICATION INTEGRATIONS;

2. 然后，在目标账户中执行刷新操作：

ALTER FAILOVER GROUP my_fg REFRESH;

第 5 步：在目标账户中配置活动状态¶

执行刷新操作后，为确保在实际中断期间实现无缝故障转移，需要在目标账户中配置活动存储位置和队列（如果使用通知集成）。

在目标账户中：

1. 对于希望在目标账户中设置为活动位置的存储位置，请按照以下主题中的说明授予 Snowflake 对存储的访问权限：

   • :doc:` 选项 1：配置 Snowflake 存储集成以访问 Amazon S3 </user-guide/data-load-s3-config-storage-integration>`

   • 为 Google Cloud Storage 配置集成

   • 配置 Azure 容器以加载数据

2. 激活辅助存储：将 MLSI设置为在目标账户中使用辅助备份存储位置。

   CopyExpand

   ALTER STORAGE INTEGRATION my_mlsi SET ACTIVE = 'my-s3-us-east-1';
   

   Show lessSee more

   Scroll to top

3. 如果使用多队列通知集成，请授予 Snowflake 访问您希望在目标账户中设置为活动的队列的消息服务的权限。按照云提供商的说明进行操作：

   • 对于 AWS，请参阅 先决条件：创建 Amazon SNS 主题和订阅 中的将 Snowflake SQS 队列订阅至 SNS 主题。

   • 对于 Google Cloud，请参阅 使用 GCS Pub/Sub 配置自动化 中的授予 Snowflake 对 Pub/Sub 订阅的访问权限。

   • 对于 Azure，请参阅 使用 Azure 事件网格配置自动化 中的授予 Snowflake 对存储队列的访问权限.

4. 激活辅助队列（如果使用 MQNI)：将活动队列设置为目标账户中的辅助位置。

   CopyExpand

   ALTER INTEGRATION my_mqni
     SET ACTIVE = 'MY_MQNI-queue2';
   

   Show lessSee more

   Scroll to top

1.验证活动集成状态¶

通过检查存储和队列的属性，确认集成指向正确的存储和队列。在输出中查看 ACTIVE 属性：

-- Check the active storage location
DESCRIBE STORAGE INTEGRATION my_mlsi;

-- Check the active message queue
DESCRIBE INTEGRATION my_mqni;

2.检查管道状态（仅限 Snowpipe）¶

使用 SYSTEM$PIPE_STATUS 函数，确保管道正在运行，并检查它是否正在积极地将来自辅助位置的新文件排队。

SELECT SYSTEM$PIPE_STATUS('my_pipe');

查看“executionState”：“RUNNING”并检查“pendingFileCount”，以确认它正在积极识别放入辅助桶中的新文件。

3.验证引入是否成功（加载历史记录）¶

要确保加载数据时没有错误或重复，请查询 COPY_HISTORY 视图。这会准确显示引入了哪些文件、源路径以及加载时间。

SELECT file_name, status, row_count, last_load_time
FROM TABLE(information_schema.copy_history(
  table_name => 'my_table',
  start_time => DATEADD(hours, -1, CURRENT_TIMESTAMP())
));

验证 file_name 路径是否反映了活动存储位置，以及状态是否显示为 LOADED。