Building a Flight Data Quality Pipeline with Snowflake and Azure Blob

-

 


I recently had the chance to build something in Snowflake, and I figured it was worth documenting the process since there were a few things that weren't immediately obvious coming from a Microsoft Fabric / Azure background.

The project: a data quality pipeline that ingests BTS On-Time Performance flight data into Snowflake via Azure Blob Storage, validates every record against a set of quality rules, and routes clean vs. rejected records into separate tables. I filtered the data to Dallas/Fort Worth International Airport (DFW), partly because the domain is interesting and partly because it made the project feel less like a toy example.

The full source is on GitHub: snowflake-dfw-flight-pipeline

The Setup

Snowflake's external stage + Snowpipe combination is the equivalent of what I'd normally do with Azure Data Factory and a Blob trigger — files land in storage, something detects them, and they get loaded into a table. The implementation is a bit more explicit in Snowflake.

First, a storage integration:

CREATE STORAGE INTEGRATION DFW_AZURE_INTEGRATION
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'AZURE'
  ENABLED = TRUE
  AZURE_TENANT_ID = '<your-tenant-id>'
  STORAGE_ALLOWED_LOCATIONS = ('azure://<your-account>.blob.core.windows.net/dfw-flight-data/');

After running DESC INTEGRATION DFW_AZURE_INTEGRATION, you get an AZURE_CONSENT_URL — paste that into your browser and grant Snowflake's service principal access, then assign it the Storage Blob Data Contributor role in Azure IAM. This tripped me up the first time because the service principal shows up in the IAM member search as a truncated name, but it's the same thing.

Once the integration is in place, the external stage is straightforward:

CREATE OR REPLACE STAGE DFW_FLIGHTS.RAW.DFW_RAW_STAGE
  STORAGE_INTEGRATION = DFW_AZURE_INTEGRATION
  URL = 'azure://<your-account>.blob.core.windows.net/dfw-flight-data/'
  FILE_FORMAT = DFW_CSV_FORMAT;

A quick LIST @DFW_FLIGHTS.RAW.DFW_RAW_STAGE; there are no errors, which confirms Snowflake can see the container.

The Schema Discovery Problem

The BTS CSV files have 110 columns — not 64, which is what most online references suggest. The difference is that BTS includes diversion detail columns that most documentation ignores. There's also a trailing comma on every row that adds a phantom 111th column if you don't account for it.

The cleanest way I found to inspect the actual headers without loading the file first:

CREATE OR REPLACE FILE FORMAT DFW_CSV_HEADERS
  TYPE = 'CSV'
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  SKIP_HEADER = 0
  NULL_IF = ('', 'NA')
  EMPTY_FIELD_AS_NULL = TRUE
  TRIM_SPACE = TRUE;

SELECT $1,$2,$3,...$110
FROM @DFW_FLIGHTS.RAW.DFW_RAW_STAGE
(FILE_FORMAT => 'DFW_CSV_HEADERS')
LIMIT 1;

This returns the header row as positional columns, which you can then use to build your CREATE TABLE statement accurately. The trailing comma gets absorbed by a TRAILING_COMMA VARCHAR column at the end.

Snowpipe

Snowpipe with Azure requires a notification integration pointing to an Azure Storage Queue, and the queue needs its own service principal authorization (this needs to be separate from the blob storage integration — two different principals, two different IAM role assignments).

CREATE OR REPLACE NOTIFICATION INTEGRATION DFW_EVENT_GRID
  ENABLED = TRUE
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = AZURE_STORAGE_QUEUE
  AZURE_STORAGE_QUEUE_PRIMARY_URI = '<your-queue-uri>'
  AZURE_TENANT_ID = '<your-tenant-id>';

CREATE OR REPLACE PIPE DFW_FLIGHTS.RAW.DFW_SNOWPIPE
  AUTO_INGEST = TRUE
  INTEGRATION = 'DFW_EVENT_GRID'
  AS
  COPY INTO DFW_FLIGHTS.RAW.FLIGHT_RAW
  FROM @DFW_FLIGHTS.RAW.DFW_RAW_STAGE
  FILE_FORMAT = (FORMAT_NAME = 'DFW_FLIGHTS.RAW.DFW_CSV_FORMAT');

