Image Generated by AI

Beyond Snowpipe: How I built a Self-Healing Data Pipeline for Dirty Files in Snowflake

I ran into a problem with my data that broke every standard tool I reached for and ultimately I ended up building something that fixed it permanently.

The Reality of Data Ingestion

The textbook version of a data pipeline is clean and elegant. A file lands in S3, Snowpipe fires, and rows flow into Snowflake without a hitch. And yes, it works, if you’re lucky enough to be working with well-structured sources, then premium tools like Fivetran handle the heavy lifting without complaint.

But anyone who’s worked with legacy systems knows the other version.

I was ingesting a 150-column pipe-delimited file from a legacy system. Somewhere buried in a comments field, a user had pressed Enter mid-sentence. That one newline split a single logical record across two physical lines and that’s it. That was enough to break everything I tried.

How you ask?

1. Well, Snowpipe is built for speed, not transformation. It expects every row to terminate cleanly, so when it hits the second half of a broken record, it reads it as a brand-new row,one that’s structurally invalid and chokes.
2. And Fivetran is excellent when data plays by the rules. When a row has the wrong column count or carries structural nulls where the schema demands values, Fivetran either silently drops those columns or stalls in an error loop or just brings in that data as-is.

Neither outcome is acceptable in production.

I needed something smarter: a pipeline that could detect the damage and repair it before the data ever reached the table. And I knew I had to do this by myself.

Phase 1: The Infrastructure

My first step was setting up a secure, automation-friendly connection. I used a Storage Integration to let Snowflake assume an IAM role directly (never use your hardcoded AWS credentials). This gives Snowflake the permissions it needs and nothing more.

IAM Policy (AWS side) Scope the policy to allow Snowflake to list and read from raw Bronze bucket. All we need is the ARN and the ExternalId from Integration object created in Snowflake:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": [
            "arn:aws:iam::..."]
            },
            "Action": ["s3:PutObject", "s3:GetObject", "s3:ListBucket"],
            "Condition": {
                "StringEquals": {
                    "sts:ExternalId": [
            "external_id_value_from_snowflake"
        ] 
                }
            },
            "Resource": ["arn:aws:s3:::my-bucket/*", "arn:aws:s3:::my-bucket"]
        }
    ]
}

Storage Integration and Directory Table (Snowflake side) Rather than hardcoding AWS credentials, a Storage Integration lets Snowflake securely assume the IAM role. Once the external stage is created, I enabled a Directory Table which added a metadata layer that Snowflake can query directly. And this is exactly what makes the next piece possible; placing a Stream over the stage so the pipeline only wakes up when a new file actually arrives.

CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::...';

CREATE STAGE raw_s3_stage
  URL = 's3://my-bucket/raw/'
  STORAGE_INTEGRATION = s3_int
  DIRECTORY = (ENABLE = TRUE);

CREATE STREAM my_stage_stream ON STAGE raw_s3_stage;

Phase 2: The Repair Procedure

This is where the real work happens. Using the snowflake-snowpark-python package, a Python Stored Procedure that acts as the pipeline's core repair engine.

I’ll explain why I went with Python, and why I almost got the architecture wrong.

Why Python, Specifically

SQL is powerful, but it can’t hold state between rows. Python can. When a line arrives with only 100 pipe characters and we expect 150, Python holds that line in a buffer, pulls the next line, and merges them. It keeps stitching until the row is structurally complete, even if the record got split across five physical lines.

The Performance Trap I Fell Into

Early versions of this procedure wrote the repaired data directly from Python into Snowflake using df.write.save_as_table(). It worked, technically. But on a file with 100,000 rows and 150 columns that's millions of individual data points. Python was converting each one into an API call or internal insert. The serialization overhead was brutal. This processing literally took hours until my patience ran out.

The Fix was Conceptually Simple: File-First, then Bulk Copy

I stopped using Python as a data loader and decided to use it only for what it’s actually good at: logic and repair. Then it can hand the file off to Snowflake’s bulk COPY engine, which processes chunks in parallel across every CPU in the warehouse.

