Snowpipe for event-driven loading in Snowflake - Commands & Configuration

Introduction

Loading data into Snowflake automatically when new files arrive can save time and reduce errors. Snowpipe helps by watching for new files and loading them right away without manual steps.

When you want new data files to be loaded into Snowflake as soon as they arrive in cloud storage.

When you have frequent small data files arriving and want to avoid batch loading delays.

When you want to automate data ingestion without writing custom scripts or manual commands.

When you want to reduce the time between data arrival and availability for queries.

When you want to integrate Snowflake with cloud storage event notifications for seamless loading.

Config File - pipe_creation.sql

pipe_creation.sql

CREATE OR REPLACE PIPE my_event_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO my_table
  FROM @my_stage
  FILE_FORMAT = (TYPE = 'CSV' FIELD_DELIMITER = ',' SKIP_HEADER = 1);

-- This pipe listens for new files in the stage and loads them automatically.

This SQL file creates a Snowpipe named my_event_pipe that automatically loads CSV files from the stage my_stage into the table my_table.

AUTO_INGEST = TRUE enables event-driven loading using cloud storage notifications.

The COPY INTO command defines how files are loaded, including file format details.

Commands

This command creates or replaces the Snowpipe named 'my_event_pipe' that automatically loads CSV files from the stage 'my_stage' into 'my_table'.

Terminal

snowsql -q "CREATE OR REPLACE PIPE my_event_pipe AUTO_INGEST = TRUE AS COPY INTO my_table FROM @my_stage FILE_FORMAT = (TYPE = 'CSV' FIELD_DELIMITER = ',' SKIP_HEADER = 1);"

Expected Output

Done. Snowpipe MY_EVENT_PIPE successfully created.

This command refreshes the pipe to ensure it is ready to receive event notifications and start loading new files.

Terminal

snowsql -q "ALTER PIPE my_event_pipe REFRESH;"

Expected Output

Done. Pipe MY_EVENT_PIPE refreshed.

This command checks the load history for 'my_table' to verify that Snowpipe has loaded new files in the last hour.

Terminal

snowsql -q "SELECT * FROM TABLE(INFORMATION_SCHEMA.LOAD_HISTORY(TABLE_NAME => 'MY_TABLE', START_TIME => DATEADD('hour', -1, CURRENT_TIMESTAMP())));"

Expected Output

LOAD_HISTORY_ID | FILE_NAME       | STATUS  | ROWS_LOADED | LOAD_TIME

123456789       | datafile1.csv   | LOADED  | 100         | 2024-06-01 12:00:00

123456790       | datafile2.csv   | LOADED  | 150         | 2024-06-01 12:05:00

Key Concept

If you remember nothing else from this pattern, remember: Snowpipe automatically loads new files into Snowflake as soon as they arrive, using cloud event notifications.

Common Mistakes

Not enabling AUTO_INGEST when creating the pipe.

Without AUTO_INGEST, Snowpipe will not listen for new file events and will not load files automatically.

Always set AUTO_INGEST = TRUE in the pipe creation SQL to enable event-driven loading.

Not configuring cloud storage event notifications to Snowflake.

Snowpipe relies on cloud storage events to trigger loading; without these notifications, files won't load automatically.

Set up cloud storage event notifications (e.g., AWS S3 event notifications) to notify Snowflake when new files arrive.

Using incorrect file format settings in the COPY INTO command.

If the file format does not match the actual files, loading will fail or produce incorrect data.

Define the correct file format options matching your data files in the COPY INTO command inside the pipe.

Summary

Create a Snowpipe with AUTO_INGEST = TRUE to enable automatic loading of new files.

Refresh the pipe to activate it and prepare for event-driven loading.

Check load history to verify that files are being loaded automatically by Snowpipe.