For this project I'm triggering ingestion manually rather than wiring up Azure Event Grid:

ALTER PIPE DFW_FLIGHTS.RAW.DFW_SNOWPIPE REFRESH;

Snowflake tracks which files have already been loaded, so running this repeatedly is safe — it only picks up new files.

The Data Quality Layer

This is the main point of the project. After ingestion, a stored procedure profiles every record in the RAW table and routes it to either CLEAN or FLAGGED:

CREATE OR REPLACE PROCEDURE DFW_FLIGHTS.RAW.SP_VALIDATE_FLIGHTS()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
BEGIN
  TRUNCATE TABLE DFW_FLIGHTS.CLEAN.FLIGHT_CLEAN;
  TRUNCATE TABLE DFW_FLIGHTS.FLAGGED.FLIGHT_FLAGGED;

  INSERT INTO DFW_FLIGHTS.FLAGGED.FLIGHT_FLAGGED
  SELECT *, 
    CASE
      WHEN FLIGHT_DATE IS NULL THEN 'NULL: FLIGHT_DATE'
      WHEN ORIGIN IS NULL THEN 'NULL: ORIGIN'
      WHEN NOT RLIKE(ORIGIN, '[A-Z]{3}') THEN 'INVALID: ORIGIN airport code'
      WHEN CANCELLED = 0 AND CRS_ELAPSED_TIME <= 0 THEN 'INVALID: CRS_ELAPSED_TIME'
      WHEN ACTUAL_ELAPSED_TIME > 1440 THEN 'INVALID: ACTUAL_ELAPSED_TIME exceeds 24 hours'
      WHEN CANCELLED = 1 AND DIVERTED = 1 THEN 'INVALID: cancelled and diverted'
      -- ... additional rules
    END AS FLAG_REASON,
    CURRENT_TIMESTAMP() AS FLAGGED_AT
  FROM DFW_FLIGHTS.RAW.FLIGHT_RAW
  WHERE FLIGHT_DATE IS NULL OR ORIGIN IS NULL
    OR NOT RLIKE(ORIGIN, '[A-Z]{3}')
    OR (CANCELLED = 0 AND CRS_ELAPSED_TIME <= 0)
    -- ... matching WHERE conditions
    ;

  INSERT INTO DFW_FLIGHTS.CLEAN.FLIGHT_CLEAN
  SELECT * EXCLUDE (TRAILING_COMMA)
  FROM DFW_FLIGHTS.RAW.FLIGHT_RAW
  WHERE FLIGHT_DATE IS NOT NULL AND ORIGIN IS NOT NULL
    AND RLIKE(ORIGIN, '[A-Z]{3}')
    -- ... inverse conditions
    ;

  RETURN 'Validation complete. Clean: ' ||
    (SELECT COUNT(*) FROM DFW_FLIGHTS.CLEAN.FLIGHT_CLEAN) ||
    ' rows. Flagged: ' ||
    (SELECT COUNT(*) FROM DFW_FLIGHTS.FLAGGED.FLIGHT_FLAGGED) || ' rows.';
END;
$$;

A couple of things worth noting here:

RLIKE, not REGEXP. Snowflake's regex function is RLIKE, not REGEXP. Using REGEXP either throws an error or silently returns no matches depending on context - I admittedly lost some time to this.

RAW is never truncated. Only CLEAN and FLAGGED get cleared on each run. RAW is the system of record.

One useful completeness check after each run:

SELECT
    (SELECT COUNT(*) FROM DFW_FLIGHTS.RAW.FLIGHT_RAW) AS RAW_COUNT,
    (SELECT COUNT(*) FROM DFW_FLIGHTS.CLEAN.FLIGHT_CLEAN) AS CLEAN_COUNT,
    (SELECT COUNT(*) FROM DFW_FLIGHTS.FLAGGED.FLIGHT_FLAGGED) AS FLAGGED_COUNT,
    RAW_COUNT - CLEAN_COUNT - FLAGGED_COUNT AS DISCREPANCY;

