针对 Amazon S3 自动执行 Snowpipe¶

第 1 步：配置 S3 桶的访问权限¶

第 2 步：在 AWS 中创建 IAM 角色¶

要在 AWS 管理控制台中为 Snowflake 配置访问权限，请执行以下操作：

1. 在 Identity and Access Management (IAM) 控制面板的左侧导航窗格中，选择 Roles。

2. 选择 Create role。

3. 选择 AWS account 作为信任实体类型。

4. 在 Account ID 字段中，暂时输入自己的 AWS 账户 ID。在稍后的步骤中，您将修改信任关系并授予对 Snowflake 的访问权限。

5. 选择 Require external ID 选项。外部 ID 用于向第三方（如 Snowflake）授予对 AWS 资源（例如 S3 桶）的访问权限。

   输入占位符 ID，例如 0000。在后续步骤中，您将修改 IAM 角色的信任关系并指定外部 ID 用于存储集成。

6. 选择 Next。

7. 选择您在 第 1 步：配置 S3 桶访问权限 （本主题内容）中创建的策略。

8. 选择 Next。

   

9. 输入角色的名称和描述，然后选择 Create role。

   您现在已经为一个桶创建了 IAM 策略，创建了一个 IAM 角色，并将策略附加到了该角色。

10. 在角色摘要页面上，找到 Role ARN 值并将其记录下来。在下一步中，您将创建引用此角色的 Snowflake 集成。

第 3 步：在 Snowflake 中创建云存储集成¶

使用 CREATE STORAGE INTEGRATION 命令创建存储集成。存储集成是一个 Snowflake 对象，用于存储生成的身份和 S3 云存储访问管理 (IAM) 用户，以及一组可选的允许或阻止的存储位置（如桶）。组织中的云提供商管理员会将存储位置的权限授予生成的用户。此选项可让用户在创建暂存区或加载数据时无需提供凭据。

单个存储集成可以支持多个外部（如 S3）暂存区。暂存区定义中的 URL 必须与为 STORAGE_ALLOWED_LOCATIONS 参数指定的 S3 桶（和可选路径）一致。

CREATE STORAGE INTEGRATION <integration_name>
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('s3://<bucket>/<path>/', 's3://<bucket>/<path>/')
  [ STORAGE_BLOCKED_LOCATIONS = ('s3://<bucket>/<path>/', 's3://<bucket>/<path>/') ]

其中：

• integration_name 是新集成的名称。

• iam_role 是您在 第 2 步：在 AWS 中创建 IAM 角色 （本主题内容）中创建的角色的 Amazon 资源名称 (ARN)。

• bucket 是存储数据文件的 S3 桶的名称（例如 mybucket）。在创建或修改引用此集成的暂存区时，必需的 STORAGE_ALLOWED_LOCATIONS 参数和可选的 STORAGE_BLOCKED_LOCATIONS 参数分别限制或阻止对这些桶的访问。

• path 是一个可选路径，可用于提供对桶中对象的精细控制。

下面的示例创建了一个集成，允许访问账户中的所有桶，但阻止访问定义的 sensitivedata 文件夹。

也使用此集成的其他外部暂存区可以引用允许的桶和路径：

CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::001234567890:role/myrole'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket1/mypath1/sensitivedata/', 's3://mybucket2/mypath2/sensitivedata/');

Copy

第 4 步：检索您的 Snowflake 账户的 AWSIAM 用户¶

1. 要检索自动为您的 Snowflake 账户创建的 IAM 用户的 ARN，请使用 DESCRIBE INTEGRATION。

   DESC INTEGRATION <integration_name>;
   

   Copy

   其中：

   • integration_name 是您在 第 3 步：在 Snowflake 中创建云存储集成 （本主题内容）中创建的集成的名称。

   例如：

   DESC INTEGRATION s3_int;
   
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
   | property                  | property_type | property_value                                                                 | property_default |
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------|
   | ENABLED                   | Boolean       | true                                                                           | false            |
   | STORAGE_ALLOWED_LOCATIONS | List          | s3://mybucket1/mypath1/,s3://mybucket2/mypath2/                                | []               |
   | STORAGE_BLOCKED_LOCATIONS | List          | s3://mybucket1/mypath1/sensitivedata/,s3://mybucket2/mypath2/sensitivedata/    | []               |
   | STORAGE_AWS_IAM_USER_ARN  | String        | arn:aws:iam::123456789001:user/abc1-b-self1234                                 |                  |
   | STORAGE_AWS_ROLE_ARN      | String        | arn:aws:iam::001234567890:role/myrole                                          |                  |
   | STORAGE_AWS_EXTERNAL_ID   | String        | MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=                               |                  |
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
   

   Copy

