Flight Tracker ELT Pipeline with dbt

Introduction

30 Second Summary

Every second, thousands of aircraft broadcast their position, speed, and altitude to receivers around the world. Data engineers at airlines and logistics companies build pipelines to capture this flood of real-time information and turn it into actionable insights.

In this project, you will build a complete ELT pipeline that ingests live flight tracking data, loads it into a cloud data warehouse, transforms it into an analytics-ready star schema, and visualizes the results in an interactive dashboard.

What You'll Build

A production-grade ELT data pipeline that extracts real-time flight data from the OpenSky Network API, transforms it in Snowflake using dbt, and displays the results in a Streamlit dashboard.

By the end of this project, you'll have:

• A Python ingestion script that extracts live flight data from a public API and loads it into Snowflake with error handling and idempotent writes.
• A star schema data model built with dbt, including staging models, dimension tables, and a fact table with automated data quality tests.
• An interactive Streamlit dashboard showing flight traffic by country, altitude distributions, and a live aircraft map connected directly to your warehouse.
• Secret Mission: Add incremental loading and a scheduled pipeline that detects and processes new flights automatically.

Are there any prerequisites?

You should be comfortable with Python basics and SQL fundamentals. You'll also need Python and VS Code installed on your machine before starting.

Set Up Your Environment

Before you can build your flight tracking pipeline, you need a solid foundation. This means having the right Python packages installed, a cloud data warehouse to load data into, and API credentials to fetch live flight data.

In this step, you will set up a Python virtual environment with all required packages, create a free Snowflake trial account, and register for an OpenSky Network account to get API credentials.

In this step, get ready to:

• Install project dependencies in a Python virtual environment.
• Create a free Snowflake trial account with $400 in credits.
• Register for a free OpenSky Network account and create API credentials.

Create the project folder and install dependencies

Your project needs several Python packages working together. You will use dbt-core for SQL transformations, snowflake-connector-python for loading data, and Streamlit for visualization.

• Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir flight-tracker
cd flight-tracker

• Create a Python virtual environment by running this command:

python3 -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation that keeps this project's packages separate from your system Python. This prevents version conflicts between projects.

• Activate the virtual environment by running this command:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try the activate command again.

On macOS, make sure you typed source before the path. Without it, the command will not activate the environment.

Help me activate my Python virtual environment.

Now you will create the requirements.txt file that lists all the packages your project needs.

• Create requirements.txt and add the project dependencies. Open VS Code in this folder by running:

code .

What does code . do?

The code . command opens VS Code with the current folder as the workspace. The dot refers to the current directory.

• In VS Code, click File in the top menu bar.
• Click New File.
• Name the file requirements.txt and press Enter.
• Paste the following contents into requirements.txt:

dbt-snowflake==1.11.5
snowflake-connector-python==4.6.0
streamlit==1.58.0
requests==2.32.3
python-dotenv==1.1.0

What does this file do?

• dbt-snowflake==1.11.5 installs dbt-core with the Snowflake adapter for SQL transformations.
• snowflake-connector-python==4.6.0 lets Python connect to your Snowflake warehouse to load data.
• streamlit==1.58.0 provides the dashboard framework for visualizing your flight data.
• requests==2.32.3 handles HTTP calls to the OpenSky Network API.
• python-dotenv==1.1.0 loads environment variables from a .env file so you can keep credentials out of your code.

• Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
• Install all the packages by running this command in your terminal:

pip install -r requirements.txt

This will take a minute or two as pip downloads and installs all the packages and their dependencies.

• Confirm the install succeeded by running:

dbt --version

✔️ I see a dbt version number

Your packages installed correctly. You should see output showing dbt-core and dbt-snowflake versions. Move on to the next substep.

ⓧ Command not found or an error appeared

Make sure your virtual environment is still active. You should see (venv) in your terminal prompt. If not, run the activate command again.

If you see a permissions error, try running pip install --user -r requirements.txt instead.

Help me fix my dbt installation.

✔️ Awesome, I've got everything!

Great. Double check you have saved requirements.txt and your virtual environment is still active.

ⓧ I'd like to double check the full code

dbt-snowflake==1.11.5
snowflake-connector-python==4.6.0
streamlit==1.58.0
requests==2.32.3
python-dotenv==1.1.0

Sign up for a free Snowflake trial account

Snowflake is a cloud data warehouse where your flight data will live. The free trial gives you $400 in credits for 30 days with no credit card required.

• Open your browser and navigate to the Snowflake signup page:

https://signup.snowflake.com/

• Enter your First name, Last name, and Work email (a personal email works fine).
• Select Personal learning and development from the Why are you signing up? dropdown.
• Click Continue.
• On the next page, select Enterprise as your Snowflake edition.
• Choose a cloud provider (AWS is recommended for this project).
• Select the region closest to you.
• Click Get Started.

Why Enterprise edition?

Enterprise gives you access to the most features during your trial. The edition, cloud provider, and region cannot be changed after signup, so pick them carefully now.

• Check your email for an activation link from Snowflake.
• Click the activation link and set a username and password for your account.
• Log in to Snowflake at https://app.snowflake.com/ using your new credentials.

You should see the Snowsight interface with a navigation menu on the left side. This confirms your account is active.

Can't find the activation email?

Check your spam or junk folder. Wait a few minutes for delivery if needed. Make sure the email address you entered during signup is correct.

Help me find my Snowflake activation email.

Note your account identifier

After logging in, look at the URL in your browser. It will look something like abc12345.us-east-1.snowflakecomputing.com. The part before .snowflakecomputing.com is your account identifier. You will need this in the next step.

Register for a free OpenSky Network account

The OpenSky Network provides real-time flight tracking data from thousands of receivers worldwide. A registered account gives you 4,000 API credits per day instead of the 400 you get with anonymous access.

• Open the OpenSky Network website in your browser:

https://opensky-network.org

• Click Sign Up or Register to create a new account.
• Fill in the registration form with your email and a password.
• Verify your email if prompted, then log in to your new account.

Now you need to create API credentials for programmatic access. Since March 2026, OpenSky requires OAuth2 client credentials instead of basic username/password authentication.

• Navigate to the Account page:

https://opensky-network.org/my-opensky/account

• Find the API Client section on the page.
• Click Create to generate a new API client.
• Copy your client_id and client_secret and save them somewhere safe (you will need them in the next step).

What are client_id and client_secret?

These are your OAuth2 credentials for the OpenSky REST API. Your ingestion script will use them to fetch an access token, which then authenticates each API request. Tokens expire after 30 minutes and your script will refresh them automatically.

Without credentials, you can still access the API anonymously, but with stricter rate limits (400 credits/day vs. 4,000).

Don't see the API Client section?

Make sure you are logged in and viewing the Account page at the URL above. If you just registered, try refreshing the page. The API Client card should appear on your account settings page.

Help me create my OpenSky API credentials.

Your environment is ready. You have all the packages installed, a cloud warehouse waiting for data, and API credentials to fetch live flights. Next up, you will configure Snowflake and create the database structure for your pipeline.

Configure Snowflake and Create the Data Warehouse Structure

In the previous step, you installed all your project dependencies and set up accounts for Snowflake and OpenSky Network.

Now you need to build the infrastructure inside Snowflake that will hold your flight data. In a real data engineering team, you would organize data into layers: raw data lands first, gets cleaned in staging, and becomes analytics-ready in a marts layer.

In this step, get ready to:

• Create a database with separate schemas for your data layers (raw, staging, marts).
• Create a virtual warehouse and a service user for programmatic access.
• Verify your local machine can connect to Snowflake using Python.

Create the database, schemas, and virtual warehouse

Your Snowflake data warehouse needs a structured home for your flight data. You will create three schemas inside a single database to separate concerns: raw data in, cleaned data out, and analytics-ready models on top.

• Return to your Snowflake web UI in your browser.
• In the left navigation menu, select Projects, then Worksheets.
• Click the + button and select SQL Worksheet.
• Make sure your role in the top-left context selector is set to ACCOUNTADMIN.
• Paste the following SQL into the worksheet and run it by clicking the Run button (or pressing Cmd+Enter on macOS, Ctrl+Enter on Windows):

-- Create the flight tracker database
CREATE DATABASE IF NOT EXISTS FLIGHT_TRACKER;

-- Create schemas for each data layer
CREATE SCHEMA IF NOT EXISTS FLIGHT_TRACKER.RAW;
CREATE SCHEMA IF NOT EXISTS FLIGHT_TRACKER.STAGING;
CREATE SCHEMA IF NOT EXISTS FLIGHT_TRACKER.MARTS;

-- Create an extra-small warehouse for compute
CREATE WAREHOUSE IF NOT EXISTS FLIGHT_WH
  WAREHOUSE_SIZE = XSMALL
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE;

What does this SQL do?

This creates the three-layer structure that professional data teams use. RAW holds untouched API data. STAGING holds cleaned versions. MARTS holds analytics-ready tables.

The virtual warehouse FLIGHT_WH is the compute engine that runs your queries. It auto-suspends after 60 seconds of inactivity to preserve your trial credits, and auto-resumes the moment you run a query.