A zero in the DISCREPANCY column means every record landed in exactly one downstream table. After loading January through March 2024 (1,658,267 records), the result was:



The 10 flagged records included 8 synthetic bad rows I inserted to test the rules, plus 2 genuine data quality issues found in the February/March data — a null CRS_ELAPSED_TIME and a CRS_DEP_TIME out of valid HHMM range.

Aggregation Views

Three views sit on top of CLEAN and update automatically as new months are loaded:

  • V_ONTIME_BY_AIRLINE — departure and arrival on-time rates, average delay minutes, and cancellation rate per carrier for DFW flights
  • V_DELAY_CAUSES — carrier, weather, NAS, security, and late aircraft delay minutes broken down by airline
  • V_MONTHLY_TRENDS — volume, on-time rate, and average delays by month

Since they're views rather than tables, there's nothing to refresh — they always reflect whatever is currently in FLIGHT_CLEAN.

Thoughts on Snowflake Coming from Fabric

The SQL dialect is clean and mostly unsurprising. SELECT * EXCLUDE (column_name) is genuinely useful and I wish it existed everywhere. The separation of compute (warehouses) and storage is explicit in a way that Fabric abstracts away, which I found easier to reason about for cost purposes.

The main friction point was the IAM setup — two separate service principals, two separate consent flows, two separate role assignments. It's not difficult but it's easy to mix up which principal needs which permission. The DESC INTEGRATION output is your friend here.

The full SQL is in the GitHub repo if you want to run it yourself. Just make sure to add your own Azure-specific details.

Comments

Post a Comment

Popular posts from this blog

Imperial to Metric Conversion (and vice-versa) Script

-
Image
 Imperial to Metric Conversion (and vice-versa) Script An idea for a project that I had was one to have a python script that would use functions to convert Imperial values to Metric and also Metric values to Imperial. This is useful here in Canada, where we will often measure temperature in Celcius and distance in kilometers, but then also measure height in feet and weight in pounds! It can all get a bit much to keep track of sometimes. I wanted to get more familiar with classes in python, so I started with a master 'Conversion' class that would handle the math, and then several local classes for each kind of unit being converted. This is the main class and the units being utilized: class Converter:     """     A class to convert values between different units of measurement.     Supports temperature, length, volume, and weight conversions.     """     def __init__(self):         """         I...
Read more

Real Estate Data Pipeline, Part 1

-
Image
 This will be the first of two posts documenting the process of creating a data pipeline using real estate data. Overall, the whole adventure will include: Web-scraping Toronto real estate data Cleaning and transforming everything in Spark Exporting and loading the data into AWS RedShift Creating a visualization that puts it all into graphical form Overall, this will be a post dealing not only with the creation of the pipeline, but things that were learned along the way. Extract - Web-Scraping with Python This was my first foray into web-scraping and, using BeautifulSoup, the process was a lot easier than I had anticipated. The web-scraping code can be found here . We begin with our import statements: from bs4 import BeautifulSoup import requests import pandas as pd import re import time BeautifulSoup, as mentioned is one of the standard web-scraping libraries, but there are also options like Selenium, Scrapy and lxml. RE will provide us with regular expression pattern...
Read more

Real Estate Data Pipeline, Part 2

-
Image
  In our last data adventure, the E and the T of our ETL process was accomplished by way of ingesting real estate data via web scraping with python; then, some transformations were done using Apache Spark to clean the data and format it as needed. Next, we're going to be loading it into AWS Redshift and doing some visualizations in Power BI. Making a connection to AWS Redshift via Python can be a daunting task at first, but there are modules like AWS Data Wrangler (now known as AWS SDK for pandas)  that can simplify the process somewhat. AWS SDK for Pandas, in their own words, is a "python initiative that extends the power of Pandas library to AWS connecting DataFrames and AWS data related services." And they boast "easy integration with Athena, Glue, Redshift, Timestream, OpenSearch, Neptune, QuickSight, Chime, CloudWatchLogs, DynamoDB, EMR, SecretManager, PostgreSQL, MySQL, SQLServer and S3 (Parquet, CSV, JSON and EXCEL)." But, this isn't an AWS Wrangler...
Read more