2. 记录以下属性的值：

   属性	描述
   STORAGE_AWS_IAM_USER_ARN	为您的 Snowflake 账户创建的 AWS IAM 用户，例如 arn:aws:iam::123456789001:user/abc1-b-self1234。Snowflake 会为整个 Snowflake 账户配置单一 IAM 用户。您账户中的所有 S3 存储集成都使用该 IAM 用户。
   STORAGE_AWS_EXTERNAL_ID	Snowflake 与 AWS 建立信任关系所使用的外部 ID。如果创建存储集成时未指定外部 ID (STORAGE_AWS_EXTERNAL_ID)，Snowflake 会生成一个 ID 供您使用。

   在下一部分中需要提供这些值。

第 5 步：授予访问桶对象的 IAM 用户权限¶

以下分步说明描述了如何在 AWS 管理控制台中配置 Snowflake 的 IAM 访问权限，以便您可以使用 S3 桶来加载和卸载数据：

1. 登录 AWS 管理控制台。

2. 选择 IAM。

3. 从左侧导航窗格中选择 Roles。

4. 选择在 第 2 步：在 AWS 中创建 IAM 角色 （本主题内容）中创建的角色。

5. 选择 Trust relationships 选项卡。

6. 选择 Edit trust policy。

7. 使用您在 第 4 步：检索 Snowflake 账户 AWS IAM 用户 （本主题内容）中记录的 DESC STORAGE INTEGRATION 输出值修改策略文档。

   IAM 角色的策略文档

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "",
         "Effect": "Allow",
         "Principal": {
           "AWS": "<snowflake_user_arn>"
         },
         "Action": "sts:AssumeRole",
         "Condition": {
           "StringEquals": {
             "sts:ExternalId": "<snowflake_external_id>"
           }
         }
       }
     ]
   }
   

   Copy

   其中：

   • snowflake_user_arn 是记录的 STORAGE_AWS_IAM_USER_ARN 值。

   • snowflake_external_id 是记录的 STORAGE_AWS_EXTERNAL_ID 值。

     在此示例中，snowflake_external_id 值为 MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=。

     备注

     出于安全原因，如果您在不指定外部 ID 的情况下创建新的存储集成（或使用 CREATE OR REPLACE STORAGE INTEGRATION 语法重新创建一个现有的存储集成），新的集成具有 不同 的外部 ID，所以它不能解析信任关系，除非更新信任策略。

8. 选择 Update policy 保存更改。

第 1 步：创建暂存区（如果需要）¶

使用 CREATE STAGE 命令创建引用您的 S3 桶的外部暂存区。Snowpipe 会从暂存区中获取数据文件，并在加载到目标表之前将它们临时加入队列。或者，您可以使用现有的外部暂存区。

以下示例会在活动架构中为用户会话创建一个名为 mystage 的暂存区。云存储 URL 包括路径 files。该暂存区引用名为 my_storage_int 的存储集成：

USE SCHEMA snowpipe_db.public;

CREATE STAGE mystage
  URL = 's3://mybucket/load/files'
  STORAGE_INTEGRATION = my_storage_int;

Copy

第 2 步：创建已启用自动引入的管道¶

使用 CREATE PIPE 命令创建管道。管道定义了 Snowpipe 用来将数据从引入队列加载到目标表中的 COPY INTO <table> 语句。

以下示例会在活动架构中为用户会话创建名为 mypipe 的管道。管道将数据从 mystage 暂存区中暂存的文件加载到 mytable 表：

create pipe snowpipe_db.public.mypipe auto_ingest=true as
  copy into snowpipe_db.public.mytable
  from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

Copy

AUTO_INGEST=true 参数指定，当新数据可供加载时，将从 S3 桶发送的事件通知读取到 SQS 队列。