The rewrite was simple in concept but dramatic in effect:

1. Python repairs the rows line by line, exactly as before
2. Instead of writing to a table, it writes to a temporary .csv.gz file in /tmp/
3. session.file.put() moves that file to a cleansed Silver stage
4. A single COPY INTO command takes it from there

Snowflake’s COPY engine processes files in parallel: it splits the file into chunks and uses every available CPU in the warehouse to pull them in simultaneously. Moving the heavy lifting away from Python and into Snowflake’s bulk loader brought processing time down from several hours to under two minutes.

Here is the pseudocode:

import io
import csv
from snowflake.snowpark.files import SnowflakeFile

def main(session):
    # 1. Fetch names of new files from the Stream
    new_files = session.sql("SELECT \"name\" FROM RAW_DATA_STREAM").collect()

    for file_row in new_files:
        file_path = file_row["name"]
        input_stage_path = f"@{raw_s3_stage}/{file_path}"
        temp_local_file = f"/tmp/repaired_{file_path.split('/')[-1]}"

        try:
            # 2. Stateful Repair Logic
            # We open the file as a stream to avoid memory overhead
            with SnowflakeFile.open(input_stage_path, 'rb') as f_in:
                with open(temp_local_file, 'w') as f_out:
                    writer = csv.writer(f_out, delimiter='|')
                    buffer = ""
                    expected_cols = 149 # 149 pipe characters give 100 columns

                    for line in f_in:
                        # Append line to buffer and check column count
                        buffer += line.strip('\r\n')
                        if buffer.count('|') >= expected_cols:
                            # If row is whole, write it out
                            row_data = next(csv.reader([buffer], delimiter='|'))
                            writer.writerow(row_data[:expected_cols])
                            buffer = ""
                        else:
                            # If incomplete, add a space (healing the newline)
                            buffer += " "

            # 3. Move repaired file to 'Cleansed' stage
            session.file.put(temp_local_file, "@CLEANSED_STAGE/processed/", overwrite=True)

            # 4. Optimized Bulk Load with Metadata
            session.sql(f"""
                COPY INTO TARGET_TABLE 
                FROM (SELECT $1, $2, ..., METADATA$FILENAME, METADATA$START_SCAN_TIME 
                      FROM @CLEANSED_STAGE/processed/)
                FILE_FORMAT = (TYPE = CSV FIELD_DELIMITER = '|')
            """).collect()

            # INSERT SUCCESSFUL TRANSACTION INTO AUDIT TABLE - PHASE 5

        except Exception as e:
            # Alerting and Error Handling logic
            # INSERT FAILED TRANSACTION INTO AUDIT TABLE - PHASE 5
            send_alert(session, file_path, str(e)) # I send an email here as shown below

    # 5. Final DML to consume the stream
    session.sql(" DELETE from TARGET_TABLE WHERE 1=0 AND EXISTS (SELECT 1 FROM RAW_DATA_STREAM) ").collect()
    return "Process Complete"

Phase 3: Metadata and Audit Trails

Every row that lands in the final table should be traceable. By using a transformation select inside the COPY INTO statement, I was able to inject the source filename and load timestamp at zero extra cost and no post-load joins.

COPY INTO final_table FROM (
  SELECT $1, $2, ... $204, METADATA$FILENAME, METADATA$START_SCAN_TIME 
  FROM @cleansed_stage
);

When something breaks and eventually something will, I’ll know exactly which file caused it and when it was loaded.

Why This Approach Holds Up

Off-the-shelf ETL tools are built for data that follows the rules. They excel at structured, well-formed sources, and they’re worth every dollar in those contexts. But legacy systems generate dirty data, and dirty data needs a pipeline with some flexibility baked in.

This design handles that flexibility at three levels.

1. The Python buffer addresses structural damage that loaders simply can’t see.
2. The file-plus-copy architecture keeps performance competitive with native ingestion and 10x faster than the row-by-row Python approach.
3. And the Stage Stream makes the whole thing incremental by default: the pipeline only runs when there’s actually something new to process, so this way I’m not paying for idle compute.

