Triggering Snowpipe ingestion from an S3 bucket notification

Question

Pulumi · Accepted Answer

To trigger Snowpipe ingestion from an S3 bucket notification, we need to integrate AWS S3 with Snowflake. The overall process involves setting up an S3 bucket with the correct permissions and configuring Snowpipe to ingest data from this bucket based on notifications of new file uploads.

Here's the high-level process to achieve this with Pulumi and TypeScript:

1. Create an S3 bucket that will receive the files.
2. Set up a Snowflake storage integration to allow Snowflake to access the S3 bucket.
3. Create an S3 event notification to send a message to an AWS SNS topic when new files are uploaded to the S3 bucket.
4. Configure the Snowflake pipe to ingest data when new messages are received on the SNS topic.

Below is the Pulumi program written in TypeScript that carries out these steps:

```typescript
import * as aws from "@pulumi/aws";
import * as snowflake from "@pulumi/snowflake";
import * as pulumi from "@pulumi/pulumi";

// Step 1: Create an S3 bucket
const s3Bucket = new aws.s3.Bucket("dataBucket", {});

// Step 2: Set up Snowflake storage integration
const storageIntegration = new snowflake.StorageIntegration("storageIntegration", {
  // Ensure that the type and other properties are set according to what Snowflake expects.
  // For example, you may need to provide an IAM role and other settings depending on your exact scenario.
  // You would replace the `storageAwsRoleArn` with your actual ARN from the IAM role with the
  // necessary Snowflake permissions.
  storageProvider: "S3",
  storageAwsRoleArn: "arn:aws:iam::123456789012:role/MySnowflakeRole",
  storageAllowedLocations: [s3Bucket.bucket.apply(bucketName => `s3://${bucketName}`)], // dynamically apply the bucket name
});

// Step 3: Create an SNS Topic to trigger the Snowpipe
const snsTopic = new aws.sns.Topic("snowpipeTrigger", {});

// Create an S3 event notification to send a message to the AWS SNS topic
const s3BucketNotification = new aws.s3.BucketNotification("s3BucketNotification", {
  bucket: s3Bucket.id, // Reference to the S3 bucket created earlier.
  topics: [{
    topicArn: snsTopic.arn,
    events: ["s3:ObjectCreated:*"],
  }],
});

// Step 4: Create the Snowpipe
const pipe = new snowflake.Pipe("snowpipe", {
  // Supply database and schema names where the pipe should be located
  database: "MyDatabase",
  schema: "MySchema",
  copyStatement: 'COPY INTO my_target_table FROM @mystage',
  autoIngest: true,
  awsSnsTopicArn: snsTopic.arn, // Reference the ARN of the SNS topic created earlier.
});

// Export the bucket name, SNS topic ARN, and Snowpipe name
export const bucketName = s3Bucket.bucket;
export const topicArn = snsTopic.arn;
export const snowpipeName = pipe.name;
```

This program creates all the necessary resources and links them together. When a new file is uploaded to the `dataBucket` S3 bucket, an event will be sent to the `snowpipeTrigger` SNS topic. This topic is referenced in the `snowpipe` resource which listens for these SNS events to trigger the ingestion of data into Snowflake.

### Important Considerations:
- Replace `"arn:aws:iam::123456789012:role/MySnowflakeRole"` with the ARN of the IAM role that Snowflake should assume when accessing your S3 bucket.
- Replace `'COPY INTO my_target_table FROM @mystage'` with the actual Snowflake SQL COPY statement configured for your use case.
- Ensure that the IAM role associated with the Snowflake Storage integration has the necessary permissions to access the specified S3 bucket.

After setting this up, you should have a working pipeline that triggers Snowpipe to ingest new data as it is uploaded to your S3 bucket. Make sure to fill in the placeholders with your specific details, such as the database and schema names, as well as properly configuring the Storage Integration with the correct IAM role.