• You should see a green success message for each statement in the results panel at the bottom of the worksheet.

Seeing a permissions error?

Make sure your role is set to ACCOUNTADMIN in the context selector at the top left of the worksheet. Trial accounts start with this role, which has full permissions.

If you see an error about the warehouse already existing, that is fine. The IF NOT EXISTS clause makes the commands safe to re-run.

Help me fix a permissions error in Snowflake

Create a service user for programmatic access

Your Python scripts need their own user account to connect to Snowflake. This keeps your personal admin credentials separate from the pipeline credentials.

• In the same SQL worksheet, paste and run the following SQL:

-- Create a service user for the pipeline
CREATE USER IF NOT EXISTS flight_pipeline_user
  PASSWORD = 'your-password-here'
  DEFAULT_WAREHOUSE = 'FLIGHT_WH'
  DEFAULT_ROLE = 'SYSADMIN'
  MUST_CHANGE_PASSWORD = FALSE;

-- Grant the SYSADMIN role to the service user
GRANT ROLE SYSADMIN TO USER flight_pipeline_user;

-- Grant warehouse usage
GRANT USAGE ON WAREHOUSE FLIGHT_WH TO ROLE SYSADMIN;

-- Grant database and schema access
GRANT USAGE ON DATABASE FLIGHT_TRACKER TO ROLE SYSADMIN;
GRANT USAGE ON ALL SCHEMAS IN DATABASE FLIGHT_TRACKER TO ROLE SYSADMIN;
GRANT CREATE TABLE ON ALL SCHEMAS IN DATABASE FLIGHT_TRACKER TO ROLE SYSADMIN;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA FLIGHT_TRACKER.RAW TO ROLE SYSADMIN;
GRANT ALL ON SCHEMA FLIGHT_TRACKER.RAW TO ROLE SYSADMIN;
GRANT ALL ON SCHEMA FLIGHT_TRACKER.STAGING TO ROLE SYSADMIN;
GRANT ALL ON SCHEMA FLIGHT_TRACKER.MARTS TO ROLE SYSADMIN;

What does this SQL do?

This creates a dedicated user called flight_pipeline_user that your Python ingestion script will use to authenticate. The SYSADMIN role gives it enough permissions to create and modify tables in all three schemas.

The password placeholder your-password-here should be replaced with a strong password of your choice. You will use this same password in your .env file shortly.

• Replace your-password-here in the SQL with a strong password you will remember.
• Run the SQL. You should see a green success message for each statement.

Getting a user creation error?

Snowflake passwords must be at least 8 characters. If you see a password policy error, choose a longer password with a mix of uppercase, lowercase, and numbers.

Help me create the service user in Snowflake

Create the .env file and verify connectivity

Your Python scripts need connection details stored in a .env file. You also need your Snowflake account identifier, which is a unique string that identifies your account.

• To find your account identifier in Snowflake, open the account selector (click the account name in the bottom-left of the Snowsight navigation).
• Select View account details.
• Copy the account identifier from the Account Details dialog. It will be in the format orgname-account_name.

What is the account identifier?

The account identifier is what the Python connector uses to find your specific Snowflake instance on the internet. It is unique to your trial account and looks something like myorg-myaccount.

• Switch back to VS Code.
• Create a new file called .env in your project folder by running:

touch .env

• Open .env in VS Code and paste the following, replacing the placeholder values with your actual credentials:

SNOWFLAKE_ACCOUNT=your-account-identifier
SNOWFLAKE_USER=flight_pipeline_user
SNOWFLAKE_PASSWORD=your-password-here
SNOWFLAKE_WAREHOUSE=FLIGHT_WH
SNOWFLAKE_DATABASE=FLIGHT_TRACKER
SNOWFLAKE_SCHEMA=RAW
OPENSKY_CLIENT_ID=your-opensky-client-id
OPENSKY_CLIENT_SECRET=your-opensky-client-secret

What goes in each field?

• SNOWFLAKE_ACCOUNT: the account identifier you copied from the Account Details dialog.
• SNOWFLAKE_PASSWORD: the password you set when creating flight_pipeline_user.
• OPENSKY_CLIENT_ID and OPENSKY_CLIENT_SECRET: your OpenSky Network OAuth credentials from the account you registered in Step 1.

• Save .env.

Now verify that your Python environment can actually connect to Snowflake. Create a quick test script.

• Create a test file by running:

touch test_connection.py

• Open test_connection.py and paste the following:

import os
import snowflake.connector
from dotenv import load_dotenv

# Load environment variables from .env
load_dotenv()

# Connect to Snowflake using the .env credentials
conn = snowflake.connector.connect(
    account=os.getenv("SNOWFLAKE_ACCOUNT"),
    user=os.getenv("SNOWFLAKE_USER"),
    password=os.getenv("SNOWFLAKE_PASSWORD"),
    warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
    database=os.getenv("SNOWFLAKE_DATABASE"),
    schema=os.getenv("SNOWFLAKE_SCHEMA"),
)

# Run a simple query to confirm the connection works
cursor = conn.cursor()
cursor.execute("SELECT CURRENT_WAREHOUSE(), CURRENT_DATABASE(), CURRENT_SCHEMA()")
result = cursor.fetchone()
print(f"Connected! Warehouse: {result[0]}, Database: {result[1]}, Schema: {result[2]}")

cursor.close()
conn.close()

What does this script do?

It loads your .env credentials using python-dotenv, opens a connection to Snowflake, and runs a simple query that returns the active warehouse, database, and schema. If you see all three values printed, your pipeline is ready to talk to Snowflake.

• Save test_connection.py and run it:

python test_connection.py

• You should see output like: Connected! Warehouse: FLIGHT_WH, Database: FLIGHT_TRACKER, Schema: RAW.

Connection failed?

If you see an error about the account identifier, double-check that the value in your .env matches what you copied from the Account Details dialog. It should be in the format orgname-account_name with no .snowflakecomputing.com suffix.

If you see an authentication error, confirm the password in .env matches the one you used in the CREATE USER SQL statement.

Help me fix my Snowflake connection error

✔️ Awesome, I've got everything!

Great. Double check you have saved both .env and test_connection.py. You can delete test_connection.py now since it was just for verification.

ⓧ I'd like to double check the full code

Here is the complete .env file you should have:

SNOWFLAKE_ACCOUNT=your-account-identifier
SNOWFLAKE_USER=flight_pipeline_user
SNOWFLAKE_PASSWORD=your-password-here
SNOWFLAKE_WAREHOUSE=FLIGHT_WH
SNOWFLAKE_DATABASE=FLIGHT_TRACKER
SNOWFLAKE_SCHEMA=RAW
OPENSKY_CLIENT_ID=your-opensky-client-id
OPENSKY_CLIENT_SECRET=your-opensky-client-secret

Your Snowflake infrastructure is ready. Next up, you will write the Python ingestion script that pulls live flight data from the OpenSky Network API and loads it into the RAW schema.

Build the Python Ingestion Script

Your Snowflake warehouse is ready and waiting for data. Now you need something to actually feed it.

In this step, you will write a Python script that pulls live flight data from the OpenSky Network API, transforms it into structured records, and loads it into your Snowflake RAW schema. This is the Extract and Load portion of your ELT pipeline.

In this step, get ready to:

• Write Python functions to authenticate with the OpenSky Network API and fetch live flight state vectors.
• Transform raw API responses into structured records and load them into Snowflake.
• Run the full ingestion pipeline and verify flight data appears in your RAW schema.

Write the API extraction functions

The OpenSky Network provides a REST API that returns real-time state vectors for every aircraft currently tracked worldwide. Your script needs to authenticate using OAuth2, then call the /states/all endpoint.

• In VS Code, create a new file called ingest.py in your project folder.
• Paste the following code into ingest.py to set up imports, logging, and the API extraction functions:

import os
import logging
import time
from datetime import datetime, timezone

import requests
import snowflake.connector
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

OPENSKY_API_URL = "https://opensky-network.org/api/states/all"
OPENSKY_TOKEN_URL = "https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token"

MAX_RETRIES = 3
RETRY_DELAY = 5