第 3 步：配置安全性¶

对于将使用 Snowpipe 执行连续数据加载的每个用户，请授予针对数据加载对象（即目标数据库、架构和表；暂存区对象和管道）的足够访问控制权限。

要使用 Snowpipe，需要具有以下权限的角色：

使用 GRANT <privileges> 命令为角色授予权限。

例如，创建名为 snowpipe_role 的角色，该角色可以访问一组 snowpipe_db.public 数据库对象以及一个名为 mypipe 的管道，然后将该角色授予用户：

-- Create a role to contain the Snowpipe privileges
use role securityadmin;

create or replace role snowpipe_role;

-- Grant the required privileges on the database objects
grant usage on database snowpipe_db to role snowpipe_role;

grant usage on schema snowpipe_db.public to role snowpipe_role;

grant insert, select on snowpipe_db.public.mytable to role snowpipe_role;

grant usage on stage snowpipe_db.public.mystage to role snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = true;

-- Grant the OWNERSHIP privilege on the pipe object
grant ownership on pipe snowpipe_db.public.mypipe to role snowpipe_role;

-- Grant the role to a user
grant role snowpipe_role to user jsmith;

-- Set the role as the default role for the user
alter user jsmith set default_role = snowpipe_role;

-- Resume the pipe
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = false;

Copy

第 4 步：配置事件通知¶

为 S3 桶配置事件通知，以便在有新数据可供加载时通知 Snowpipe。自动引入功能依赖 SQS 队列将事件通知从 S3 传递到 Snowpipe。

为便于使用，Snowpipe SQS 队列由 Snowflake 创建和管理。SHOW PIPES 命令输出显示 SQS 队列的 Amazon 资源名称 (ARN)。

1. 执行 SHOW PIPES 命令：

   SHOW PIPES;
   

   Copy

   记住针对 notification_channel 列中暂存区的 SQS 队列的 ARN。将 ARN 复制到方便的位置。

   备注

   根据 AWS 准则，Snowflake 为每个 S3 桶指定的 SQS 队列不超过一个。SQS 队列可由同一 AWS 账户的同一区域中的多个桶共享。SQS 队列会协调所有管道的通知，将 S3 桶的外部暂存区连接到目标表。将数据文件上传到桶时，与暂存区目录路径匹配的所有管道都会将文件一次性加载到其相应的目标表中。

2. 登录到 Amazon S3 控制台。

