Snowpipe & Snowpipe Streaming: Automated Data Ingestion in Snowflake

In modern data pipelines, waiting for batch loads is a bottleneck. Snowflake's Snowpipe and Snowpipe Streaming solve this by automating continuous data ingestion. Snowpipe automatically loads new files from cloud storage (AWS S3, Azure Blob, GCS) as soon as they appear, using event notifications. Snowpipe Streaming goes further, allowing you to stream rows directly via a REST API, achieving sub-second latency. This tutorial covers both approaches, from setup to production debugging, with real-world examples. You'll learn how to create external stages, define pipes, monitor ingestion, and handle common pitfalls. By the end, you'll be able to build a robust, near-real-time data ingestion pipeline into Snowflake.

What is Snowpipe?

Snowpipe is a fully managed service that automates the loading of data from cloud storage into Snowflake tables. It uses the COPY command internally but is triggered automatically when new files arrive. Snowpipe supports AWS S3, Azure Blob Storage, and Google Cloud Storage. It can be configured with event notifications (SQS, Event Grid, Pub/Sub) to detect new files, or you can call the REST API to trigger a load. Snowpipe is ideal for near-real-time ingestion with latencies typically under a minute.

-- Create an external stage pointing to S3
CREATE OR REPLACE STAGE my_s3_stage
  URL = 's3://my-bucket/data/'
  CREDENTIALS = (AWS_KEY_ID = '...' AWS_SECRET_KEY = '...');

-- Define file format
CREATE OR REPLACE FILE FORMAT my_csv_format
  TYPE = CSV
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  SKIP_HEADER = 1;

-- Create the pipe
CREATE OR REPLACE PIPE my_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO my_table
  FROM @my_s3_stage
  FILE_FORMAT = my_csv_format;

-- Check pipe status
SELECT SYSTEM$PIPE_STATUS('my_pipe');

Setting Up Auto-Ingest with S3 Event Notifications

To enable auto-ingest, you need to configure an S3 bucket to send event notifications to an SQS queue, and then grant Snowflake access to that queue. Steps: 1) Create an SQS queue in AWS. 2) Configure S3 bucket to send events (s3:ObjectCreated:*) to the queue. 3) Create an IAM role that Snowflake can assume to read from the queue and the bucket. 4) In Snowflake, create a storage integration and pipe with AUTO_INGEST = TRUE. The pipe will automatically start loading files as they arrive. Below is a simplified example of the SQL to create the storage integration and pipe.

-- Create storage integration (requires AWS IAM role ARN)
CREATE OR REPLACE STORAGE INTEGRATION my_s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = S3
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/my-snowflake-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://my-bucket/data/');

-- Create stage using integration
CREATE OR REPLACE STAGE my_stage
  URL = 's3://my-bucket/data/'
  STORAGE_INTEGRATION = my_s3_int
  FILE_FORMAT = my_csv_format;

-- Create pipe with auto-ingest
CREATE OR REPLACE PIPE my_pipe
  AUTO_INGEST = TRUE
  AWS_SNS_TOPIC = 'arn:aws:sns:us-east-1:123456789012:snowpipe-topic'
  AS
  COPY INTO my_table
  FROM @my_stage;

Snowpipe Streaming: Real-Time Row Ingestion

Snowpipe Streaming is a newer feature that allows you to insert rows into Snowflake tables with sub-second latency using a REST API. Instead of waiting for files, you send rows directly via HTTP requests. This is ideal for streaming data from applications, IoT devices, or change data capture (CDC) systems. Snowpipe Streaming uses channels (similar to Kafka partitions) to ensure ordering and deduplication. Each channel is a sequence of rows that are committed in order. You can create multiple channels for parallelism. The client SDK (Java, Python) handles batching and retries. Below is an example using the Python SDK.

from snowflake.ingest import SimpleIngestManager
from snowflake.ingest import StagedFile
from snowflake.ingest.utils.urllib import urllib