def get_opensky_token():
    """Fetch an OAuth2 access token from the OpenSky Network."""
    client_id = os.getenv("OPENSKY_CLIENT_ID")
    client_secret = os.getenv("OPENSKY_CLIENT_SECRET")

    if not client_id or not client_secret:
        logger.info("No OpenSky credentials found. Using anonymous access.")
        return None

    response = requests.post(
        OPENSKY_TOKEN_URL,
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    response.raise_for_status()
    return response.json()["access_token"]


def fetch_flight_data(token=None):
    """Fetch current flight state vectors from OpenSky Network API."""
    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    for attempt in range(MAX_RETRIES):
        try:
            response = requests.get(
                OPENSKY_API_URL,
                headers=headers,
                timeout=30,
            )
            response.raise_for_status()
            data = response.json()
            logger.info(f"Fetched {len(data.get('states', []))} flight states")
            return data
        except requests.exceptions.RequestException as e:
            logger.warning(f"Attempt {attempt + 1} failed: {e}")
            if attempt < MAX_RETRIES - 1:
                time.sleep(RETRY_DELAY)
            else:
                raise

What does this code do?

This sets up the extraction layer of your pipeline with three main sections:

• Imports and configuration loads environment variables from .env and sets up structured logging so you can trace every pipeline run.
• get_opensky_token() exchanges your OAuth2 client credentials for a Bearer token. If no credentials are found, it gracefully falls back to anonymous access (lower rate limits).
• fetch_flight_data() calls the /states/all endpoint with automatic retries. If a request fails, it waits 5 seconds and tries again up to 3 times before raising the error.

Why retry logic matters

Public APIs are unreliable. Network blips, rate limits, and server errors all happen in production. The retry pattern with exponential backoff (here simplified as a fixed 5-second delay) ensures your pipeline does not fail on transient issues.

• Save ingest.py.

You will not see any output yet. The next section adds the functions that transform and load this data into Snowflake.

Transform the data and load it into Snowflake

The OpenSky API returns state vectors as a nested array where each aircraft is a list of positional values. You need to map these array indices to named fields and load the structured result into Snowflake.

• In ingest.py, add the following three functions below fetch_flight_data:

def transform_states(raw_data):
    """Transform raw API response into structured records."""
    if not raw_data or not raw_data.get("states"):
        return []

    api_time = raw_data["time"]
    records = []

    for state in raw_data["states"]:
        records.append({
            "icao24": state[0],
            "callsign": state[1].strip() if state[1] else None,
            "origin_country": state[2],
            "time_position": state[3],
            "last_contact": state[4],
            "longitude": state[5],
            "latitude": state[6],
            "baro_altitude": state[7],
            "on_ground": state[8],
            "velocity": state[9],
            "true_track": state[10],
            "vertical_rate": state[11],
            "geo_altitude": state[13],
            "squawk": state[14],
            "position_source": state[16],
            "category": state[17] if len(state) > 17 else None,
            "api_timestamp": api_time,
        })

    return records


def get_snowflake_connection():
    """Create a Snowflake connection using environment variables."""
    return snowflake.connector.connect(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema=os.getenv("SNOWFLAKE_SCHEMA"),
    )


def load_to_snowflake(records):
    """Load flight records into the Snowflake raw layer using MERGE for idempotency."""
    if not records:
        logger.warning("No records to load.")
        return

    conn = get_snowflake_connection()
    cursor = conn.cursor()

    try:
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS RAW.flight_states (
                icao24 VARCHAR(6),
                callsign VARCHAR(8),
                origin_country VARCHAR(100),
                time_position INTEGER,
                last_contact INTEGER,
                longitude FLOAT,
                latitude FLOAT,
                baro_altitude FLOAT,
                on_ground BOOLEAN,
                velocity FLOAT,
                true_track FLOAT,
                vertical_rate FLOAT,
                geo_altitude FLOAT,
                squawk VARCHAR(4),
                position_source INTEGER,
                category INTEGER,
                api_timestamp INTEGER,
                loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
                PRIMARY KEY (icao24, api_timestamp)
            )
        """)

        insert_sql = """
            INSERT INTO RAW.flight_states (
                icao24, callsign, origin_country, time_position,
                last_contact, longitude, latitude, baro_altitude,
                on_ground, velocity, true_track, vertical_rate,
                geo_altitude, squawk, position_source, category,
                api_timestamp
            )
            SELECT %(icao24)s, %(callsign)s, %(origin_country)s, %(time_position)s,
                   %(last_contact)s, %(longitude)s, %(latitude)s, %(baro_altitude)s,
                   %(on_ground)s, %(velocity)s, %(true_track)s, %(vertical_rate)s,
                   %(geo_altitude)s, %(squawk)s, %(position_source)s, %(category)s,
                   %(api_timestamp)s
            WHERE NOT EXISTS (
                SELECT 1 FROM RAW.flight_states
                WHERE icao24 = %(icao24)s AND api_timestamp = %(api_timestamp)s
            )
        """

        loaded_count = 0
        for record in records:
            cursor.execute(insert_sql, record)
            loaded_count += cursor.rowcount

        conn.commit()
        logger.info(f"Loaded {loaded_count} new records into Snowflake (skipped {len(records) - loaded_count} duplicates)")

    finally:
        cursor.close()
        conn.close()

What does this code do?

These three functions handle the Transform and Load stages:

• transform_states() maps each state vector array to a dictionary with named fields. The API returns aircraft data as a flat list (e.g. index 0 is icao24, index 6 is latitude). This function makes the data readable and usable.
• get_snowflake_connection() creates a connection to Snowflake using the credentials from your .env file.
• load_to_snowflake() creates the RAW.flight_states table if it does not exist, then inserts each record using an idempotent pattern.

What is idempotent loading?

The INSERT ... WHERE NOT EXISTS pattern ensures that running the script multiple times never creates duplicate rows. It checks whether a record with the same icao24 and api_timestamp already exists before inserting.

This is critical for production pipelines. If your script crashes halfway through a load, you can safely re-run it without corrupting your data.

• Save ingest.py.

You still cannot run the script yet. The final piece is the main() function that orchestrates everything together.

Run the pipeline and verify the data

The orchestrator function ties every stage together: authenticate, fetch, transform, load. Once this is in place, you can run the pipeline end-to-end and see real flight data land in Snowflake.

• Add the main() function and entry point at the bottom of ingest.py:

def main():
    """Run the full ingestion pipeline."""
    logger.info("Starting flight data ingestion pipeline")
    start_time = time.time()

    token = get_opensky_token()
    raw_data = fetch_flight_data(token)
    records = transform_states(raw_data)
    load_to_snowflake(records)

    elapsed = time.time() - start_time
    logger.info(f"Pipeline completed in {elapsed:.2f} seconds")


if __name__ == "__main__":
    main()

What does this code do?

The main() function runs the full ELT pipeline in sequence: get a token, fetch data, transform it, and load it into Snowflake. It also measures and logs the total elapsed time.

The if __name__ == "__main__" guard means the pipeline only runs when you execute the file directly, not when it is imported as a module.

• Save ingest.py.
• Run the ingestion pipeline by executing this command in your terminal:

python ingest.py

The script takes 20-60 seconds depending on your internet connection. You should see logging output similar to this:

2026-06-08 14:23:01,123 - INFO - Starting flight data ingestion pipeline
2026-06-08 14:23:02,456 - INFO - No OpenSky credentials found. Using anonymous access.
2026-06-08 14:23:15,789 - INFO - Fetched 6423 flight states
2026-06-08 14:23:45,012 - INFO - Loaded 6423 new records into Snowflake (skipped 0 duplicates)
2026-06-08 14:23:45,013 - INFO - Pipeline completed in 43.89 seconds

What should I see?

The exact number of flight states varies depending on global air traffic at the moment you run the script. Anywhere between 3,000 and 10,000 records is normal.

If you configured your OpenSky credentials in .env, you will see an authenticated token message instead of the anonymous access message. Both paths work.

Seeing an error?

• Connection refused or timeout from OpenSky: the API may be temporarily overloaded. Wait 30 seconds and try again. The retry logic handles transient failures, but if all 3 attempts fail, a persistent outage is likely.
• Snowflake connection error: double-check the values in your .env file. Your SNOWFLAKE_ACCOUNT should be the account identifier from your Snowflake URL (the part before .snowflakecomputing.com).
• ModuleNotFoundError: make sure your virtual environment is activated. Run pip list to confirm snowflake-connector-python and requests are installed.

Help me debug my ingest.py script

• Switch to your Snowflake web UI.
• Open a new worksheet and run this query to verify the data loaded:

SELECT COUNT(*) AS total_rows FROM FLIGHT_TRACKER.RAW.FLIGHT_STATES;

• You should see a row count matching the number from your terminal output (e.g. 6,000+).
• Run a quick sample to see the actual flight data:

SELECT icao24, callsign, origin_country, latitude, longitude, baro_altitude
FROM FLIGHT_TRACKER.RAW.FLIGHT_STATES
LIMIT 10;

You should see rows with real aircraft identifiers, country names, and GPS coordinates.

Why is idempotency important here?

Try running python ingest.py a second time. Notice the log says it loaded 0 new records and skipped all duplicates. Your pipeline is safe to re-run without creating duplicate data.

✔️ Awesome, I've got everything!

Great. Double check you have saved ingest.py and confirmed data is visible in Snowflake.

ⓧ I'd like to double check the full code

Here is the complete ingest.py file:

import os
import logging
import time
from datetime import datetime, timezone

import requests
import snowflake.connector
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

OPENSKY_API_URL = "https://opensky-network.org/api/states/all"
OPENSKY_TOKEN_URL = "https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token"

MAX_RETRIES = 3
RETRY_DELAY = 5


def get_opensky_token():
    """Fetch an OAuth2 access token from the OpenSky Network."""
    client_id = os.getenv("OPENSKY_CLIENT_ID")
    client_secret = os.getenv("OPENSKY_CLIENT_SECRET")

    if not client_id or not client_secret:
        logger.info("No OpenSky credentials found. Using anonymous access.")
        return None

    response = requests.post(
        OPENSKY_TOKEN_URL,
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    response.raise_for_status()
    return response.json()["access_token"]


def fetch_flight_data(token=None):
    """Fetch current flight state vectors from OpenSky Network API."""
    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    for attempt in range(MAX_RETRIES):
        try:
            response = requests.get(
                OPENSKY_API_URL,
                headers=headers,
                timeout=30,
            )
            response.raise_for_status()
            data = response.json()
            logger.info(f"Fetched {len(data.get('states', []))} flight states")
            return data
        except requests.exceptions.RequestException as e:
            logger.warning(f"Attempt {attempt + 1} failed: {e}")
            if attempt < MAX_RETRIES - 1:
                time.sleep(RETRY_DELAY)
            else:
                raise


def transform_states(raw_data):
    """Transform raw API response into structured records."""
    if not raw_data or not raw_data.get("states"):
        return []

    api_time = raw_data["time"]
    records = []

    for state in raw_data["states"]:
        records.append({
            "icao24": state[0],
            "callsign": state[1].strip() if state[1] else None,
            "origin_country": state[2],
            "time_position": state[3],
            "last_contact": state[4],
            "longitude": state[5],
            "latitude": state[6],
            "baro_altitude": state[7],
            "on_ground": state[8],
            "velocity": state[9],
            "true_track": state[10],
            "vertical_rate": state[11],
            "geo_altitude": state[13],
            "squawk": state[14],
            "position_source": state[16],
            "category": state[17] if len(state) > 17 else None,
            "api_timestamp": api_time,
        })

    return records


def get_snowflake_connection():
    """Create a Snowflake connection using environment variables."""
    return snowflake.connector.connect(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema=os.getenv("SNOWFLAKE_SCHEMA"),
    )


def load_to_snowflake(records):
    """Load flight records into the Snowflake raw layer using MERGE for idempotency."""
    if not records:
        logger.warning("No records to load.")
        return

    conn = get_snowflake_connection()
    cursor = conn.cursor()

    try:
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS RAW.flight_states (
                icao24 VARCHAR(6),
                callsign VARCHAR(8),
                origin_country VARCHAR(100),
                time_position INTEGER,
                last_contact INTEGER,
                longitude FLOAT,
                latitude FLOAT,
                baro_altitude FLOAT,
                on_ground BOOLEAN,
                velocity FLOAT,
                true_track FLOAT,
                vertical_rate FLOAT,
                geo_altitude FLOAT,
                squawk VARCHAR(4),
                position_source INTEGER,
                category INTEGER,
                api_timestamp INTEGER,
                loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
                PRIMARY KEY (icao24, api_timestamp)
            )
        """)

        insert_sql = """
            INSERT INTO RAW.flight_states (
                icao24, callsign, origin_country, time_position,
                last_contact, longitude, latitude, baro_altitude,
                on_ground, velocity, true_track, vertical_rate,
                geo_altitude, squawk, position_source, category,
                api_timestamp
            )
            SELECT %(icao24)s, %(callsign)s, %(origin_country)s, %(time_position)s,
                   %(last_contact)s, %(longitude)s, %(latitude)s, %(baro_altitude)s,
                   %(on_ground)s, %(velocity)s, %(true_track)s, %(vertical_rate)s,
                   %(geo_altitude)s, %(squawk)s, %(position_source)s, %(category)s,
                   %(api_timestamp)s
            WHERE NOT EXISTS (
                SELECT 1 FROM RAW.flight_states
                WHERE icao24 = %(icao24)s AND api_timestamp = %(api_timestamp)s
            )
        """

        loaded_count = 0
        for record in records:
            cursor.execute(insert_sql, record)
            loaded_count += cursor.rowcount

        conn.commit()
        logger.info(f"Loaded {loaded_count} new records into Snowflake (skipped {len(records) - loaded_count} duplicates)")

    finally:
        cursor.close()
        conn.close()


