Snowpipe: Automated Data Ingestion & Streaming
Architecture Diagram 1: Snowpipe Ingestion Pipeline
┌─────────────────────────────────────────────────────────────────────────────┐
│ SNOWPIPE INGESTION ARCHITECTURE │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ DATA SOURCES │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │
│ │ │ Cloud │ │ IoT │ │ Kafka │ │ Application │ │ │
│ │ │ Storage │ │ Devices │ │ Streams │ │ Events │ │ │
│ │ │ (S3/ │ │ Sensors │ │ Topics │ │ (Webhooks) │ │ │
│ │ │ Azure/ │ │ │ │ │ │ │ │ │
│ │ │ GCS) │ │ │ │ │ │ │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ NOTIFICATION LAYER │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ S3 Event Notification / Azure Event Grid / GCS Triggers │ │ │
│ │ │ │ │ │
│ │ │ Event Types: │ │ │
│ │ │ • ObjectCreated • ObjectRemoved • ObjectTagging │ │ │
│ │ │ │ │ │
│ │ │ ┌────────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ SQS Queue / Event Hub / Pub/Sub │ │ │ │
│ │ │ │ (Message buffering and delivery guarantee) │ │ │ │
│ │ │ └────────────────────────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ SNOWPIPE SERVICE (Serverless) │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │ │ │
│ │ │ │ File │ │ Data │ │ Load │ │ │ │
│ │ │ │ Format │ │ Validation │ │ Optimization │ │ │ │
│ │ │ │ Detection │ │ │ │ │ │ │ │
│ │ │ └─────────────┘ └─────────────┘ └─────────────────────┘ │ │ │
│ │ │ │ │ │
│ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ COPY INTO Engine (Shared across all Snowpipe) │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐│ │ │ │
│ │ │ │ │ Parser │ │Transform│ │Compress │ │Write ││ │ │ │
│ │ │ │ │ │ │ │ │ │ │Micro- ││ │ │ │
│ │ │ │ │ │ │ │ │ │ │Partition││ │ │ │
│ │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘│ │ │ │
│ │ │ └──────────────────────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ MONITORING & HISTORY │ │ │
│ │ │ │ │ │
│ │ │ COPY_HISTORY │ LOAD_HISTORY │ PIPE_USAGE_HISTORY │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ TARGET TABLES │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌──────────┐ │ │
│ │ │ Table 1 │ │ Table 2 │ │ Table 3 │ │ Table N │ │ │
│ │ │ (Raw) │ │ (Staged) │ │ (Curated) │ │ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │
│ │ │ Micro-parts │ │ Micro-parts │ │ Micro-parts │ │ │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Architecture Diagram 2: File Format Processing Pipeline
┌─────────────────────────────────────────────────────────────────────────────┐
│ FILE FORMAT PROCESSING PIPELINE │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ INPUT FILE: sales_2024_01_15.csv │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Raw CSV Content: │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ id,name,amount,region,date │ │ │
│ │ │ 1,John Smith,1500.50,US,2024-01-15 │ │ │
│ │ │ 2,Jane Doe,2300.75,EU,2024-01-15 │ │ │
│ │ │ 3,Bob Johnson,890.25,ASIA,2024-01-15 │ │ │
│ │ │ ... │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ │ File Size: 500 MB (uncompressed) │ │
│ │ Rows: ~2,500,000 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ STAGE 1: FORMAT DETECTION & VALIDATION │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ File Format Configuration: │ │ │
│ │ │ │ │ │
│ │ │ TYPE = CSV │ │ │
│ │ │ FIELD_DELIMITER = ',' │ │ │
│ │ │ SKIP_HEADER = 1 │ │ │
│ │ │ NULL_IF = ('', 'NULL', 'null') │ │ │
│ │ │ DATE_FORMAT = 'YYYY-MM-DD' │ │ │
│ │ │ TIMESTAMP_FORMAT = 'YYYY-MM-DD HH24:MI:SS' │ │ │
│ │ │ ENCODING = 'UTF-8' │ │ │
│ │ │ │ │ │
│ │ │ Validation Checks: │ │ │
│ │ │ ✓ Header row exists and matches expected columns │ │ │
│ │ │ ✓ Data types match column definitions │ │ │
│ │ │ ✓ No malformed rows │ │ │
│ │ │ ✓ Encoding is valid │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ STAGE 2: DATA TRANSFORMATION │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ Transformations Applied: │ │ │
│ │ │ │ │ │
│ │ │ 1. Type Casting: │ │ │
│ │ │ '1500.50' → DECIMAL(10,2) │ │ │
│ │ │ '2024-01-15' → DATE │ │ │
│ │ │ │ │ │
│ │ │ 2. Null Handling: │ │ │
│ │ │ '' → NULL │ │ │
│ │ │ 'NULL' → NULL │ │ │
│ │ │ │ │ │
│ │ │ 3. Column Mapping: │ │ │
│ │ │ CSV Column 1 → Table Column 'ID' │ │ │
│ │ │ CSV Column 2 → Table Column 'NAME' │ │ │
│ │ │ CSV Column 3 → Table Column 'AMOUNT' │ │ │
│ │ │ CSV Column 4 → Table Column 'REGION' │ │ │
│ │ │ CSV Column 5 → Table Column 'TRANSACTION_DATE' │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ STAGE 3: MICRO-PARTITION CREATION │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ Output Micro-Partitions: │ │ │
│ │ │ │ │ │
│ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌──────┐ │ │ │
│ │ │ │ MP_001 │ │ MP_002 │ │ MP_003 │ │ MP_004 │ │ .. │ │ │ │
│ │ │ │ 125 MB │ │ 128 MB │ │ 130 MB │ │ 117 MB │ │ │ │ │ │
│ │ │ │ 625K │ │ 640K │ │ 650K │ │ 585K │ │ │ │ │ │
│ │ │ │ rows │ │ rows │ │ rows │ │ rows │ │ │ │ │ │
│ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ └──────┘ │ │ │
│ │ │ │ │ │
│ │ │ Total: 125 MB compressed (from 500 MB) = 4:1 ratio │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ STAGE 4: LOADED FILE TRACKING │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ COPY_HISTORY: │ │ │
│ │ │ ┌──────────────────┬────────────┬──────────┬─────────────┐ │ │ │
│ │ │ │ File Name │ Rows Loaded│ Errors │ Status │ │ │ │
│ │ │ ├──────────────────┼────────────┼──────────┼─────────────┤ │ │ │
│ │ │ │ sales_2024_01_15│ 2,500,000 │ 0 │ LOADED │ │ │ │
│ │ │ │ .csv │ │ │ │ │ │ │
│ │ │ └──────────────────┴────────────┴──────────┴─────────────┘ │ │ │
│ │ │ │ │ │
│ │ │ This file will NOT be reloaded (idempotent ingestion) │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Architecture Diagram 3: Auto-Ingest Configuration Flow
┌─────────────────────────────────────────────────────────────────────────────┐
│ AUTO-INGEST CONFIGURATION FLOW │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ STEP 1: CREATE STORAGE INTEGRATION │
│ ───────────────────────────────────── │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Snowflake ──────────────────────────────────────────────────▶ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ Storage Integration: s3_ingestion │ │ │
│ │ │ │ │ │
│ │ │ Type: EXTERNAL_STAGE │ │ │
│ │ │ Provider: AWS │ │ │
│ │ │ Storage AWS Role: arn:aws:iam::123:role/snowflake_role │ │ │
│ │ │ Enabled: TRUE │ │ │
│ │ │ │ │ │
│ │ │ ┌────────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ Snowflake generates IAM policy: │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ { │ │ │ │
│ │ │ │ "Effect": "Allow", │ │ │ │
│ │ │ │ "Action": [ │ │ │ │
│ │ │ │ "s3:GetObject", │ │ │ │
│ │ │ │ "s3:GetObjectVersion" │ │ │ │
│ │ │ │ ], │ │ │ │
│ │ │ │ "Resource": "arn:aws:s3:::bucket/*" │ │ │ │
│ │ │ │ } │ │ │ │
│ │ │ └────────────────────────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ STEP 2: CREATE PIPE │
│ ───────────────────── │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ Pipe: sales_pipe │ │ │
│ │ │ │ │ │
│ │ │ CREATE PIPE sales_pipe │ │ │
│ │ │ AUTO_INGEST = TRUE │ │ │
│ │ │ INTEGRATION = 's3_ingestion' │ │ │
│ │ │ AS │ │ │
│ │ │ COPY INTO sales_raw │ │ │
│ │ │ FROM '@s3_ingestion/data/' │ │ │
│ │ │ FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1) │ │ │
│ │ │ │ │ │
│ │ │ ┌────────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ Pipe returns notification channel: │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ SQS Queue: arn:aws:sqs:us-east-1:123456:myQueue │ │ │ │
│ │ │ │ (Configure in S3 bucket event notification) │ │ │ │
│ │ │ └────────────────────────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ STEP 3: CONFIGURE S3 EVENT NOTIFICATION │
│ ────────────────────────────────────────── │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ S3 Bucket Configuration: │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ Bucket: data-lake-sales │ │ │
│ │ │ │ │ │
│ │ │ Event Notification: │ │ │
│ │ │ ┌────────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ { │ │ │ │
│ │ │ │ "LambdaFunctionConfigurations": [ │ │ │ │
│ │ │ │ { │ │ │ │
│ │ │ │ "Event": "s3:ObjectCreated:*", │ │ │ │
│ │ │ │ "Prefix": "data/", │ │ │ │
│ │ │ │ "Suffix": ".csv", │ │ │ │
│ │ │ │ "QueueConfigurations": [ │ │ │ │
│ │ │ │ { │ │ │ │
│ │ │ │ "QueueArn": "arn:aws:sqs:...:snowflake_queue"│ │ │ │
│ │ │ │ } │ │ │ │
│ │ │ │ ] │ │ │ │
│ │ │ │ } │ │ │ │
│ │ │ │ ] │ │ │ │
│ │ │ │ } │ │ │ │
│ │ │ └────────────────────────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ STEP 4: INGESTION FLOW │
│ ────────────────────── │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ File Upload → S3 Event → SQS Queue → Snowpipe → COPY INTO │ │
│ │ │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌──────┐ │ │
│ │ │ Upload │───▶│ Event │───▶│ Queue │───▶│Detect │───▶│ Load │ │ │
│ │ │ File │ │ Fire │ │ Buffer │ │ New │ │ Data │ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘ └──────┘ │ │
│ │ │ │
│ │ Typical latency: 1-3 minutes end-to-end │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Snowpipe is Snowflake's serverless, continuous data ingestion service that automatically loads data from cloud storage into tables. It operates on a micro-batch model (1–3 minute latency), handling format detection, validation, transformation, compression, and micro-partition creation without manual orchestration.
Idempotent ingestion ensures loading the same file multiple times produces identical results. Snowpipe tracks loaded files in COPY_HISTORY and automatically excludes duplicates, enabling safe retries and data consistency even with duplicate events.
Snowpipe charges per file processed (serverless pricing). Use Parquet for analytics workloads (built-in compression, schema info, predicate pushdown). CSV is simpler but lacks these benefits. JSON supports semi-structured data with parsing overhead.
- Create Storage Integration: Define IAM roles for Snowflake access to cloud storage
- Create Pipe: Specify ingestion logic (COPY INTO statement, file format, target table)
- Configure Event Notifications: Set up S3/Azure/GCS triggers to route file events to Snowpipe
- Test Ingestion: Upload sample files and verify data loads correctly
- Monitor: Check PIPE_STATUS for pending files, COPY_HISTORY for load results
- Alert: Set up alerts for pipe lag (pending files > 100) and error rates (> 1%)
- Serverless: No warehouse management — Snowflake handles compute
- Latency: 1–3 minutes end-to-end (file upload → queryable)
- Idempotent: Duplicate files are automatically excluded
- File sizing: Target 100–250 MB compressed for optimal parallelism
- Monitoring: Use PIPE_STATUS, COPY_HISTORY, and PIPE_ERR views
Detailed Explanation
Snowpipe Fundamentals
Snowpipe is Snowflake's serverless, continuous data ingestion service that automates the loading of data from cloud storage into Snowflake tables. Unlike bulk COPY operations that require explicit execution, Snowpipe operates continuously, monitoring designated storage locations for new files and loading them automatically as they arrive. This architecture eliminates the need for manual intervention or custom ETL orchestration for simple data loading scenarios.
The service operates on a micro-batch processing model, typically detecting and loading new files within 1-3 minutes of their arrival in cloud storage. This near-real-time capability makes Snowpipe suitable for use cases requiring low-latency data availability without the complexity of streaming architectures. Snowpipe handles all aspects of the ingestion process, including file format detection, data validation, transformation, compression, and micro-partition creation.
Auto-Ingest Architecture
Auto-ingest is the recommended approach for production Snowpipe implementations. It leverages cloud-native event notification systems (S3 Event Notifications, Azure Event Grid, GCS Triggers) to detect new files and automatically trigger the ingestion process. This event-driven architecture provides several advantages over polling-based approaches, including lower latency, reduced cloud storage API costs, and elimination of polling overhead.
The auto-ingest configuration requires three components: a storage integration that defines Snowflake's access permissions to cloud storage, a pipe that specifies the ingestion logic and target table, and cloud-side event notifications that route file creation events to Snowpipe. The storage integration uses cloud IAM roles to grant Snowflake read access to specific storage locations without exposing credentials.
File Format Optimization
Snowpipe supports multiple file formats, each optimized for different data characteristics. CSV files are suitable for simple, flat data structures but lack built-in schema information and compression. JSON files support semi-structured data with nested structures but incur parsing overhead. Parquet and ORC formats provide columnar storage with built-in compression, schema evolution, and predicate pushdown capabilities, making them ideal for analytics workloads.
File size optimization is critical for Snowpipe performance. Snowflake recommends file sizes between 100MB and 250MB compressed. Smaller files create overhead from processing many micro-partitions, while larger files can cause timeout issues and reduce parallelism. The optimal file size depends on data complexity, compression ratio, and ingestion velocity requirements.
Error Handling and Idempotency
Snowpipe implements comprehensive error handling with automatic retry logic for transient failures. When a file fails to load due to network issues, format errors, or constraint violations, Snowpipe retries the operation with exponential backoff. Persistent errors are logged to the PIPE_ERR table, providing detailed error messages and file information for troubleshooting.
A key design principle of Snowpipe is idempotency—loading the same file multiple times produces the same result as loading it once. Snowpipe tracks all successfully loaded files in the COPY_HISTORY table and automatically excludes them from subsequent loads. This behavior enables safe retries and ensures data consistency even in the presence of duplicate events.
Monitoring and Alerting
Snowpipe provides comprehensive monitoring through several system views and tables. The COPY_HISTORY table tracks all completed load operations with detailed metrics including rows loaded, bytes processed, and error counts. The PIPE_STATUS function provides real-time information about pipe health, including pending file counts and last error details. The LOAD_HISTORY table provides historical information about file loading patterns.
Alerting can be configured using Snowflake's alerting features or external monitoring tools. Common alerts include pipe lag (pending files growing), error rates exceeding thresholds, and ingestion latency exceeding SLA requirements. Integration with cloud monitoring services (CloudWatch, Azure Monitor, Stackdriver) enables centralized alerting across the entire data pipeline.
Key Concepts Table
| Component | Purpose | Configuration | Cost Model |
|---|---|---|---|
| Storage Integration | Cloud access permissions | IAM roles, external stages | Free |
| Pipe | Ingestion logic definition | SQL COPY statement | Free (compute charged separately) |
| Auto-Ingest | Event-driven loading | Cloud event notifications | Serverless (per file processed) |
| COPY_HISTORY | Load operation tracking | Automatic | Free |
| PIPE_ERR | Error logging | Automatic | Free |
| File Format | Compression | Schema Support | Best For |
|---|---|---|---|
| CSV | External | None | Simple flat data |
| JSON | External | Full nesting | Semi-structured data |
| Parquet | Built-in (Snappy/ZSTD) | Full schema | Analytics workloads |
| ORC | Built-in (Zlib/ZSTD) | Full schema | Hive ecosystem |
| Avro | Built-in (Deflate) | Full schema | Schema evolution |
| Metric | Target | Warning | Critical |
|---|---|---|---|
| Ingestion Latency | < 3 min | 3-5 min | > 5 min |
| File Size | 100-250 MB | 50-100 MB | < 50 MB or > 500 MB |
| Error Rate | < 0.1% | 0.1-1% | > 1% |
| Pipe Lag | < 10 files | 10-100 files | > 100 files |
| Daily Throughput | Per SLA | 80-100% of SLA | > 100% of SLA |
Code Examples
-- Example 1: Create storage integration for S3
CREATE STORAGE INTEGRATION s3_sales_integration
TYPE = EXTERNAL_STAGE
ENABLED = TRUE
STORAGE_PROVIDER = S3
STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/SnowflakeAccessRole'
STORAGE_ALLOWED_LOCATIONS = ('s3://data-lake-sales/data/', 's3://data-lake-sales/archive/');
-- Example 2: Create pipe with auto-ingest
CREATE PIPE sales_auto_ingest
AUTO_INGEST = TRUE
INTEGRATION = 's3_sales_integration'
ERROR_INTEGRATION = 's3_error_integration'
COMMENT = 'Auto-ingest sales data from S3'
AS
COPY INTO sales_raw
FROM '@s3_sales_integration/data/'
FILE_FORMAT = (
TYPE = CSV
SKIP_HEADER = 1
NULL_IF = ('', 'NULL', 'null')
FIELD_OPTIONALLY_ENCLOSED_BY = '"'
TRIM_SPACE = TRUE
ERROR_ON_COLUMN_COUNT_MISMATCH = TRUE
)
ON_ERROR = 'CONTINUE';
-- Example 3: Create pipe for JSON data
CREATE PIPE json_events_pipe
AUTO_INGEST = TRUE
INTEGRATION = 's3_events_integration'
AS
COPY INTO events_raw
FROM '@s3_events_integration/events/'
FILE_FORMAT = (
TYPE = JSON
STRIP_OUTER_ARRAY = TRUE
IGNORE_UTF8_ERRORS = TRUE
)
ON_ERROR = 'SKIP_FILE';
-- Example 4: Monitor pipe status
SELECT *
FROM TABLE(INFORMATION_SCHEMA.PIPE_STATUS('sales_auto_ingest'));
-- Check pending files
SELECT
pipe_name,
pending_file_count,
last_loaded_file,
last_loaded_time,
error_count
FROM TABLE(INFORMATION_SCHEMA.PIPE_STATUS('sales_auto_ingest'));
-- Example 5: Query copy history
SELECT
pipe_name,
file_name,
file_size,
row_count,
error_count,
status,
load_time
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
TABLE_NAME => 'sales_raw',
START_TIME => DATEADD(day, -7, CURRENT_TIMESTAMP())
))
ORDER BY load_time DESC;
-- Example 6: Analyze load errors
SELECT
file_name,
error_code,
error_message,
row_number,
column_number,
loaded_time
FROM sales_raw$errors
ORDER BY loaded_time DESC
LIMIT 100;
-- Example 7: Create file format for semi-structured data
CREATE OR REPLACE FILE FORMAT json_format
TYPE = JSON
STRIP_OUTER_ARRAY = FALSE
STRIP_NULL_VALUES = FALSE
IGNORE_UTF8_ERRORS = FALSE
COMMENT = 'Standard JSON format for event data';
CREATE OR REPLACE FILE FORMAT parquet_format
TYPE = PARQUET
TRIM_SPACE = TRUE
IGNORE_UTF8_ERRORS = FALSE
COMMENT = 'Parquet format for analytics data';
-- Example 8: Advanced pipe configuration
CREATE PIPE analytics_ingest
AUTO_INGEST = TRUE
INTEGRATION = 'azure_analytics_integration'
PATTERN = '.*sales_[0-9]{8}\\.csv$'
ERROR_INTEGRATION = 'azure_error_queue'
COMMENT = 'Ingest sales data with pattern matching'
AS
COPY INTO analytics.sales_fact
FROM @azure_analytics_integration/raw_data/
FILE_FORMAT = (
TYPE = CSV
SKIP_HEADER = 1
NULL_IF = ('NA', 'N/A', '')
FIELD_DELIMITER = '|'
DATE_FORMAT = 'YYYY-MM-DD'
TIMESTAMP_FORMAT = 'YYYY-MM-DD HH24:MI:SS.FF3'
)
VALIDATION_MODE = RETURN_ERRORS
ON_ERROR = 'SKIP_FILE';
-- Example 9: Create alert for pipe lag
CREATE OR REPLACE ALERT pipe_lag_alert
WAREHOUSE = 'monitoring_wh'
SCHEDULE = 'USING CRON 0 */4 * * * America/New_York'
IF (
SELECT COUNT(*)
FROM TABLE(INFORMATION_SCHEMA.PIPE_STATUS('sales_auto_ingest'))
WHERE pending_file_count > 100
) > 0
THEN
CALL SYSTEM$SEND_EMAIL(
'alert_channel',
'data-team@company.com',
'Snowpipe Lag Alert',
'The sales_auto_ingest pipe has excessive pending files.'
);
Performance Metrics
| Metric | Value | Description |
|---|---|---|
| Typical Latency | 1-3 minutes | End-to-end from file upload to queryable |
| Max File Size | 5 GB | Maximum supported file size |
| Recommended Size | 100-250 MB | Optimal for parallel processing |
| Max Pipes per Account | 100 | Default limit (can be increased) |
| Max Integration per Pipe | 1 | One integration per pipe |
| Daily Quota | Unlimited | No limit on daily ingestion |
Best Practices
-
Optimize file sizes: Target 100-250 MB compressed files for optimal parallelism and processing efficiency.
-
Use Parquet for analytics: Prefer Parquet over CSV for analytics workloads due to built-in compression, schema information, and predicate pushdown.
-
Implement monitoring: Set up alerts for pipe lag, error rates, and ingestion latency to catch issues proactively.
-
Handle errors gracefully: Use ON_ERROR = 'CONTINUE' for non-critical data and 'SKIP_FILE' for malformed files. Review PIPE_ERR tables regularly.
-
Use pattern matching: Configure PATTERN parameters to filter specific file types and avoid accidental ingestion of unrelated files.
-
Implement idempotent loads: Leverage Snowpipe's automatic deduplication to safely handle retries and duplicate events.
-
Separate concerns: Create dedicated pipes for different data sources, file types, or target tables to simplify monitoring and error isolation.
-
Monitor pipe health: Regularly check PIPE_STATUS for pending file counts and error conditions. Address issues before they impact SLAs.
-
Test with small files: Validate file format configurations and error handling with sample files before production deployment.
-
Plan for scale: Design ingestion architecture to handle peak loads by using multiple pipes and optimizing cloud storage configurations.
See Also
- PySpark Iceberg - Iceberg table ingestion patterns
- Delta Lake on Databricks - Delta Lake ingestion comparison
- Data Warehouse Concepts - Data warehouse design principles