Phase 4: Governance & Monitoring: Tracking the “Heals”

A self-healing pipeline is only as good as its transparency otherwise how would I know what’s happening behind the scenes. If the logic is stitching records together in the background, I need to know how often it’s happening and more importantly if it ever encounters a file that is too “broken” to fix. And this is how I did it.

1. The Audit Log

Before the procedure finishes, I thought it would be a good practice to write a summary of the run to an audit table. This allows me to build a dashboard showing “Raw Records” vs. “Healed Records” over time.

INSERT INTO pipeline_audit_log (file_name, raw_count, healed_count, status, run_time)
VALUES (:file_name, :raw_count, :healed_count, 'SUCCESS', CURRENT_TIMESTAMP());

2. Real-Time Alerting with SYSTEM$SEND_EMAIL

If the procedure encounters a file where the column count is still wrong even after attempting to heal it, I don’t want to find out three days later. So I thought of using Snowflake’s native email integration to alert the team immediately.

CREATE OR REPLACE NOTIFICATION INTEGRATION MY_EMAIL_INT
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('data_ops@company.com');

Inside the Python procedure’s except block, I trigger a notification:

# If the heal logic fails or an error occurs:
error_msg = f"Critical failure in Pipeline: {str(e)}"
session.call("SYSTEM$SEND_EMAIL", 
             'MY_EMAIL_INT', 
             'data_ops@company.com', 
             'DATA PIPELINE ALERT: Structural Failure', 
             error_msg)

3. Why This Matters

By adding alerting, I move from Reactive to Proactive Data Engineering. Instead of waiting for a downstream analyst to complain about missing data, my pipeline now “heals” what it can and “screams” when it can’t. This builds immense trust within stakeholders.

Phase 5: The Orchestration Layer: Automating the Weekly Workflow

Building a resilient Python procedure is only half the battle. In a production environment, I do not want to execute this setup manually and needed a reliable way to ensure the stage is refreshed and the procedure is called in the correct order.

So I set up a Task DAG (Directed Acyclic Graph) to handle this. This allows me to create a parent-child relationship between the “Refresh” and the “Process” steps.

1. The Parent: The Stage Refresher

My first task tells Snowflake to look at the S3 bucket and update its metadata. I have scheduled this for every Wednesday at midnight.

CREATE OR REPLACE TASK WEEKLY_STAGE_REFRESH_TASK
  WAREHOUSE = 'COMPUTE_WH'
  SCHEDULE = 'USING CRON 0 0 * * 3 UTC' 
AS
  ALTER STAGE raw_s3_stage REFRESH;

2. The Child: The Self-Healing Processor

Next, I created the task that calls the Python procedure. By using the AFTER keyword, I ensure it only starts once the refresh is complete. More importantly, I use a WHEN clause to check the Stream. If the refresh didn't find any new files, the task won't even spin up a warehouse, saving significant compute costs.

CREATE OR REPLACE TASK WEEKLY_PROCESSOR_TASK
  WAREHOUSE = 'COMPUTE_WH'
  AFTER WEEKLY_STAGE_REFRESH_TASK
  WHEN SYSTEM$STREAM_HAS_DATA('MY_STAGE_STREAM')
AS
  CALL REPAIR_AND_LOAD_PROC();

Why Orchestration is the “Secret Sauce”

By decoupling the Refresh from the Processing, I created a modular system. If in the future my AWS credentials expire, the Parent task fails, and the Child task safely never starts. This prevents “partial loads” and ensures that my audit logs remain clean and accurate.

It’s Not Always About the Most Expensive Tool

All I want to say in conclusion is that I’ve worked with Fivetran. I’ve used Snowpipe occasionally. They’re genuinely excellent for what they’re designed for. But they’re designed for data that follows the rules and legacy systems often don’t.

What I built here isn’t a replacement for those tools. It’s what you reach for when your data is too messy for them to handle gracefully. Combining Snowflake’s native infrastructure with Python’s stateful logic took a consistently failing ingestion process and made it something I could stop worrying about. And in production, that’s the whole point.