def main():
    """Run the full ingestion pipeline."""
    logger.info("Starting flight data ingestion pipeline")
    start_time = time.time()

    token = get_opensky_token()
    raw_data = fetch_flight_data(token)
    records = transform_states(raw_data)
    load_to_snowflake(records)

    elapsed = time.time() - start_time
    logger.info(f"Pipeline completed in {elapsed:.2f} seconds")


if __name__ == "__main__":
    main()

Your raw data layer is loaded with live flight data from around the world. Next up, you will initialize a dbt project and transform this raw data into a clean, analytics-ready star schema.

Initialize the dbt Project and Configure the Connection

Your raw flight data is sitting in Snowflake, ready to be transformed. But raw data is messy and hard to query directly for analytics.

In this step, you'll scaffold a dbt project that connects to your Snowflake warehouse. dbt (data build tool) lets you write SQL transformations as modular, testable models instead of brittle ad-hoc scripts.

In this step, get ready to:

• Scaffold a new dbt project with the standard directory structure.
• Configure the connection profile to point dbt at your Snowflake warehouse.
• Create a source definition so dbt knows where your raw flight data lives.

Run dbt init to scaffold the project

The dbt init command creates a new project folder with all the directories and configuration files dbt needs. It also generates a starter profiles.yml that stores your database connection credentials.

• Scaffold the dbt project by running this command in your terminal:

dbt init flight_tracker_dbt

What does this command do?

dbt init flight_tracker_dbt creates a new folder called flight_tracker_dbt with the standard dbt directory layout: models/, tests/, macros/, and a dbt_project.yml configuration file.

It also writes a connection profile to ~/.dbt/profiles.yml on your local machine so dbt knows how to reach your database.

• When prompted to choose a database adapter, select snowflake.
• The wizard will ask for your Snowflake connection details. Enter these values when prompted:

