Enabling Snowpipe error notifications for Amazon SNS

This topic provides instructions for pushing Snowpipe error notifications to the Amazon Simple Notification Service (SNS) service. SNS is a publish/subscribe messaging service.

This feature can push error notifications for the following types of loads:

  • Auto-ingest Snowpipe.

  • Calls to the Snowpipe insertFiles REST API endpoint.

  • Loads from Apache Kafka using the Snowflake Connector for Kafka with the Snowpipe ingestion method only.

Cloud platform support

Currently, this feature is limited to Snowflake accounts hosted on Amazon Web Services (AWS). Snowpipe can load data from files in any supported cloud storage service; however, push notifications to SNS are only supported in Snowflake accounts hosted on AWS.

Notes

  • This feature is implemented using the notification integration object. A notification integration is a Snowflake object that provides an interface between Snowflake and third-party cloud message queuing services. A single notification integration can support multiple pipes.

  • Snowflake guarantees at-least-once message delivery of error notifications (i.e. multiple attempts are made to deliver messages to ensure at least one attempt succeeds, which can result in duplicate messages).

Enabling error notifications

Step 1: Creating an Amazon SNS Topic

Create an SNS topic in your AWS account to handle all error messages. Record the Amazon Resource Name (ARN) for the SNS topic.

Note

Only standard SNS topics are supported. Do not create SNS FIFO (first in, first out) topics for use with error notifications. Currently, error notifications sent to FIFO topics fail silently.

To reduce latency and avoid data egress charges for sending notifications across regions, we recommend creating the SNS topic in the same region as your Snowflake account.

For instructions, see the Creating an Amazon SNS topic in the SNS documentation.

Step 2: Creating the IAM Policy

Create an AWS Identity and Access Management (IAM) policy that grants permissions to publish to the SNS topic. The policy defines the following actions:

  • sns:publish

    Publish to the SNS topic.

  1. Log into the AWS Management Console.

  2. From the home dashboard, choose Identity & Access Management (IAM):

  3. Choose Account settings from the left-hand navigation pane.

  4. Expand the Security Token Service Regions list, find the AWS region corresponding to the region where your account is located, and choose Activate if the status is Inactive.

  5. Choose Policies from the left-hand navigation pane.

  6. Click Create Policy.

  7. Click the JSON tab.

  8. Add a policy document that defines actions that can be taken on your SNS topic.

    Copy and paste the following text into the policy editor:

    {
        "Version": "2012-10-17",
        "Statement": [
          {
             "Effect": "Allow",
             "Action": [
                 "sns:Publish"
             ],
             "Resource": "<sns_topic_arn>"
          }
        ]
     }
    
    Copy

    Replace sns_topic_arn with the ARN of the SNS topic you created in Step 1: Creating an Amazon SNS Topic (in this topic).

  9. Click Review policy.

  10. Enter the policy name (e.g. snowflake_sns_topic) and an optional description. Click Create policy.

Step 3: Creating the AWS IAM Role

Create an AWS IAM role on which to assign privileges on the SNS topic.

  1. Log into the AWS Management Console.

  2. From the home dashboard, choose Identity & Access Management (IAM):

  3. Choose Roles from the left-hand navigation pane.

  4. Click the Create role button.

  5. Select Another AWS account as the trusted entity type.

  6. In the Account ID field, enter your own AWS account ID temporarily.

  7. Select the Require external ID option. This option enables you to grant permissions on your Amazon account resources (i.e. SNS) to a third party (i.e. Snowflake).

    For now, enter a dummy ID such as 0000. Later, you will modify the trust relationship and replace the dummy ID with the external ID for the Snowflake IAM user generated for your account. A condition in the trust policy for your IAM role allows your Snowflake users to assume the role using the notification integration object you will create later.

  8. Click the Next button.

  9. Locate the policy you created in Step 2: Creating the IAM Policy (in this topic), and select this policy.

  10. Click the Next button.

  11. Enter a name and description for the role, and click the Create role button.

  12. Record the Role ARN value located on the role summary page. You will specify this value in one or more later steps.

Step 4: Creating the Notification Integration

Create a notification integration using CREATE NOTIFICATION INTEGRATION. An integration is a Snowflake object that references the SNS topic you created.

A single notification integration can support multiple tasks and pipes.

Note

Only account administrators (users with the ACCOUNTADMIN role) or a role with the global CREATE INTEGRATION privilege can execute this SQL command.

CREATE NOTIFICATION INTEGRATION <integration_name>
  ENABLED = true
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = AWS_SNS
  DIRECTION = OUTBOUND
  AWS_SNS_TOPIC_ARN = '<topic_arn>'
  AWS_SNS_ROLE_ARN = '<iam_role_arn>'
Copy

Where:

<integration_name>

Name of the new integration.

DIRECTION = OUTBOUND

Direction of the cloud messaging with respect to Snowflake. Required only when configuring error notifications.

<topic_arn>

SNS topic ARN you recorded in Step 1: Creating an Amazon SNS Topic (in this topic).

<iam_role_arn>

IAM role ARN you recorded in Step 3: Creating the AWS IAM Role (in this topic).

For example:

CREATE NOTIFICATION INTEGRATION my_notification_int
  ENABLED = true
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = AWS_SNS
  DIRECTION = OUTBOUND
  AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-east-2:111122223333:sns_topic'
  AWS_SNS_ROLE_ARN = 'arn:aws:iam::111122223333:role/error_sns_role';
Copy

Step 5: Granting Snowflake Access to the SNS Topic