3. 使用 Amazon S3 文档 (https://docs.aws.amazon.com/AmazonS3/latest/user-guide/enable-event-notifications.html) 中提供的说明为您的 S3 桶配置事件通知。填写如下字段：

   • Name：事件通知的名称（例如 Auto-ingest Snowflake）。

   • Events：选择 ObjectCreate (All) 选项。

   • Send to：从下拉列表中选择 SQS Queue。

   • SQS：从下拉列表中选择 Add SQS queue ARN。

   • SQS queue ARN：粘贴 SHOW PIPES 输出中的 SQS 队列名称。

现已配置附带自动引入功能的 Snowpipe！

当新数据文件添加到 S3 桶时，事件通知会告知 Snowpipe，将数据文件加载到管道中定义的目标表中。

第 5 步：加载历史文件¶

要加载在配置 SQS 通知 之前 外部暂存区中已存在的数据文件积压，请参阅 正在加载历史数据。

第 6 步：删除暂存文件¶

在成功加载数据并且不再需要这些文件后删除暂存文件。有关信息，请参阅 在 Snowpipe 加载数据后删除暂存文件。

先决条件：创建 Amazon SNS 主题和订阅¶

1. 在您的 SNS 账户中创建一个 AWS 主题来处理 S3 桶上 Snowflake 暂存区位置的所有消息。

2. 将 S3 事件通知的目标目的地（如其他 SQS 队列或 AWS Lambda 工作负载）订阅到此主题。SNS 将您的桶的事件通知发布给该主题的所有订阅者。

有关信息，请参阅 SNS 文档 (https://aws.amazon.com/documentation/sns/)。

第 1 步：将 Snowflake SQS 队列订阅至 SNS 主题¶

1. 登录 AWS 管理控制台。

2. 从主页仪表板中选择 Simple Notification Service (SNS)。

3. 从左侧导航窗格中选择 Topics。

4. 找到您的 S3 桶主题。注意主题 ARN。

5. 使用 Snowflake 客户端，用您的 SNS 主题 ARN 查询 SYSTEM$GET_AWS_SNS_IAM_POLICY 系统函数：

   select system$get_aws_sns_iam_policy('<sns_topic_arn>');
   

   Copy

   该函数返回一个授予 Snowflake SQS 队列权限以订阅 SNS 主题的 IAM 策略。

   例如：

   select system$get_aws_sns_iam_policy('arn:aws:sns:us-west-2:001234567890:s3_mybucket');
   
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | SYSTEM$GET_AWS_SNS_IAM_POLICY('ARN:AWS:SNS:US-WEST-2:001234567890:S3_MYBUCKET')                                                                                                                                                                   |
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | {"Version":"2012-10-17","Statement":[{"Sid":"1","Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"},"Action":["sns:Subscribe"],"Resource":["arn:aws:sns:us-west-2:001234567890:s3_mybucket"]}]}                 |
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   

   Copy

6. 返回 AWS 管理控制台。从左侧导航窗格中选择 Topics。

7. 选择您的 S3 桶的主题，然后点击 Edit 按钮。Edit 页面随即打开。

8. 点击 Access policy - Optional 以展开页面的此区域。

9. 将 SYSTEM$GET_AWS_SNS_IAM_POLICY 函数结果中添加的 IAM 策略合并到 JSON 文档中。

   例如：

   原始 IAM 策略（缩写）：

   {
     "Version":"2008-10-17",
     "Id":"__default_policy_ID",
     "Statement":[
        {
           "Sid":"__default_statement_ID",
           "Effect":"Allow",
           "Principal":{
              "AWS":"*"
           }
           ..
        }
      ]
    }
   

   Copy

   合并的 IAM 策略：

   {
     "Version":"2008-10-17",
     "Id":"__default_policy_ID",
     "Statement":[
        {
           "Sid":"__default_statement_ID",
           "Effect":"Allow",
           "Principal":{
              "AWS":"*"
           }
           ..
        },
        {
           "Sid":"1",
           "Effect":"Allow",
           "Principal":{
             "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
            },
            "Action":[
              "sns:Subscribe"
            ],
            "Resource":[
              "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
            ]
        }
      ]
    }
   

   Copy

10. 添加额外的策略授权以允许 S3 将桶的事件通知发布到 SNS 主题。

    例如（使用 SNS 主题 ARN 以及这些说明中使用的 S3 桶）：

    {
        "Sid":"s3-event-notifier",
        "Effect":"Allow",
        "Principal":{
           "Service":"s3.amazonaws.com"
        },
        "Action":"SNS:Publish",
        "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
        "Condition":{
           "ArnLike":{
              "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
           }
        }
     }
    

    Copy

    合并的 IAM 策略：

    {
      "Version":"2008-10-17",
      "Id":"__default_policy_ID",
      "Statement":[
         {
            "Sid":"__default_statement_ID",
            "Effect":"Allow",
            "Principal":{
               "AWS":"*"
            }
            ..
         },
         {
            "Sid":"1",
            "Effect":"Allow",
            "Principal":{
              "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
             },
             "Action":[
               "sns:Subscribe"
             ],
             "Resource":[
               "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
             ]
         },
         {
            "Sid":"s3-event-notifier",
            "Effect":"Allow",
            "Principal":{
               "Service":"s3.amazonaws.com"
            },
            "Action":"SNS:Publish",
            "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
            "Condition":{
               "ArnLike":{
                  "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
               }
            }
          }
       ]
     }
    

    Copy

11. 点击 Save changes 按钮。

第 2 步：创建暂存区（如果需要）¶

使用 CREATE STAGE 命令创建引用您的 S3 桶的外部暂存区。Snowpipe 会从暂存区中获取数据文件，并在加载到目标表之前将它们临时加入队列。

或者，您可以使用现有的外部暂存区。

以下示例会在活动架构中为用户会话创建一个名为 mystage 的暂存区。云存储 URL 包括路径 files。该暂存区引用名为 my_storage_int 的存储集成：

CREATE STAGE mystage
  URL = 's3://mybucket/load/files'
  STORAGE_INTEGRATION = my_storage_int;

Copy

第 3 步：创建已启用自动引入的管道¶

使用 CREATE PIPE 命令创建管道。管道定义了 Snowpipe 用来将数据从引入队列加载到目标表中的 COPY INTO <table> 语句。在 COPY 语句中，从 先决条件：创建 Amazon SNS 主题和订阅 中识别 SNS 主题 ARN。

以下示例会在活动架构中为用户会话创建名为 mypipe 的管道。管道将数据从 mystage 暂存区中暂存的文件加载到 mytable 表：

create pipe snowpipe_db.public.mypipe
  auto_ingest=true
  aws_sns_topic='<sns_topic_arn>'
  as
    copy into snowpipe_db.public.mytable
    from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

Copy

其中：

AUTO_INGEST = true

指定在新数据可供加载时将从 S3 桶发送的事件通知读取到 SQS 队列。

AWS_SNS_TOPIC = '<sns_topic_arn>'

为 S3 桶的 SNS 主题指定 ARN，例如，当前示例中为 arn:aws:sns:us-west-2:001234567890:s3_mybucket。CREATE PIPE 语句将 Snowflake SQS 队列订阅至指定的 SNS 主题。请注意，管道只会将文件复制到事件通知通过 SNS 主题触发的引入队列。

要从管道中移除任一参数，当前需要使用语法 CREATE OR REPLACE PIPE 重新创建管道。

第 4 步：配置安全性¶

对于将使用 Snowpipe 执行连续数据加载的每个用户，请授予针对数据加载对象（即目标数据库、架构和表；暂存区对象和管道）的足够访问控制权限。

要使用 Snowpipe，需要具有以下权限的角色：

使用 GRANT <privileges> 命令为角色授予权限。

例如，创建名为 snowpipe_role 的角色，该角色可以访问一组 snowpipe_db.public 数据库对象以及一个名为 mypipe 的管道，然后将该角色授予用户：

-- Create a role to contain the Snowpipe privileges
use role securityadmin;

create or replace role snowpipe_role;

-- Grant the required privileges on the database objects
grant usage on database snowpipe_db to role snowpipe_role;

grant usage on schema snowpipe_db.public to role snowpipe_role;

grant insert, select on snowpipe_db.public.mytable to role snowpipe_role;

grant usage, read on stage snowpipe_db.public.mystage to role snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = true;

-- Grant the OWNERSHIP privilege on the pipe object
grant ownership on pipe snowpipe_db.public.mypipe to role snowpipe_role;

-- Grant the role to a user
grant role snowpipe_role to user jsmith;

-- Set the role as the default role for the user
alter user jsmith set default_role = snowpipe_role;

-- Resume the pipe
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = false;

Copy

现已配置附带自动引入功能的 Snowpipe！

当新数据文件添加到 S3 桶时，事件通知会告知 Snowpipe，将数据文件加载到管道中定义的目标表中。

第 5 步：加载历史文件¶

要加载在配置 SQS 通知 之前 外部暂存区中已存在的数据文件积压，请参阅 正在加载历史数据。

第 6 步：删除暂存文件¶

在成功加载数据并且不再需要这些文件后删除暂存文件。有关信息，请参阅 在 Snowpipe 加载数据后删除暂存文件。

第 1 步：创建 Amazon SNS 主题¶

按照（本主题内容） 先决条件：创建 Amazon SNS 主题和订阅 操作。

第 2 步：创建 EventBridge 规则以订阅 S3 桶并向 SNS 主题发送通知¶

• 为 S3 桶 启用 Amazon EventBridge (https://docs.aws.amazon.com/AmazonS3/latest/userguide/enable-event-notifications-eventbridge.html)。

• 创建 EventBridge 规则，将通知发送 (https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-s3-object-created-tutorial.html) 至第 1 步中创建的 SNS 主题。

第 3 步：配置 Amazon SNS 以使用 SQS 通知自动执行 Snowpipe¶

按照 选项 2：配置 Amazon SNS 以使用 SQS 通知自动执行 Snowpipe （本主题内容）操作。