• account: your Snowflake account identifier (the same value from your .env file's SNOWFLAKE_ACCOUNT field).
• user: flight_pipeline_user.
• password: the password you set for flight_pipeline_user.
• role: press Enter to leave blank (your user's default role will be used).
• warehouse: FLIGHT_WH.
• database: FLIGHT_TRACKER.
• schema: RAW.
• threads: 4.

• Confirm the files were created by running:

ls flight_tracker_dbt

You should see folders like models, tests, macros, and a dbt_project.yml file in the output.

Command not found?

If you see a "command not found" error for dbt, make sure your Python virtual environment is activated. Run source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows) and try again.

help me troubleshoot why dbt init is not found after installing dbt-snowflake in my virtual environment

Configure dbt_project.yml and profiles.yml

The dbt init command generated a default dbt_project.yml. You need to update it to define how your staging and marts models should be materialized.

• Open flight_tracker_dbt/dbt_project.yml in VS Code.
• Replace the entire contents with the following configuration:

name: 'flight_tracker_dbt'
version: '1.0.0'

profile: 'flight_tracker_dbt'

model-paths: ["models"]
test-paths: ["tests"]
analysis-paths: ["analyses"]
macro-paths: ["macros"]

clean-targets:
  - "target"
  - "dbt_packages"

models:
  flight_tracker_dbt:
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: marts

What does this configuration do?

The profile field tells dbt which connection profile to use from ~/.dbt/profiles.yml. The models section configures different materialization strategies per folder.

Staging models are materialized as views because they're lightweight cleaning layers that don't need to store data. Marts models are materialized as tables because they're the final analytics layer that dashboards query directly.

• Save dbt_project.yml.

Now verify that your profiles.yml is correct. This file lives at ~/.dbt/profiles.yml and was created by dbt init.

• Open your profiles file by running:

cat ~/.dbt/profiles.yml

• Verify the output matches this structure (with your own account identifier and password):

flight_tracker_dbt:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: your-account-identifier
      user: flight_pipeline_user
      password: your-password-here
      role: ""
      warehouse: FLIGHT_WH
      database: FLIGHT_TRACKER
      schema: RAW
      threads: 4

Why is the profile stored outside the project?

The profiles.yml file lives in your home directory (~/.dbt/) rather than inside the project folder. This keeps credentials out of version control so you never accidentally commit passwords to Git.

• If any values are incorrect, open the file in VS Code and fix them. Use code ~/.dbt/profiles.yml to open it.

Now validate that dbt can connect to Snowflake using your profile.

• Move into the dbt project directory by running:

cd flight_tracker_dbt

• Validate the connection by running:

dbt debug

You should see a series of checks pass, ending with Connection test: OK connection ok.

Connection test failed?

If dbt debug reports a failed connection, double-check these common causes:

• Your account value in profiles.yml must match the identifier from your Snowflake trial URL (e.g. abc12345.us-east-1).
• The password must match what you set when creating flight_pipeline_user.
• Your Snowflake warehouse FLIGHT_WH must be running. Resume it in the Snowflake web UI if it was suspended.

help me debug why dbt debug is failing to connect to my Snowflake account

Create source definitions for the raw flight data

dbt needs to know where your raw data lives before it can transform it. A source definition maps your Snowflake tables to names that dbt models can reference with the {{ source() }} function.

• Create the staging models directory by running:

mkdir -p models/staging

• Create a new file called models/staging/schema.yml and add the following source definition:

version: 2

sources:
  - name: raw
    database: FLIGHT_TRACKER
    schema: RAW
    tables:
      - name: flight_states
        description: "Raw flight state vectors from OpenSky Network API"

models:
  - name: stg_flight_states
    description: "Cleaned and validated flight state vectors"
    columns:
      - name: icao24
        description: "Unique ICAO 24-bit transponder address (hex)"
        tests:
          - not_null
      - name: latitude
        tests:
          - not_null
      - name: longitude
        tests:
          - not_null

What does this file do?

The sources block tells dbt about the RAW.flight_states table you loaded in the previous step. In your SQL models, you can now reference it as {{ source('raw', 'flight_states') }} instead of hardcoding the table path.

The models block pre-declares the staging model you'll build next, along with not_null tests on critical columns. These tests catch bad data before it reaches your analytics layer.

• Save schema.yml.
• Confirm the file exists by running:

ls models/staging/

You should see schema.yml in the output.

Directory not found?

Make sure you are inside the flight_tracker_dbt directory before running mkdir. Run pwd to check your current location. If you're in the parent project folder, run cd flight_tracker_dbt first.

help me figure out why my dbt models/staging directory is not being created in the right place

✔️ Awesome, I've got everything!

Great. Double check you've saved both files.

ⓧ I'd like to double check the full code

Here is the full dbt_project.yml:

name: 'flight_tracker_dbt'
version: '1.0.0'

profile: 'flight_tracker_dbt'

model-paths: ["models"]
test-paths: ["tests"]
analysis-paths: ["analyses"]
macro-paths: ["macros"]

clean-targets:
  - "target"
  - "dbt_packages"

models:
  flight_tracker_dbt:
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: marts

And here is the full models/staging/schema.yml:

version: 2

sources:
  - name: raw
    database: FLIGHT_TRACKER
    schema: RAW
    tables:
      - name: flight_states
        description: "Raw flight state vectors from OpenSky Network API"

models:
  - name: stg_flight_states
    description: "Cleaned and validated flight state vectors"
    columns:
      - name: icao24
        description: "Unique ICAO 24-bit transponder address (hex)"
        tests:
          - not_null
      - name: latitude
        tests:
          - not_null
      - name: longitude
        tests:
          - not_null

Your dbt project is initialized, connected to Snowflake, and aware of the raw flight data. Next up, you'll write SQL models that clean and standardize this raw data into a proper staging layer.

Build Staging Models

Your raw flight data is sitting in Snowflake, but it's messy. Some records have null coordinates. Timestamps are stored as raw Unix integers. Column names are whatever the API gave you.

Staging models are the first transformation layer in the ELT pattern. They sit between your raw data and your analytics-ready tables, cleaning and standardizing everything so downstream models can trust the data they receive.

In this step, get ready to:

• Create a staging SQL model that cleans and filters raw flight data.
• Run dbt run to materialize the staging view in Snowflake.
• Verify the cleaned data appears in the STAGING schema.

Create the staging SQL model

In dbt, a staging model is a SQL file that reads from a raw source and outputs a cleaned version. Your staging model will do three things: select all columns from the raw table, convert the Unix timestamp into a proper UTC timestamp, and filter out any records with missing critical fields.

• In VS Code, create a new file called stg_flight_states.sql inside the flight_tracker_dbt/models/staging/ folder.
• Paste the following SQL into the file:

with source as (
    select * from {{ source('raw', 'flight_states') }}
),

cleaned as (
    select
        icao24,
        callsign,
        origin_country,
        time_position,
        last_contact,
        longitude,
        latitude,
        baro_altitude,
        on_ground,
        velocity,
        true_track,
        vertical_rate,
        geo_altitude,
        squawk,
        position_source,
        category,
        api_timestamp,
        -- Convert Unix timestamp to a readable UTC timestamp
        to_timestamp(api_timestamp) as api_timestamp_utc,
        loaded_at
    from source
    where
        -- Filter out records with missing critical fields
        icao24 is not null
        and latitude is not null
        and longitude is not null
        and last_contact is not null
)

select * from cleaned

What does this SQL do?

This model uses two CTEs (Common Table Expressions) to organize the transformation:

• source pulls everything from the raw flight_states table using the {{ source() }} function you defined in schema.yml.
• cleaned selects all columns, adds a human-readable api_timestamp_utc column, and filters out rows where the aircraft ID, position, or last contact time is null.
• The final select * from cleaned outputs the cleaned data as the model's result.

Why filter out null positions?

Aircraft sometimes transmit partial state vectors. A flight with no latitude or longitude cannot be plotted on a map or used in geographic analysis. By filtering these at the staging layer, every downstream model is guaranteed to have usable position data.

• Save stg_flight_states.sql.

What about the schema.yml model definition?

Your schema.yml already has the stg_flight_states model definition with not_null tests on icao24, latitude, and longitude from the previous step. dbt will match the SQL file you just created to that model definition automatically by name.

Run dbt to materialize the staging model

With the SQL file saved, dbt can compile and execute it against Snowflake. Because your dbt_project.yml sets +materialized: view for the staging folder, dbt will create a view (not a table) in Snowflake.

• In your terminal, make sure you are inside the flight_tracker_dbt directory.
• Run dbt to build your staging model:

dbt run

What does this command do?

dbt run compiles your SQL model (replacing Jinja templates like {{ source() }} with actual table references), then executes the compiled SQL against Snowflake. For a view materialization, it runs a CREATE VIEW statement.

• You should see output indicating a successful run with a message like 1 of 1 OK created sql view model.

Seeing a compilation or database error?

• Make sure you are running the command from inside the flight_tracker_dbt directory, not the parent project folder.
• Check that your schema.yml source name (raw) and table name (flight_states) match what your SQL file references in {{ source('raw', 'flight_states') }}.
• Verify your virtual warehouse is not suspended. In the Snowflake web UI, check that FLIGHT_WH is running.

Help me debug my dbt run error for the staging model.

Verify the staging view in Snowflake

The terminal output says the model was created, but you should always verify directly in your data warehouse.

• Switch back to the Snowflake web UI.
• Open a new worksheet and run this query to confirm the view exists and contains cleaned data:

SELECT COUNT(*) AS total_rows
FROM FLIGHT_TRACKER.STAGING.STG_FLIGHT_STATES;

• You should see a row count. This number will be slightly lower than the count in RAW.flight_states because the staging model filtered out records with null positions.
• Run this query to confirm the api_timestamp_utc column was created correctly:

SELECT icao24, api_timestamp, api_timestamp_utc
FROM FLIGHT_TRACKER.STAGING.STG_FLIGHT_STATES
LIMIT 5;

• You should see the raw integer api_timestamp alongside a human-readable api_timestamp_utc value (a proper date-time format).

View not found in the STAGING schema?

• Make sure your dbt_project.yml has +schema: staging under the staging: key. Without this, dbt creates the view in the default schema from your profile.
• Check that the STAGING schema exists in the FLIGHT_TRACKER database. You created it in an earlier step.

Help me find my dbt staging view in Snowflake.

✔️ Awesome, I've got everything!

Great. Double check you have saved stg_flight_states.sql and confirmed the view in Snowflake.

ⓧ I'd like to double check the full code

Here is the complete stg_flight_states.sql file:

with source as (
    select * from {{ source('raw', 'flight_states') }}
),

cleaned as (
    select
        icao24,
        callsign,
        origin_country,
        time_position,
        last_contact,
        longitude,
        latitude,
        baro_altitude,
        on_ground,
        velocity,
        true_track,
        vertical_rate,
        geo_altitude,
        squawk,
        position_source,
        category,
        api_timestamp,
        to_timestamp(api_timestamp) as api_timestamp_utc,
        loaded_at
    from source
    where
        icao24 is not null
        and latitude is not null
        and longitude is not null
        and last_contact is not null
)

select * from cleaned

Your raw data now has a clean, reliable layer sitting on top of it. Next up, you will build dimension and fact tables that turn this staged data into an analytics-ready star schema.

Build the Star Schema

Your staging layer gives you clean, validated flight data. Now you need to shape that data into a structure optimized for analytics queries.

In data warehousing, a star schema organizes data into dimension tables (the "who" and "what") and fact tables (the measurable events). In this step, you will build three mart models that transform your staging data into this analytics-ready shape.

In this step, get ready to:

• Create dimension models for aircraft and countries.
• Create a fact table for flight position observations.
• Run dbt run to materialize the full star schema in Snowflake.

Create the aircraft dimension model

A dimension table captures the descriptive attributes of entities. Your aircraft dimension will hold one row per unique aircraft, showing its most recent origin country and size category.

• In VS Code, create a new folder called marts inside flight_tracker_dbt/models/.
• Create a new file called dim_aircraft.sql inside the marts folder, then paste the following code:

with aircraft as (
    select distinct
        icao24,
        first_value(origin_country) over (
            partition by icao24
            order by api_timestamp desc
        ) as origin_country,
        first_value(category) over (
            partition by icao24
            order by api_timestamp desc
        ) as category
    from {{ ref('stg_flight_states') }}
)

select
    icao24,
    origin_country,
    category,
    case category
        when 0 then 'No information'
        when 1 then 'No ADS-B category'
        when 2 then 'Light (< 15500 lbs)'
        when 3 then 'Small (15500-75000 lbs)'
        when 4 then 'Large (75000-300000 lbs)'
        when 5 then 'High Vortex Large'
        when 6 then 'Heavy (> 300000 lbs)'
        when 7 then 'High Performance'
        when 8 then 'Rotorcraft'
        when 9 then 'Glider/Sailplane'
        when 14 then 'UAV'
        else 'Other'
    end as category_name
from aircraft
qualify row_number() over (partition by icao24 order by icao24) = 1

What does this model do?

This model builds a deduplicated aircraft lookup table from your staging data.

• The first_value window function grabs the most recent origin_country and category for each aircraft, ordered by api_timestamp desc.
• The CASE statement translates numeric category codes into human-readable labels like "Heavy" or "Rotorcraft".
• The QUALIFY clause deduplicates the output to exactly one row per icao24 address.

• Save dim_aircraft.sql.

You will not see a change yet. The next model builds on this one, and you will run them all together at the end of this step.

Create the country dimension and fact models

Your country dimension aggregates flight activity per country. The fact table captures every individual position observation with its metrics.

• Create a new file called dim_country.sql inside the marts folder, then paste the following code:

select
    origin_country,
    count(distinct icao24) as total_aircraft,
    count(*) as total_observations
from {{ ref('stg_flight_states') }}
group by origin_country

What does this model do?

This model creates one row per country, counting how many distinct aircraft and total observations came from that country. It gives you a quick way to answer "which countries have the most flights?" without scanning the full fact table.

• Save dim_country.sql.
• Create a new file called fact_flight_positions.sql inside the marts folder, then paste the following code:

select
    icao24,
    callsign,
    origin_country,
    latitude,
    longitude,
    baro_altitude,
    geo_altitude,
    velocity,
    true_track,
    vertical_rate,
    on_ground,
    position_source,
    api_timestamp,
    api_timestamp_utc,
    loaded_at
from {{ ref('stg_flight_states') }}

What does the fact table do?

The fact table selects every cleaned flight position from staging with all its measurable metrics: altitude, velocity, vertical rate, and geographic coordinates. This is the table your dashboard will query most heavily.

Unlike the dimension tables which summarize or deduplicate, the fact table preserves the full grain of the data. Every observation is a row.

• Save fact_flight_positions.sql.

Why separate dimensions from facts?

Dimension tables hold descriptive attributes (country names, aircraft categories) while fact tables hold measurable events (positions, speeds, altitudes). Keeping them separate lets you update dimensions without touching the much larger fact table.

This separation also makes queries faster. When your dashboard wants "flights by country", it reads the small dim_country table instead of scanning millions of position records.

Add schema definitions and build the full DAG

Before running your models, you need a schema file that tells dbt about your mart models and their data quality expectations.

• Create a new file called schema.yml inside the marts folder, then paste the following:

version: 2

models:
  - name: dim_aircraft
    description: "Dimension table of unique aircraft with their attributes"
    columns:
      - name: icao24
        description: "Unique aircraft transponder address"
        tests:
          - unique
          - not_null

  - name: dim_country
    description: "Dimension table of countries with flight activity metrics"
    columns:
      - name: origin_country
        tests:
          - unique
          - not_null

  - name: fact_flight_positions
    description: "Fact table of flight position observations"
    columns:
      - name: icao24
        tests:
          - not_null
      - name: api_timestamp
        tests:
          - not_null

What does this schema file do?

This YAML file registers your mart models with dbt and defines data quality tests. The unique test ensures no duplicate rows on the specified column. The not_null test ensures no missing values.

These tests run automatically when you execute dbt test later. They catch data quality issues before bad data reaches your dashboard.

• Save schema.yml.
• In your terminal, make sure you are inside the flight_tracker_dbt directory. Build all your models by running:

dbt run

• You should see output showing each model being built. Look for a summary line confirming all models passed with no errors.
• Switch to the Snowflake web UI. Navigate to **Data** in the left sidebar, expand **FLIGHT_TRACKER**, then expand the **MARTS** schema.
• You should see three tables: DIM_AIRCRAFT, DIM_COUNTRY, and FACT_FLIGHT_POSITIONS.

Models not appearing in Snowflake?

• Check that your terminal is inside the flight_tracker_dbt directory when running dbt run.
• Verify that your dbt_project.yml has the marts section with +materialized: table and +schema: marts.
• If you see a "relation not found" error for stg_flight_states, run dbt run again. dbt resolves the dependency graph automatically and will build staging first.

Why aren't my dbt mart models showing up in Snowflake after running dbt run?

✔️ Awesome, I've got everything!

Great. Double check that you have saved all four files in your marts folder.

ⓧ I'd like to double check the full code

Here are all four files in flight_tracker_dbt/models/marts/:

with aircraft as (
    select distinct
        icao24,
        first_value(origin_country) over (
            partition by icao24
            order by api_timestamp desc
        ) as origin_country,
        first_value(category) over (
            partition by icao24
            order by api_timestamp desc
        ) as category
    from {{ ref('stg_flight_states') }}
)

select
    icao24,
    origin_country,
    category,
    case category
        when 0 then 'No information'
        when 1 then 'No ADS-B category'
        when 2 then 'Light (< 15500 lbs)'
        when 3 then 'Small (15500-75000 lbs)'
        when 4 then 'Large (75000-300000 lbs)'
        when 5 then 'High Vortex Large'
        when 6 then 'Heavy (> 300000 lbs)'
        when 7 then 'High Performance'
        when 8 then 'Rotorcraft'
        when 9 then 'Glider/Sailplane'
        when 14 then 'UAV'
        else 'Other'
    end as category_name
from aircraft
qualify row_number() over (partition by icao24 order by icao24) = 1

select
    origin_country,
    count(distinct icao24) as total_aircraft,
    count(*) as total_observations
from {{ ref('stg_flight_states') }}
group by origin_country

select
    icao24,
    callsign,
    origin_country,
    latitude,
    longitude,
    baro_altitude,
    geo_altitude,
    velocity,
    true_track,
    vertical_rate,
    on_ground,
    position_source,
    api_timestamp,
    api_timestamp_utc,
    loaded_at
from {{ ref('stg_flight_states') }}

version: 2

models:
  - name: dim_aircraft
    description: "Dimension table of unique aircraft with their attributes"
    columns:
      - name: icao24
        description: "Unique aircraft transponder address"
        tests:
          - unique
          - not_null

  - name: dim_country
    description: "Dimension table of countries with flight activity metrics"
    columns:
      - name: origin_country
        tests:
          - unique
          - not_null

  - name: fact_flight_positions
    description: "Fact table of flight position observations"
    columns:
      - name: icao24
        tests:
          - not_null
      - name: api_timestamp
        tests:
          - not_null

Your star schema is live in Snowflake. Next up, you will add data quality tests and generate documentation to make your pipeline production-ready.

Add Data Quality Tests and Documentation

Your star schema is built and materialized in Snowflake. But how do you know the data flowing through your pipeline is correct?

Data quality tests are automated assertions that validate your data meets expectations every time the pipeline runs. In this step, you'll run the schema tests you've already defined, add a custom freshness test, and generate browsable documentation for your entire project.

In this step, get ready to:

• Run your existing schema tests with dbt test.
• Write a custom singular test that validates data freshness.
• Generate and serve dbt documentation locally.

Run the existing schema tests

Your schema.yml files already define several generic tests across your staging and marts models. These tests check constraints like not_null and unique on key columns.

• Make sure you're inside the flight_tracker_dbt directory in your terminal.
• Run all tests in your project by running this command:

dbt test

What does dbt test do?

dbt constructs a SQL query for each test you've defined in your schema.yml files. Each query tries to find rows that violate the assertion.

If the query returns zero rows, the test passes. For example, a not_null test runs a query like SELECT * FROM model WHERE column IS NULL. Zero results means no nulls exist, so the test passes.

You should see each test run and pass, with a final summary line showing all tests passed.

What should I see?

dbt prints each test as it runs. You'll see lines like PASS not_null_dim_aircraft_icao24 for each test. At the end, a summary shows the total count.

The final line looks like PASS=7 WARN=0 ERROR=0 SKIP=0 TOTAL=7 (your exact count depends on your schema definitions).

Seeing test failures?

If a unique test fails on dim_aircraft.icao24, check that you ran dbt run in the previous step to rebuild all models with the latest SQL (including the qualify deduplication clause).

If you see a connection error, verify your profiles.yml credentials are correct by running dbt debug first.

Still stuck? Help me debug my failing dbt tests

Add a custom singular test for data freshness

Generic tests like not_null and unique check column-level constraints. But what about pipeline-level concerns like data freshness? A singular test is a custom SQL file that returns failing rows if something is wrong.

• Create a new file called test_data_freshness.sql inside the flight_tracker_dbt/tests/ directory.
• Add the following SQL to check that no records are older than 24 hours:

-- Singular test: fail if any staged records have a loaded_at older than 24 hours
-- A non-empty result set means the test fails

select
    loaded_at
from {{ ref('stg_flight_states') }}
where loaded_at < dateadd(hour, -24, current_timestamp())
limit 1

What does this test do?

This query looks for any records in your staging model with a loaded_at timestamp older than 24 hours. If the query returns any rows, that means your pipeline hasn't loaded fresh data recently, and the test fails.

Since you just ran the ingestion script, all your data should be recent and this test should pass. In production, this test would catch situations where the pipeline stopped running without anyone noticing.

• Save test_data_freshness.sql.
• Run dbt test again to confirm the new singular test passes alongside the schema tests:

dbt test

You should see one additional test in the output (your singular test). The final summary should show all tests passing with zero errors.

Freshness test failing?

If the freshness test fails, it means your loaded_at timestamps are older than 24 hours. Re-run your ingestion script (python ingest.py from the project root) to load fresh data, then run dbt run followed by dbt test.

Need help? Why is my freshness test failing even after re-running the ingestion?

Generate and serve dbt documentation

dbt can auto-generate a documentation website from your project's models, descriptions, and test definitions. The site includes a visual DAG (directed acyclic graph) showing how all your models connect.

• Generate the documentation artifacts by running this command:

dbt docs generate

What does dbt docs generate do?

This command compiles your project resources into manifest.json and queries your Snowflake warehouse metadata to produce catalog.json. Together these files power the documentation site, which includes model descriptions, column types, test results, and lineage information.

• Serve the documentation locally by running:

dbt docs serve

What does dbt docs serve do?

This starts a local web server on port 8080 and opens the documentation site in your default browser. You can explore your models, view the full DAG, and see all column descriptions and test results in one place.

• Your browser should open automatically to http://localhost:8080. Explore the site by clicking on models in the left sidebar.
• Click the DAG icon (the green circle with nodes) in the bottom-right corner to see your full model lineage graph showing raw.flight_states flowing through stg_flight_states into your three mart tables.
• When you're done exploring, press Ctrl+C in your terminal to stop the docs server.

Browser didn't open automatically?

Manually navigate to http://localhost:8080 in your browser. If that port is in use, stop the server with Ctrl+C and restart with a different port:

dbt docs serve --port 8001

Still having trouble? Help me get dbt docs serve working

Your pipeline now has automated quality checks and browsable documentation. Next up, you'll build an interactive Streamlit dashboard that queries your mart tables and visualizes real-time flight traffic patterns.

Build the Streamlit Dashboard

Your data pipeline is fully operational. Data flows from the OpenSky API into Snowflake, gets cleaned in staging, and lands in your star schema marts.

The missing piece is visibility. In this step, you will build an interactive Streamlit dashboard that connects directly to your Snowflake marts and visualizes the flight data in real time.

In this step, get ready to:

• Create a Streamlit app with cached Snowflake connection helpers.
• Build visualizations showing flight traffic by country, altitude distribution, and a live aircraft map.
• Run the dashboard locally and verify the full end-to-end pipeline.

Create the Streamlit app with a Snowflake connection

Your dashboard needs two foundational pieces: a Snowflake connection and a reusable query function. Streamlit's caching decorators keep the connection alive across page refreshes without reconnecting every time.

• Create a new file called dashboard.py in your project root (the same folder as ingest.py).
• Add the imports, page configuration, and connection helper functions by pasting this code:

import os

import streamlit as st
import pandas as pd
import snowflake.connector
from dotenv import load_dotenv

load_dotenv()

st.set_page_config(page_title="Flight Tracker Dashboard", layout="wide")
st.title("✈️ Flight Tracker Dashboard")


@st.cache_resource
def get_connection():
    return snowflake.connector.connect(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema="MARTS",
    )


@st.cache_data(ttl=300)
def query_snowflake(sql):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(sql)
    columns = [desc[0] for desc in cursor.description]
    data = cursor.fetchall()
    return pd.DataFrame(data, columns=columns)

What does this code do?

• @st.cache_resource keeps the Snowflake connection alive across Streamlit re-renders. Without it, the app would open a new database connection every time you interact with a widget.
• @st.cache_data(ttl=300) caches query results for 5 minutes. This prevents hammering Snowflake with the same query on every page load.
• The schema="MARTS" parameter connects directly to the marts layer where your dimension and fact tables live.
• query_snowflake() returns a pandas DataFrame, which Streamlit chart functions accept natively.

• Save dashboard.py.

You won't see a change yet. The next chunks add the visualizations that use these helpers.

Build the flight visualizations

Each visualization queries a different mart table and renders it with a single Streamlit call. You will add four sections: a country bar chart, an altitude distribution chart, a live aircraft map, and summary metrics.

• Add the flights-by-country bar chart below your query_snowflake function in dashboard.py:

# Flights by Country
st.subheader("🌍 Top 15 Countries by Flight Activity")
country_data = query_snowflake("""
    SELECT ORIGIN_COUNTRY, TOTAL_AIRCRAFT, TOTAL_OBSERVATIONS
    FROM MARTS.DIM_COUNTRY
    ORDER BY TOTAL_OBSERVATIONS DESC
    LIMIT 15
""")
st.bar_chart(country_data.set_index("ORIGIN_COUNTRY")["TOTAL_OBSERVATIONS"])

What does this code do?

• This queries your DIM_COUNTRY dimension table to get the top 15 countries ranked by total flight observations.
• st.bar_chart() renders a bar chart using the ORIGIN_COUNTRY column as the x-axis and TOTAL_OBSERVATIONS as the height.

• Save dashboard.py.
• You won't see the chart until you run the dashboard later in this step. The next chunk adds the altitude chart.

• Add the altitude distribution chart below the flights-by-country section:

# Altitude Distribution
st.subheader("📊 Altitude Distribution of Active Flights")
altitude_data = query_snowflake("""
    SELECT
        CASE
            WHEN BARO_ALTITUDE IS NULL THEN 'Unknown'
            WHEN BARO_ALTITUDE < 1000 THEN '0-1km'
            WHEN BARO_ALTITUDE < 3000 THEN '1-3km'
            WHEN BARO_ALTITUDE < 6000 THEN '3-6km'
            WHEN BARO_ALTITUDE < 9000 THEN '6-9km'
            WHEN BARO_ALTITUDE < 12000 THEN '9-12km'
            ELSE '12km+'
        END AS ALTITUDE_BAND,
        COUNT(*) AS FLIGHT_COUNT
    FROM MARTS.FACT_FLIGHT_POSITIONS
    WHERE ON_GROUND = FALSE
    GROUP BY ALTITUDE_BAND
    ORDER BY FLIGHT_COUNT DESC
""")
st.bar_chart(altitude_data.set_index("ALTITUDE_BAND")["FLIGHT_COUNT"])

What does this code do?

• The CASE statement groups continuous altitude values into human-readable bands (0-1km, 1-3km, etc.).
• WHERE ON_GROUND = FALSE filters out parked aircraft so the chart only shows airborne flights.
• This queries the FACT_FLIGHT_POSITIONS table directly, using SQL to compute the grouping on the fly.

• Save dashboard.py.

Next, add a live map showing aircraft positions around the world.

• Add the aircraft map section below the altitude distribution chart:

# Aircraft Map
st.subheader("🗺️ Live Aircraft Positions")
map_data = query_snowflake("""
    SELECT LATITUDE, LONGITUDE
    FROM MARTS.FACT_FLIGHT_POSITIONS
    WHERE ON_GROUND = FALSE
        AND LATITUDE IS NOT NULL
        AND LONGITUDE IS NOT NULL
    LIMIT 5000
""")
if not map_data.empty:
    st.map(map_data.rename(columns={"LATITUDE": "latitude", "LONGITUDE": "longitude"}))

What does this code do?

• st.map() renders a scatter-plot map using latitude and longitude columns. The column names must be lowercase latitude and longitude, which is why the rename() call converts from Snowflake's uppercase column names.
• LIMIT 5000 caps the points to keep the map responsive. Rendering tens of thousands of markers would slow down the browser.
• The if not map_data.empty guard prevents Streamlit from throwing an error if no data is available yet.

• Save dashboard.py.

The final section adds three headline metrics at the bottom of the dashboard.

• Add the summary metrics section below the aircraft map:

# Summary Metrics
st.subheader("📈 Pipeline Summary")
col1, col2, col3 = st.columns(3)
summary = query_snowflake("""
    SELECT
        COUNT(DISTINCT ICAO24) AS UNIQUE_AIRCRAFT,
        COUNT(*) AS TOTAL_RECORDS,
        COUNT(DISTINCT ORIGIN_COUNTRY) AS COUNTRIES
    FROM MARTS.FACT_FLIGHT_POSITIONS
""")
col1.metric("Unique Aircraft", f"{summary['UNIQUE_AIRCRAFT'].iloc[0]:,}")
col2.metric("Total Records", f"{summary['TOTAL_RECORDS'].iloc[0]:,}")
col3.metric("Countries", f"{summary['COUNTRIES'].iloc[0]:,}")

What does this code do?

• st.columns(3) creates a three-column layout so the metrics sit side by side.
• Each .metric() call renders a large-font KPI card with a label and formatted number. The :, format adds thousands separators for readability.
• This gives stakeholders a quick snapshot of pipeline scale: how many unique aircraft, total records processed, and countries represented.

• Save dashboard.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard.py with all four visualization sections.

ⓧ I'd like to double check the full code

import os

import streamlit as st
import pandas as pd
import snowflake.connector
from dotenv import load_dotenv

load_dotenv()

st.set_page_config(page_title="Flight Tracker Dashboard", layout="wide")
st.title("✈️ Flight Tracker Dashboard")


@st.cache_resource
def get_connection():
    return snowflake.connector.connect(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema="MARTS",
    )


@st.cache_data(ttl=300)
def query_snowflake(sql):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(sql)
    columns = [desc[0] for desc in cursor.description]
    data = cursor.fetchall()
    return pd.DataFrame(data, columns=columns)


# Flights by Country
st.subheader("🌍 Top 15 Countries by Flight Activity")
country_data = query_snowflake("""
    SELECT ORIGIN_COUNTRY, TOTAL_AIRCRAFT, TOTAL_OBSERVATIONS
    FROM MARTS.DIM_COUNTRY
    ORDER BY TOTAL_OBSERVATIONS DESC
    LIMIT 15
""")
st.bar_chart(country_data.set_index("ORIGIN_COUNTRY")["TOTAL_OBSERVATIONS"])

# Altitude Distribution
st.subheader("📊 Altitude Distribution of Active Flights")
altitude_data = query_snowflake("""
    SELECT
        CASE
            WHEN BARO_ALTITUDE IS NULL THEN 'Unknown'
            WHEN BARO_ALTITUDE < 1000 THEN '0-1km'
            WHEN BARO_ALTITUDE < 3000 THEN '1-3km'
            WHEN BARO_ALTITUDE < 6000 THEN '3-6km'
            WHEN BARO_ALTITUDE < 9000 THEN '6-9km'
            WHEN BARO_ALTITUDE < 12000 THEN '9-12km'
            ELSE '12km+'
        END AS ALTITUDE_BAND,
        COUNT(*) AS FLIGHT_COUNT
    FROM MARTS.FACT_FLIGHT_POSITIONS
    WHERE ON_GROUND = FALSE
    GROUP BY ALTITUDE_BAND
    ORDER BY FLIGHT_COUNT DESC
""")
st.bar_chart(altitude_data.set_index("ALTITUDE_BAND")["FLIGHT_COUNT"])

# Aircraft Map
st.subheader("🗺️ Live Aircraft Positions")
map_data = query_snowflake("""
    SELECT LATITUDE, LONGITUDE
    FROM MARTS.FACT_FLIGHT_POSITIONS
    WHERE ON_GROUND = FALSE
        AND LATITUDE IS NOT NULL
        AND LONGITUDE IS NOT NULL
    LIMIT 5000
""")
if not map_data.empty:
    st.map(map_data.rename(columns={"LATITUDE": "latitude", "LONGITUDE": "longitude"}))

# Summary Metrics
st.subheader("📈 Pipeline Summary")
col1, col2, col3 = st.columns(3)
summary = query_snowflake("""
    SELECT
        COUNT(DISTINCT ICAO24) AS UNIQUE_AIRCRAFT,
        COUNT(*) AS TOTAL_RECORDS,
        COUNT(DISTINCT ORIGIN_COUNTRY) AS COUNTRIES
    FROM MARTS.FACT_FLIGHT_POSITIONS
""")
col1.metric("Unique Aircraft", f"{summary['UNIQUE_AIRCRAFT'].iloc[0]:,}")
col2.metric("Total Records", f"{summary['TOTAL_RECORDS'].iloc[0]:,}")
col3.metric("Countries", f"{summary['COUNTRIES'].iloc[0]:,}")

Run the dashboard and verify the end-to-end pipeline

With the dashboard code complete, it is time to launch it and confirm that the entire pipeline works from ingestion to visualization.

• Launch the Streamlit dashboard by running this command in your terminal:

streamlit run dashboard.py

• Your browser should open automatically to http://localhost:8501. You should see four sections: a country bar chart, an altitude distribution chart, an aircraft map with dots scattered globally, and three summary metric cards.

What should I see?

The country bar chart shows countries like the United States, China, and Germany with the tallest bars. The altitude chart peaks in the 9-12km band (typical cruising altitude). The map shows clusters along major flight corridors.

The summary metrics at the bottom show the total unique aircraft, total records, and number of countries in your dataset. These numbers depend on when you ran your ingestion script.

Seeing a connection error?

• Make sure your virtual environment is activated. The streamlit command must come from the same environment where you installed the packages.
• Verify that your .env file is in the same directory as dashboard.py. The load_dotenv() call looks for .env in the current working directory.
• Check that your Snowflake warehouse is not suspended. Return to the Snowflake web UI and confirm FLIGHT_WH shows a running state.

Help me troubleshoot my Streamlit dashboard connection to Snowflake.

Why Streamlit for this dashboard?

Streamlit is Python-native, so it fits naturally into a Python data pipeline without switching languages. It is also owned by Snowflake, which means first-class integration with Snowflake connectors and caching.

Alternatives like Looker Studio do not connect natively to Snowflake. With Streamlit, your entire stack stays in Python and deploys free on Streamlit Cloud if you choose to share it later.

Your flight tracker pipeline is complete. Data flows from the OpenSky API through Snowflake into a live dashboard. Next up, the Secret Mission challenges you to make this pipeline run continuously with incremental loading.

Secret mission

Add Incremental Loading and Scheduling

Your pipeline works, but it rebuilds the entire fact table on every run. Can you make it smarter? Convert to incremental loading so only new flight data gets processed, add watermark tracking, and schedule the pipeline to run automatically every 5 minutes.

Clean Up Your Resources

Clean Up Your Resources

Your Snowflake trial account has $400 in credits and expires after 30 days. The virtual warehouse consumes credits only while it is running, so decide whether to keep your resources active, pause them, or delete everything.

Cost warning

The FLIGHT_WH warehouse uses Snowflake credits whenever it is in a running state. If you also have a scheduler running every 5 minutes, it will keep the warehouse active continuously. Suspend the warehouse or disable your scheduler if you want to preserve credits.

Resources you used:

• Snowflake FLIGHT_TRACKER database (RAW, STAGING, MARTS schemas and all tables/views).
• Snowflake FLIGHT_WH virtual warehouse.
• Snowflake service user (flight_pipeline_user).
• Scheduler (cron job or Python schedule script).
• Local project folder on your Desktop (ingest.py, dashboard.py, flight_tracker_dbt/, .env, virtual environment).
• dbt profile at ~/.dbt/profiles.yml.

Keep everything running

No action needed. Choose this if you are still actively building or want to keep ingesting flight data on a schedule.

• Monitor your credit balance in the Snowflake web UI under Admin > Billing.
• Your trial expires after 30 days regardless of remaining credits.
• Your scheduler will keep the warehouse active and consuming credits. Disable it when you are done experimenting.

Pause - I'll come back to this later

Suspend compute and stop the scheduler to preserve credits. Your data and project files stay intact.

• Disable your scheduler. If you used a cron job, run crontab -e and delete the line that runs your pipeline.
• Suspend your warehouse by running this SQL in the Snowflake web UI:

ALTER WAREHOUSE FLIGHT_WH SUSPEND;

• The warehouse will auto-resume next time you run a query or your pipeline.
• Deactivate your Python virtual environment by running deactivate in your terminal.

Delete - I don't want to use this again

Remove all Snowflake resources and local project files. This is irreversible.

• Disable your scheduler first. If you used a cron job, run crontab -e and delete the pipeline line. If you used a Python schedule script, stop the running process.
• In the Snowflake web UI, open a new worksheet and run these SQL commands to drop all resources:

DROP DATABASE IF EXISTS FLIGHT_TRACKER;
DROP WAREHOUSE IF EXISTS FLIGHT_WH;
DROP USER IF EXISTS flight_pipeline_user;

• Remove the dbt profile entry by deleting the ~/.dbt/profiles.yml file (or just the flight_tracker_dbt section if you have other dbt projects).
• Delete your local project folder. On macOS/Linux:

cd ~/Desktop
rm -rf flight-tracker

• On Windows:

cd ~\Desktop
rmdir /s /q flight-tracker

• Your Snowflake trial account will expire automatically after 30 days. No further action is needed to close it.

Nice Work!

Nice Work!

You just built a complete, production-grade flight tracking ELT pipeline from scratch. Real-time data flows from the OpenSky Network API through Snowflake and dbt, all the way to a live dashboard.

You've learned how to:

• Build a Python ingestion script with retry logic, logging, and idempotent loads that extracts live flight data from the OpenSky Network API and loads it into Snowflake.
• Transform raw data into an analytics-ready star schema using dbt, with staging models, dimension tables, a fact table, and automated data quality tests.
• Create an interactive Streamlit dashboard that connects directly to your Snowflake warehouse and visualizes flight traffic by country, altitude distributions, and live aircraft positions.
• Implement incremental loading and automated scheduling so your pipeline continuously ingests new flight data without creating duplicates.

Ready to quiz yourself?