# Initialize ingest manager
manager = SimpleIngestManager(
    account='myaccount',
    user='myuser',
    password='mypassword',
    pipe='mydb.myschema.mypipe',
    host='myaccount.snowflakecomputing.com'
)

# Create a staged file reference
staged_file = StagedFile('s3://my-bucket/data/file.csv', None)

# Ingest the file
manager.ingest([staged_file])
print('File ingested successfully')

Monitoring and Managing Snowpipe

Snowflake provides several views and functions to monitor Snowpipe activity. Key views: INFORMATION_SCHEMA.COPY_HISTORY shows load history per table; SNOWPIPE_STATUS function gives current pipe state; LOAD_HISTORY shows file-level load details. You can also use ACCOUNT_USAGE views for historical data. To manage pipes, you can pause, resume, drop, or refresh them. Refreshing a pipe re-ingests files from the stage that have not been loaded yet. This is useful for recovery. Below are common monitoring queries.

-- Check pipe status
SELECT SYSTEM$PIPE_STATUS('mydb.myschema.mypipe');

-- View recent copy history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
    TABLE_NAME=>'mytable',
    START_TIME=>DATEADD(hours, -24, CURRENT_TIMESTAMP())))
ORDER BY LAST_LOAD_TIME DESC;

-- View load history for a pipe
SELECT *
FROM TABLE(INFORMATION_SCHEMA.LOAD_HISTORY(
    PIPE_NAME=>'mydb.myschema.mypipe',
    START_TIME=>DATEADD(hours, -24, CURRENT_TIMESTAMP())));

-- Pause a pipe
ALTER PIPE mypipe SET PIPE_EXECUTION_PAUSED = TRUE;

-- Resume a pipe
ALTER PIPE mypipe SET PIPE_EXECUTION_PAUSED = FALSE;

-- Refresh a pipe (re-ingest missed files)
ALTER PIPE mypipe REFRESH;

Error Handling and Data Validation

Snowpipe can encounter errors like malformed files, schema mismatches, or permission issues. By default, Snowpipe skips files with errors and continues. You can configure error handling using the ON_ERROR option in the COPY command within the pipe definition. Options: CONTINUE (default), SKIP_FILE, SKIP_FILE_% (skip a percentage), ABORT_STATEMENT. For Snowpipe Streaming, errors are returned in the API response; you should implement retry logic. Use VALIDATE function to test files before loading. Below is an example of a pipe with error handling.

-- Create pipe with error handling
CREATE OR REPLACE PIPE my_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO my_table
  FROM @my_stage
  FILE_FORMAT = my_csv_format
  ON_ERROR = 'SKIP_FILE';

-- Validate a file before loading
SELECT *
FROM TABLE(
    VALIDATE(my_table, JOB_ID => '_last')
);

-- View load errors
SELECT *
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
    TABLE_NAME=>'mytable',
    START_TIME=>DATEADD(hours, -1, CURRENT_TIMESTAMP())))
WHERE STATUS = 'LOAD_FAILED';

Performance Optimization and Best Practices

To optimize Snowpipe performance: 1) Use compressed files (gzip, snappy) to reduce transfer time. 2) Partition files by date or other keys to avoid large scans. 3) Set appropriate file sizes (100-250 MB compressed) to balance parallelism. 4) Use multiple pipes for different tables or partitions. 5) For Snowpipe Streaming, batch rows in the client (e.g., 1000 rows per request) and use multiple channels. 6) Monitor the Snowpipe ingestion queue length; if it grows, increase the number of pipes or reduce file size. Below is an example of a pipe with optimized settings.

-- Create a pipe with optimized settings
CREATE OR REPLACE PIPE my_optimized_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO my_table
  FROM (
    SELECT $1, $2, $3, METADATA$FILENAME, METADATA$FILE_ROW_NUMBER
    FROM @my_stage
  )
  FILE_FORMAT = (TYPE = CSV COMPRESSION = GZIP)
  ON_ERROR = CONTINUE;

-- Use clustering on the table for better query performance
ALTER TABLE my_table CLUSTER BY (ingestion_date);