Retrieve the IAM User ARN and SNS Topic External ID

  1. Execute DESCRIBE INTEGRATION:

    DESC NOTIFICATION INTEGRATION <integration_name>;
    
    Copy

    Where:

    • integration_name is the name of the notification integration you created in Step 4: Creating the Notification Integration (in this topic).

    For example:

    DESC NOTIFICATION INTEGRATION my_notification_int;
    
    +---------------------------+-------------------+------------------------------------------------------+----------------------+
    |   property                |   property_type   |   property_value                                     |   property_default   |
    +---------------------------+-------------------+------------------------------------------------------+----------------------+
    |   ENABLED                 |   Boolean         |   true                                               |   false              |
    |   NOTIFICATION_PROVIDER   |   String          |   AWS_SNS                                            |                      |
    |   DIRECTION               |   String          |   OUTBOUND                                           |   INBOUND            |
    |   AWS_SNS_TOPIC_ARN       |   String          |   arn:aws:sns:us-east-2:111122223333:myaccount       |                      |
    |   AWS_SNS_ROLE_ARN        |   String          |   arn:aws:iam::111122223333:role/myrole              |                      |
    |   SF_AWS_IAM_USER_ARN     |   String          |   arn:aws:iam::123456789001:user/c_myaccount         |                      |
    |   SF_AWS_EXTERNAL_ID      |   String          |   MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=   |                      |
    +---------------------------+-------------------+------------------------------------------------------+----------------------+
    
    Copy
  2. Record the following generated values:

    SF_AWS_IAM_USER_ARN:

    ARN for the Snowflake IAM user created for your account. Users in your Snowflake account will assume the IAM role you created in Step 3: Creating the AWS IAM Role by submitting the external ID for this user using your notification integration.

    SF_AWS_EXTERNAL_ID:

    External ID for the Snowflake IAM user created for your account.

    In the next step, you will update the trust relationship for the IAM role with these values.

Note the DIRECTION property, which indicates the direction of the cloud messaging with respect to Snowflake.

Modify the Trust Relationship in the IAM Role

  1. Log into the AWS Management Console.

  2. From the home dashboard, choose Identity & Access Management (IAM):

  3. Choose Roles from the left-hand navigation pane.

  4. Click on the role you created in Step 3: Creating the AWS IAM Role (in this topic).

  5. Click on the Trust relationships tab.

  6. Click the Edit trust relationship button.

  7. Modify the policy document with the DESC NOTIFICATION INTEGRATION output values you recorded in Retrieve the IAM User ARN and SNS Topic External ID (in this topic):

    Policy document for IAM role

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

    Where:

    • sf_aws_iam_user_arn is the SF_AWS_IAM_USER_ARN value you recorded.

    • sf_aws_external_id is the SF_AWS_EXTERNAL_ID value you recorded.

  8. Click the Update Trust Policy button. The changes are saved.

Step 6: Enabling error notifications in pipes

A single notification integration can be shared by multiple pipes. The body of error messages identifies the pipe, external stage and path, and file where the error originated, among other details.

To enable error notifications for a pipe, specify an ERROR_INTEGRATION parameter value.

Note

Creating or modifying a pipe that references a notification integration requires a role that has the USAGE privilege on the notification integration. In addition, the role must have either the CREATE PIPE privilege on the schema or the OWNERSHIP privilege on the pipe, respectively.

Note that operating on any object in a schema also requires the USAGE privilege on the parent database and schema.

For instructions on creating a custom role with a specified set of privileges, see Creating custom roles.

For general information about roles and privilege grants for performing SQL actions on securable objects, see Overview of Access Control.

New pipe

Create a new pipe using CREATE PIPE. Note that the configuring automated data loads (i.e. auto-ingest Snowpipe) requires additional pipe parameters. For instructions, see Automating continuous data loading using cloud messaging.

CREATE PIPE <name>
  [ AUTO_INGEST = TRUE | FALSE  ]
  ERROR_INTEGRATION = <integration_name>
  AS <copy_statement>
Copy

Where:

ERROR_INTEGRATION = <integration_name>

Name of the notification integration you created in Step 4: Creating the Notification Integration (in this topic).

The following example shows a CREATE PIPE statement that supports both error notifications and automated data loads:

CREATE PIPE mypipe
  AUTO_INGEST = TRUE
  ERROR_INTEGRATION = my_notification_int
  AS
  COPY INTO mydb.public.mytable
  FROM @mydb.public.mystage;
Copy

Existing pipe

Modify an existing pipe using ALTER PIPE:

ALTER PIPE <name> SET ERROR_INTEGRATION = <integration_name>;
Copy

Where <integration_name> is the name of the notification integration you created in Step 4: Creating the Notification Integration (in this topic).

For example:

ALTER PIPE mypipe SET ERROR_INTEGRATION = my_notification_int;
Copy

Error notification message payload

The body of error messages identifies the pipe and the errors encountered during a load.

The following is a sample message payload describing a Snowpipe error. The payload can include one or more error messages.

{\"version\":\"1.0\",\"messageId\":\"a62e34bc-6141-4e95-92d8-f04fe43b43f5\",\"messageType\":\"INGEST_FAILED_FILE\",\"timestamp\":\"2021-10-22T19:15:29.471Z\",\"accountName\":\"MYACCOUNT\",\"pipeName\":\"MYDB.MYSCHEMA.MYPIPE\",\"tableName\":\"MYDB.MYSCHEMA.MYTABLE\",\"stageLocation\":\"s3://mybucket/mypath\",\"messages\":[{\"fileName\":\"/file1.csv_0_0_0.csv.gz\",\"firstError\":\"Numeric value 'abc' is not recognized\"}]}
Copy

Note that you must parse the string into a JSON object to process values in the payload.