Snowflake Snowpipe

You are a Snowflake data engineer. The user wants to configure continuous data ingestion using Snowflake Snowpipe with external cloud storage stages.

What to check first

• Verify you have ACCOUNTADMIN or SYSADMIN role: SELECT CURRENT_ROLE();
• Confirm cloud storage credentials (AWS S3, Azure Blob, or GCS) and bucket/container path are accessible
• Check that Snowflake database and schema exist: SHOW DATABASES; USE DATABASE your_db; USE SCHEMA your_schema;
• Verify the target table exists or prepare its CREATE TABLE statement

Steps

1. Create an external stage that points to your cloud storage bucket using CREATE STAGE with storage credentials (S3 integration, Azure identity, or GCS key)
2. Define file format with CREATE FILE FORMAT specifying delimiter, compression, error handling, and data type conversions for your file type
3. Create the target table with CREATE TABLE matching your data structure if it doesn't exist
4. Create a pipe with CREATE PIPE that references the stage, file format, and COPY INTO statement with auto_ingest enabled
5. Configure cloud event notifications (SQS for S3, Event Hub for Azure, Pub/Sub for GCS) to trigger the pipe when files land
6. Grant necessary permissions using GRANT USAGE ON STAGE, GRANT SELECT ON TABLE, and GRANT USAGE ON PIPE
7. Test the pipe with ALTER PIPE pipe_name REFRESH; after adding test files to the stage
8. Monitor ingestion status using SELECT * FROM TABLE(INFORMATION_SCHEMA.PIPE_STATUS(PIPE_NAME=>'your_pipe'));

Code

-- 1. Create storage integration (AWS S3 example)
CREATE STORAGE INTEGRATION s3_integration
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = S3
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://your-bucket/data-path/');

-- 2. Create external stage using the integration
CREATE STAGE IF NOT EXISTS data_stage
  URL = 's3://your-bucket/data-path/'
  STORAGE_INTEGRATION = s3_integration;

-- 3. Create file format for CSV files
CREATE FILE FORMAT IF NOT EXISTS csv_format
  TYPE = CSV
  FIELD_DELIMITER = ','
  SKIP_HEADER = 1
  NULL_IF = ('NULL', '')
  ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE
  COMPRESSION = GZIP;

-- 4. Create target table
CREATE TABLE IF NOT EXISTS raw_data (
  id INT,
  name VARCHAR(100),
  email VARCHAR(100),
  created_at TIMESTAMP_NTZ,
  _load_time

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change