Build a Local ETL Quality Pipeline

Build a local DuckDB ETL pipeline with SQL quality checks and reconciliation.

Introduction

30 Second Summary

Interviewers often ask how you would catch bad data before you have access to a company's systems. A spoken answer can be hard to prove under time pressure.

In this project, you will build a local ETL quality pipeline in Python over official Colorado Department of Transportation road data. The finished demo shows exactly how your pipeline responds when the same batch arrives twice.

What You'll Build

Your final demo pairs passing terminal checks with a county-level quality report that you can open in PyCharm during the recruiter conversation.

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

  • A working local load of official CDOT road records in DuckDB. The terminal prints a positive raw row count after the first batch.
  • A duplicate replay audit that prints [FAIL] when repeated OBJECTID values appear. That result gives you a concrete SQL reconciliation story for the interview.
  • An idempotent warehouse workflow whose raw, staging, and mart counts stay stable across reruns. Its Markdown report maps the local design to Azure Data Factory, Azure SQL, SSIS, and Snowflake.
  • Secret Mission: Add a quarantine path for invalid road identifiers and segment lengths. Prove every raw record reaches either staging or quarantine.

Are there any prerequisites?

No cloud account is required, but you need internet access for the public dataset and DuckDB 1.5.5. Step 1 checks that your existing Python version is 3.10 or newer.

Before We Start

Before any setup begins, this checkpoint locks in what you are building for tomorrow's Data Engineer interview. You are committing to a recruiter-ready local Python and DuckDB ETL quality pipeline over official CDOT Local Roads data.

Set Up the Local Data Engineering Workspace

Tomorrow's recruiter demo needs a setup that behaves the same way each time you run it. A clean virtual environment keeps package conflicts out of your walkthrough.

Python provides the runtime. DuckDB supplies the local analytical database.

Git records this clean starting point. PyCharm keeps your project files beside the terminal you use to run them.

In this step, get ready to:
  • Verify Python 3.10 or newer.
  • Create the colorado-roads-pipeline project with an isolated environment.
  • Prove a pinned DuckDB installation with a smoke test.
Verify the Python version

DuckDB 1.5.5 requires Python 3.10 or newer. Checking first prevents an incompatible interpreter from becoming the foundation of your project.

  • Press Cmd+Space to open Spotlight.
  • Type Terminal into the search field.
  • Press Return to open Terminal.
  • Check your installed Python version by running this command:
python3 --version

What does this command do?

The command asks your default Python 3 interpreter to print its version. The first two numbers determine whether it meets the project's minimum.

✔️ I see version 3.10 or higher

Good, your interpreter can support the pinned DuckDB package. You can use python3 throughout the rest of this project.

ⓧ I see an older version

The current interpreter is below the project's minimum. The official stable Python 3.14.7 release gives your virtual environment a supported base.

  • Visit the official Python releases for macOS page.
  • Select Download macOS installer for Python 3.14.7.
  • Open the downloaded installer package.
  • Complete the installer with its default options.
  • Close every Terminal window after the installation completes.
  • Reopen Terminal through Spotlight.
  • Check the updated default version by running:
python3 --version

What should the new check show?

The reopened shell reads the path added by the official installer. The version check should now report Python 3.14.7.

Some Macs can still resolve python3 to an older Apple-controlled interpreter. The official installer also provides a versioned command for the newer release.

  • Check the versioned interpreter by running:
python3.14 --version

Why use the versioned command?

The python3.14 command selects the installed 3.14 interpreter directly. This avoids ambiguity when multiple Python releases coexist.

  • Use python3.14 in place of python3 for the remaining Python commands if this check shows the supported release.

Still seeing an older version?

Confirm that you completed the Python 3.14.7 installer. Reopen Terminal before checking again.

If both version checks still show an older release, help me select the Python 3.14 interpreter on macOS.

ⓧ Command not found

Your shell cannot find a Python 3 interpreter yet. Install the official stable Python 3.14.7 release before creating the project.

  • Visit the official Python releases for macOS page.
  • Select Download macOS installer for Python 3.14.7.
  • Open the downloaded installer package.
  • Complete the installer with its default options.
  • Close every Terminal window after the installation completes.
  • Reopen Terminal through Spotlight.
  • Confirm the installation by running:
python3 --version

What should the check show?

The command should now print Python 3.14.7. That result confirms the interpreter is available to your shell.

Command still unavailable?

Make sure the installer completed before you reopened Terminal. An existing shell session may not see the newly installed interpreter.

If a fresh Terminal session still cannot find Python, help me troubleshoot the Python macOS installation.

Create the isolated project workspace

The project directory gives every script one predictable home. Its .venv directory isolates this project's packages from your other Python work.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type PyCharm into the search field.
  • Press Return (macOS) or Enter (Windows) to open PyCharm.
  • Click New Project on the welcome screen.
  • Choose Pure Python as the project type.
  • Enter ~/Desktop/colorado-roads-pipeline in the Location field.
  • Switch the environment option to Custom environment.
  • Select the Python interpreter that passed your version check.

This menu has several interpreter options, so taking a moment here is normal. The selected interpreter becomes the base used to create .venv.

  • Click Create.

You will see the colorado-roads-pipeline directory in PyCharm's Project tool window. The project now has a fixed location on your Desktop.

  • Click View in the PyCharm menu bar.
  • Select Tool Windows.
  • Select Terminal.

You will see a terminal session with colorado-roads-pipeline as its current directory. Commands run here apply to this project.

  • Initialize the project as a Git repository by running this command:
git init

What does this command do?

The command creates a hidden .git directory inside colorado-roads-pipeline. Git uses that directory to track future changes.

Git reports that it initialized an empty repository. Your project now has local version tracking.

  • Create the isolated environment by running:
python3 -m venv .venv

What does this command create?

Python creates a dedicated interpreter inside .venv. Packages installed after activation stay scoped to this project.

  • Check the Project tool window for the new .venv directory.

You will see .venv beneath the project directory. That folder is the visible proof that the isolated environment exists.

Missing the .venv directory?

Confirm that the PyCharm terminal is inside colorado-roads-pipeline. Check that your selected interpreter meets the minimum version.

If the directory still does not appear, help me create the virtual environment in my PyCharm project.

  • Activate the new environment by running:
source .venv/bin/activate

What does activation change?

Activation updates the current terminal session to use the interpreter inside .venv. Package installation commands now target the isolated environment.

You will see (.venv) near the start of your terminal prompt. The environment is now active.

Activation did not change the prompt?

Confirm that the terminal is inside colorado-roads-pipeline. Check that .venv appears in the Project tool window.

If both locations look correct, help me activate my .venv environment on macOS.

Pin DuckDB and prove the setup

A dependency pin records the exact database package used by the demo. The smoke test then proves that the active interpreter can import that package and execute SQL.

  • Select the colorado-roads-pipeline directory in the Project tool window.
  • Right-click the selected directory.
  • Select New.
  • Select File.
  • Enter requirements.txt as the file name.
  • Add the dependency pin below:
duckdb==1.5.5

Why pin the package version?

The pin tells pip to install DuckDB 1.5.5 every time this project is set up. A fixed dependency gives your recruiter demo a repeatable software baseline.

  • Save requirements.txt.
  • Install the pinned dependency by running:
python3 -m pip install -r requirements.txt

What does this installation command do?

The command asks the active Python interpreter to read requirements.txt. It installs the listed DuckDB release inside .venv.

The terminal completes the installation without reporting a package error. Your isolated environment now contains the database client.

DuckDB did not install?

Check that (.venv) appears in your terminal prompt. Confirm that requirements.txt contains the exact dependency pin.

The installation also needs internet access to download the package. Help me troubleshoot the DuckDB installation.

Generated environments and local data do not belong in version control. The ignore file keeps those reproducible or downloaded artifacts out of future Git staging.

  • Select the colorado-roads-pipeline directory in the Project tool window.
  • Right-click the selected directory.
  • Select New.
  • Select File.
  • Enter .gitignore as the file name.
  • Add the ignore patterns below:
.venv/
__pycache__/
*.pyc
data/*.csv
data/*.duckdb

What do these patterns protect?

  • The .venv/ pattern excludes installed project packages.
  • The Python cache patterns exclude generated bytecode.
  • The data patterns exclude downloaded CSV files.
  • The final pattern excludes the local DuckDB database.
  • Save .gitignore.
  • Check which files Git can see by running:
git status

What does this check prove?

Git lists untracked project files that can be staged later. The ignored .venv directory stays out of that list.

You will see requirements.txt listed as an untracked file. You will not see the .venv directory.

Is .venv still listed by Git?

Confirm that the ignore file is named .gitignore with the dot at the beginning. Check that .venv/ matches the directory name exactly.

If the directory still appears, help me fix my .gitignore patterns.

  • Select the colorado-roads-pipeline directory in the Project tool window.
  • Right-click the selected directory.
  • Select New.
  • Select Python File.
  • Enter smoke_test as the file name.
  • Add the database smoke test below:
import duckdb

connection = duckdb.connect(":memory:")
answer = connection.execute("SELECT 42").fetchone()[0]
print(f"DuckDB ready: {answer}")
connection.close()

What does this smoke test prove?

  • The import duckdb line proves the package is available to the active interpreter.
  • The duckdb.connect(":memory:") call creates a temporary database connection.
  • The query returns 42 through the connection.
  • The final line closes the database connection cleanly.
  • Save smoke_test.py.

Before you run the smoke test, what output would prove that Python can query DuckDB?

  • Run the smoke test from the activated environment with this command:
python3 smoke_test.py

What should the result show?

The script imports DuckDB through .venv. It executes SELECT 42 through a live connection.

You will see DuckDB ready: 42 in the terminal. That output confirms the runtime plus database client work together.

That's the environment hurdle cleared. Your Git-tracked workspace can now execute DuckDB SQL from Python.

Smoke test did not print 42?

Confirm that (.venv) appears in the terminal prompt. Check that smoke_test.py matches the reference code below.

If Python cannot import the package, rerun the requirements installation inside the activated environment. Help me debug the DuckDB smoke test.

✔️ Awesome, I've got everything!

Great, save all three files before moving on. Your workspace now matches the required starting point.

ⓧ I'd like to double check the full code

Compare your ignore rules with the complete .gitignore file below.

.venv/
__pycache__/
*.pyc
data/*.csv
data/*.duckdb

What should match?

Your file should contain these five patterns in this order. Each pattern keeps a generated or local artifact out of Git.

Compare your dependency pin with the complete requirements.txt file below.

duckdb==1.5.5

What should match?

The file should contain one dependency pin. The package name plus version must match exactly.

Compare your smoke test with the complete smoke_test.py file below.

import duckdb

connection = duckdb.connect(":memory:")
answer = connection.execute("SELECT 42").fetchone()[0]
print(f"DuckDB ready: {answer}")
connection.close()

What should match?

The script should import DuckDB before opening the in-memory connection. It should print the fetched value before closing the connection.

Your local workspace is reproducible and ready for data. Next, you will download the official CDOT Local Roads CSV and load the first raw batch.

Load the First Raw Data Batch

The smoke test proved that your DuckDB client can execute SQL. The next proof is a real batch of official road data landing in a local raw layer.

The first loader uses append-only ingestion to produce visible rows quickly. This approach leaves the table vulnerable when the same batch is replayed.

In this step, get ready to:
  • Create a reusable helper that downloads the official CDOT Local Roads data.
  • Build a loader that appends one batch to a local raw table.
  • Run the loader to confirm a positive raw row count.
Create the source download helper

The source helper keeps download details separate from the loader. It also gives later scripts one place to find the CSV file plus the local database.

  • In PyCharm's project sidebar, right-click the colorado-roads-pipeline folder.
  • Choose the option for creating a new Python file.
  • Name the file common.py.

You will see common.py appear beside smoke_test.py in the project sidebar.

  • Paste the source configuration plus download helper into common.py using this code:
import os
import urllib.request

SOURCE_URL = "https://geodata.colorado.gov/api/download/v1/items/f031695aeaeb42a0b86e4887108d03c8/csv?layers=14"
RAW_CSV = "data/local_roads.csv"
DB_PATH = "data/colorado_roads.duckdb"
REPORT_PATH = "reports/quality_report.md"


def ensure_directories():
    os.makedirs("data", exist_ok=True)
    os.makedirs("reports", exist_ok=True)


def download_source():
    ensure_directories()
    if os.path.exists(RAW_CSV):
        print(f"Using existing source file: {RAW_CSV}")
        return

    print("Downloading official CDOT Local Roads data...")
    urllib.request.urlretrieve(SOURCE_URL, RAW_CSV)
    print(f"Downloaded source file: {RAW_CSV}")

What does this helper do?

  • The SOURCE_URL constant holds the official CDOT download address.
  • The RAW_CSV constant points to the downloaded source file.
  • The DB_PATH constant gives every database script the same local destination.
  • The ensure_directories() function creates the folders used by the pipeline.
  • The download_source() function reuses an existing download when one is already present.
  • Save common.py.
  • Confirm common.py remains listed beside smoke_test.py in the project sidebar.

Seeing a syntax warning?

  • Check that both function bodies are indented with four spaces.
  • Check that every quoted file path has matching quotation marks.

Ask for help with the exact file contents: help me find the syntax problem in my common.py source download helper.

✔️ Awesome, I've got everything!

Your saved source helper now contains the shared paths plus download logic.

ⓧ I'd like to double check the full code

import os
import urllib.request

SOURCE_URL = "https://geodata.colorado.gov/api/download/v1/items/f031695aeaeb42a0b86e4887108d03c8/csv?layers=14"
RAW_CSV = "data/local_roads.csv"
DB_PATH = "data/colorado_roads.duckdb"
REPORT_PATH = "reports/quality_report.md"


def ensure_directories():
    os.makedirs("data", exist_ok=True)
    os.makedirs("reports", exist_ok=True)


def download_source():
    ensure_directories()
    if os.path.exists(RAW_CSV):
        print(f"Using existing source file: {RAW_CSV}")
        return

    print("Downloading official CDOT Local Roads data...")
    urllib.request.urlretrieve(SOURCE_URL, RAW_CSV)
    print(f"Downloaded source file: {RAW_CSV}")

What should match?

This complete reference contains the source constants plus both helper functions. The blank lines also separate configuration from each function.

Build the append-only raw loader

The loader uses DuckDB's CSV reader to infer the source structure for raw_local_roads. It then appends every road record into that table.

  • In PyCharm's project sidebar, right-click the colorado-roads-pipeline folder.
  • Choose the option for creating a new Python file.
  • Name the file naive_load.py.

You will see naive_load.py appear beside common.py in the project sidebar.

  • Paste the append-only loader into naive_load.py using this code:
import duckdb

from common import DB_PATH, download_source


download_source()
connection = duckdb.connect(DB_PATH)

connection.execute("""
    CREATE TABLE IF NOT EXISTS raw_local_roads AS
    SELECT *
    FROM read_csv('data/local_roads.csv')
    LIMIT 0
""")

connection.execute("""
    INSERT INTO raw_local_roads
    SELECT *
    FROM read_csv('data/local_roads.csv')
""")

row_count = connection.execute(
    "SELECT COUNT(*) FROM raw_local_roads"
).fetchone()[0]

print(f"Naive load complete. Raw rows: {row_count:,}")
connection.close()

What does this loader do?

  • The loader calls download_source() before connecting to the database at DB_PATH.
  • The first query creates raw_local_roads with the structure inferred from the source file.
  • The second query appends the downloaded records to the raw table.
  • The row_count query measures how many records now exist.
  • The final lines print the result before closing the database connection.
  • Save naive_load.py.
  • Confirm naive_load.py remains listed beside common.py in the project sidebar.

Does the loader show a warning?

  • Check that each SQL statement has an opening triple quote plus a closing triple quote.
  • Check that DB_PATH plus download_source are imported from common.

Ask for a focused comparison: help me compare naive_load.py with the expected append-only DuckDB loader.

✔️ Awesome, I've got everything!

Your saved loader now downloads the source before appending one raw batch.

ⓧ I'd like to double check the full code

import duckdb

from common import DB_PATH, download_source


download_source()
connection = duckdb.connect(DB_PATH)

connection.execute("""
    CREATE TABLE IF NOT EXISTS raw_local_roads AS
    SELECT *
    FROM read_csv('data/local_roads.csv')
    LIMIT 0
""")

connection.execute("""
    INSERT INTO raw_local_roads
    SELECT *
    FROM read_csv('data/local_roads.csv')
""")

row_count = connection.execute(
    "SELECT COUNT(*) FROM raw_local_roads"
).fetchone()[0]

print(f"Naive load complete. Raw rows: {row_count:,}")
connection.close()

What should match?

This complete reference contains the download call plus both SQL operations. It also includes the row-count query plus the closed connection.

Run the first raw batch

The first execution downloads the official source before creating the database file. The same execution fills raw_local_roads with one batch.

Before you run the loader, do you expect the raw row count to be zero or positive?

  • Load the first raw batch by running this command in the activated terminal:
python3 naive_load.py

What does this command do?

The command starts the loader inside your activated virtual environment. The loader downloads the source before writing one batch to the local DuckDB file.

You will see a download confirmation before the load completes. The final line begins with Naive load complete. Raw rows:.

Its displayed row count is greater than 0. This proves that official road records now exist in raw_local_roads.

Loader did not finish?

  • Confirm the terminal prompt still begins with (.venv).
  • Confirm naive_load.py plus common.py sit inside the colorado-roads-pipeline folder.
  • Check your internet connection if the source download cannot complete.

Share the terminal output without changing the files: help me diagnose why my first CDOT raw batch did not load.

That first successful load gives you a real pipeline result. Official road records now live in your local raw table.

Your raw layer now contains one append-only batch. Next, you will replay that batch plus use reconciliation to expose the duplicate records.

Replay the Batch and Expose Duplicate Data

Your first raw load put official road records into DuckDB. That result proved the source file could reach your local database.

A batch replay deliberately repeats a completed load. A SQL reconciliation audit compares total rows with distinct keys to reveal whether the dataset stayed consistent.

In this step, get ready to:
  • Replay the same source batch through the append-only loader.
  • Create a SQL audit that compares total rows with unique road identifiers.
  • Capture the audit result for your interview explanation.
Replay the source batch

An append-only loader keeps every insert. Reusing the existing source file tests how it handles a repeated batch.

Before you run it, consider whether the raw row count will stay stable or change.

  • Replay the downloaded batch by running this command:
python3 naive_load.py

You'll see a new positive raw row count. The number is twice the count from your first load.

Why did the row count double?

The existing raw_local_roads table survives because the loader creates it only when it is missing. Each run inserts every source row again.

Replay command did not complete?

  • Confirm your terminal prompt still begins with (.venv).
  • Check that your terminal is inside the colorado-roads-pipeline folder.
  • Confirm naive_load.py remains unchanged from the previous step.

Help me diagnose why the replay command did not complete.

Build the reconciliation audit

A reconciliation audit turns the doubled count into a measurable quality failure. The road identifier OBJECTID provides the distinct-key comparison.

  • Return to the project panel in PyCharm.
  • Right-click the colorado-roads-pipeline folder.
  • Choose the option for creating a Python file.
  • Enter audit_raw as the file name.
  • Create the audit in audit_raw.py by pasting this code:
import duckdb

from common import DB_PATH


connection = duckdb.connect(DB_PATH)
total_rows, unique_road_ids = connection.execute("""
    SELECT
        COUNT(*) AS total_rows,
        COUNT(DISTINCT OBJECTID) AS unique_road_ids
    FROM raw_local_roads
""").fetchone()
connection.close()

duplicate_rows = total_rows - unique_road_ids

print(f"Total raw rows: {total_rows:,}")
print(f"Unique road IDs: {unique_road_ids:,}")

if duplicate_rows == 0:
    print("[PASS] No duplicate OBJECTID values found.")
else:
    print(f"[FAIL] Duplicate rows found: {duplicate_rows:,}")

What does this audit measure?

  • The DB_PATH import points the audit at the database used by the loader.
  • COUNT(*) measures every stored row.
  • COUNT(DISTINCT OBJECTID) measures unique road identifiers.
  • duplicate_rows captures the difference between those measurements.
  • The final condition prints a status that makes the quality result visible.
  • Save audit_raw.py.
  • Confirm audit_raw.py appears beside naive_load.py in the project panel.

Audit file missing from the project?

  • Verify that the file sits directly inside the colorado-roads-pipeline folder.
  • Check that the filename ends with .py.
  • Confirm the filename is exactly audit_raw.py.

Help me place audit_raw.py in the correct PyCharm project folder.

✔️ Awesome, I've got everything!

Your saved audit is ready to compare the replayed batch with its unique road identifiers.

ⓧ I'd like to double check the full code

import duckdb

from common import DB_PATH


connection = duckdb.connect(DB_PATH)
total_rows, unique_road_ids = connection.execute("""
    SELECT
        COUNT(*) AS total_rows,
        COUNT(DISTINCT OBJECTID) AS unique_road_ids
    FROM raw_local_roads
""").fetchone()
connection.close()

duplicate_rows = total_rows - unique_road_ids

print(f"Total raw rows: {total_rows:,}")
print(f"Unique road IDs: {unique_road_ids:,}")

if duplicate_rows == 0:
    print("[PASS] No duplicate OBJECTID values found.")
else:
    print(f"[FAIL] Duplicate rows found: {duplicate_rows:,}")
Run the audit

The audit converts your replay experiment into a clear pass-or-fail result. Its output gives you concrete evidence for explaining the ingestion problem.

Before you run it, consider which status should appear after the same source entered the raw table twice.

  • Run the reconciliation audit with this command:
python3 audit_raw.py

You'll see a total row count larger than the unique road ID count. The final line begins with [FAIL] Duplicate rows found:.

The reported duplicate count is greater than zero. This confirms the replay created a reproducible data-quality failure.

You now have the evidence behind the problem. That failing audit gives your interview explanation a concrete cause plus a measurable impact.

Audit not showing a failure?

  • Confirm the second loader run completed before the audit.
  • Check that both scripts import the same DB_PATH from common.py.
  • Rerun the loader from the replay substep.

Help me find why audit_raw.py is not detecting the replayed rows.

Frame the interview story

  • The same source batch entered an append-only raw table twice.
  • A SQL reconciliation query compared total rows with distinct OBJECTID values.
  • The mismatch produced a repeatable failure with a measurable duplicate count.

Your duplicate-load problem is now visible plus reproducible. Next, you'll replace the fragile append with an idempotent pipeline that rebuilds clean raw, staging, and mart layers.

Build an Idempotent Warehouse Pipeline

Your failing audit proved that replaying an append-only batch changes the warehouse result. A production ETL workflow needs one source batch to produce the same state on every run.

That repeatable behavior is idempotency. In this step, a DuckDB pipeline rebuilds each warehouse layer before automated checks verify the result.

In this step, get ready to:
  • Build an idempotent pipeline with a raw layer, a staging layer, and a mart layer.
  • Add five automated quality checks for ingestion, uniqueness, reconciliation, required identifiers, and mart output.
  • Run the pipeline twice to prove that every warehouse count remains stable.
Rebuild the warehouse layers

The CREATE OR REPLACE TABLE statement discards stale table contents before recreating them from the current CSV. Each raw load therefore starts from one source batch.

  • Use PyCharm's project file tree to create pipeline.py inside colorado-roads-pipeline.
  • Add the connection setup plus the raw-table rebuild by pasting this code into pipeline.py:
import duckdb

from common import DB_PATH, download_source


download_source()
connection = duckdb.connect(DB_PATH)

connection.execute("""
    CREATE OR REPLACE TABLE raw_local_roads AS
    SELECT *
    FROM read_csv('data/local_roads.csv')
""")

What does this code do?

  • The download_source() call makes the official source CSV available before the load begins.
  • The DB_PATH connection points the pipeline at the existing local warehouse.
  • The raw-table statement replaces the replayed rows with one clean copy of the CSV.
  • Save pipeline.py.
  • Run the partial pipeline from the activated terminal by using this command:
python3 pipeline.py

What does this run prove?

The command executes the source download check before replacing raw_local_roads. A successful return to the terminal prompt proves that the doubled raw table can be rebuilt safely.

You should see Using existing source file: data/local_roads.csv. The terminal then returns without a traceback.

Does the raw rebuild fail?

Confirm that your terminal still shows the activated .venv. Check that the raw-table block ends with its closing triple quotes plus parenthesis.

Ask for help with the exact traceback if the command still fails: help me diagnose why the raw-table rebuild in pipeline.py fails

The staging transformation gives source fields useful analytical types. It also ranks repeated OBJECTID values so only the first copy reaches the curated layer.

  • Append the staging transformation below the raw-table block in pipeline.py by pasting this code:
connection.execute("""
    CREATE OR REPLACE TABLE staging_local_roads AS
    WITH ranked AS (
        SELECT
            *,
            ROW_NUMBER() OVER (
                PARTITION BY OBJECTID
                ORDER BY OBJECTID
            ) AS duplicate_rank
        FROM raw_local_roads
    )
    SELECT
        TRY_CAST(OBJECTID AS BIGINT) AS road_id,
        NULLIF(TRIM(FIPSCOUNTY), '') AS county_code,
        NULLIF(TRIM(ROUTENAME), '') AS route_name,
        TRY_CAST(SEG_LENGTH AS DOUBLE) AS segment_length,
        NULLIF(TRIM(SURFNAME), '') AS surface_name,
        TRY_CAST(THRULNQTY AS INTEGER) AS lane_count,
        TRY_CAST(AADT AS BIGINT) AS average_daily_traffic,
        TRY_CAST(AADTYR AS INTEGER) AS traffic_year,
        NULLIF(TRIM(URBAN), '') AS urban_code
    FROM ranked
    WHERE duplicate_rank = 1
""")

How does the staging layer work?

  • The ranked query assigns a position to each row within its road identifier group.
  • The duplicate_rank = 1 filter retains one row for each source identifier.
  • The TRY_CAST expressions convert identifiers, lengths, lane counts, traffic counts, and years into analytical types.
  • The NULLIF expressions standardize empty text as null values.
  • Save pipeline.py.
  • Check that the staging transformation executes by running this command:
python3 pipeline.py

What does this run prove?

The script now rebuilds the raw table before creating staging_local_roads. Returning to the terminal prompt without a traceback proves that the complete staging query executed.

You should see the existing-source message before the terminal prompt returns. The database now contains the typed staging table.

Does the staging query fail?

Compare every source field with the code block above. A missing comma or a misspelled field can stop the transformation.

Use the traceback to narrow the problem: help me debug the staging_local_roads query in pipeline.py

A mart turns detailed road records into a county-level view that is quick to review. Its metrics expose volume plus missing-data patterns for downstream reporting.

  • Append the county quality mart below the staging block in pipeline.py by pasting this code:
connection.execute("""
    CREATE OR REPLACE TABLE mart_road_quality_summary AS
    SELECT
        COALESCE(county_code, 'UNKNOWN') AS county_code,
        COUNT(*) AS segment_count,
        ROUND(SUM(segment_length), 2) AS total_segment_length,
        SUM(CASE WHEN route_name IS NULL THEN 1 ELSE 0 END) AS missing_route_names,
        SUM(
            CASE WHEN average_daily_traffic IS NULL THEN 1 ELSE 0 END
        ) AS missing_traffic_counts
    FROM staging_local_roads
    GROUP BY 1
    ORDER BY segment_count DESC
""")

What does the mart summarize?

  • The county_code grouping creates one quality summary per county value.
  • The segment_count plus total_segment_length fields measure the road data represented by each group.
  • The missing-value metrics count absent route names plus absent traffic counts.
  • Save pipeline.py.
  • Check that the mart query completes by running this command:
python3 pipeline.py

What does this run prove?

The script now reaches the county aggregation after rebuilding the first two layers. Returning to the terminal prompt proves that mart_road_quality_summary was created successfully.

You should see the source-file message before the terminal prompt returns. The three warehouse tables now rebuild in sequence.

Does the mart query fail?

Confirm that the staging block appears above the mart block. The mart depends on staging_local_roads already existing during the run.

Share the query plus traceback for targeted help: help me debug the mart_road_quality_summary query

Printed counts make each warehouse layer observable from the terminal. They give you a quick reconciliation signal after every pipeline run.

  • Append the row-count queries plus terminal output below the mart block in pipeline.py by pasting this code:
raw_rows = connection.execute(
    "SELECT COUNT(*) FROM raw_local_roads"
).fetchone()[0]
staging_rows = connection.execute(
    "SELECT COUNT(*) FROM staging_local_roads"
).fetchone()[0]
mart_rows = connection.execute(
    "SELECT COUNT(*) FROM mart_road_quality_summary"
).fetchone()[0]

print(f"Raw rows: {raw_rows:,}")
print(f"Staging rows: {staging_rows:,}")
print(f"Mart rows: {mart_rows:,}")
connection.close()

Why print all three counts?

  • The three queries read the number of rows in each warehouse layer.
  • The formatted output makes rerun stability visible without opening the database directly.
  • The final connection.close() call releases the database connection after the checks finish.
  • Save pipeline.py.

Before you run the complete pipeline, do you expect the raw count to stay doubled or return to one source batch?

  • Run the complete pipeline by using this command:
python3 pipeline.py

What does this result prove?

The raw replacement removes the replayed copy before the printed counts are calculated. Matching raw plus staging counts show that one source row reaches the typed layer for each retained identifier.

You should see matching values after Raw rows: and Staging rows:. You should also see a positive value after Mart rows:.

You have removed the duplicate replay from the active warehouse. Each layer now rebuilds from a clean source state.

Do the counts look wrong?

Check that the raw statement says CREATE OR REPLACE TABLE raw_local_roads. Confirm that the staging filter says WHERE duplicate_rank = 1.

Ask for help comparing the three counts: help me diagnose unexpected raw, staging, and mart counts

✔️ Awesome, I've got everything!

Great. Save pipeline.py before you continue.

ⓧ I'd like to double check the full code

import duckdb

from common import DB_PATH, download_source


download_source()
connection = duckdb.connect(DB_PATH)

connection.execute("""
    CREATE OR REPLACE TABLE raw_local_roads AS
    SELECT *
    FROM read_csv('data/local_roads.csv')
""")

connection.execute("""
    CREATE OR REPLACE TABLE staging_local_roads AS
    WITH ranked AS (
        SELECT
            *,
            ROW_NUMBER() OVER (
                PARTITION BY OBJECTID
                ORDER BY OBJECTID
            ) AS duplicate_rank
        FROM raw_local_roads
    )
    SELECT
        TRY_CAST(OBJECTID AS BIGINT) AS road_id,
        NULLIF(TRIM(FIPSCOUNTY), '') AS county_code,
        NULLIF(TRIM(ROUTENAME), '') AS route_name,
        TRY_CAST(SEG_LENGTH AS DOUBLE) AS segment_length,
        NULLIF(TRIM(SURFNAME), '') AS surface_name,
        TRY_CAST(THRULNQTY AS INTEGER) AS lane_count,
        TRY_CAST(AADT AS BIGINT) AS average_daily_traffic,
        TRY_CAST(AADTYR AS INTEGER) AS traffic_year,
        NULLIF(TRIM(URBAN), '') AS urban_code
    FROM ranked
    WHERE duplicate_rank = 1
""")

connection.execute("""
    CREATE OR REPLACE TABLE mart_road_quality_summary AS
    SELECT
        COALESCE(county_code, 'UNKNOWN') AS county_code,
        COUNT(*) AS segment_count,
        ROUND(SUM(segment_length), 2) AS total_segment_length,
        SUM(CASE WHEN route_name IS NULL THEN 1 ELSE 0 END) AS missing_route_names,
        SUM(
            CASE WHEN average_daily_traffic IS NULL THEN 1 ELSE 0 END
        ) AS missing_traffic_counts
    FROM staging_local_roads
    GROUP BY 1
    ORDER BY segment_count DESC
""")

raw_rows = connection.execute(
    "SELECT COUNT(*) FROM raw_local_roads"
).fetchone()[0]
staging_rows = connection.execute(
    "SELECT COUNT(*) FROM staging_local_roads"
).fetchone()[0]
mart_rows = connection.execute(
    "SELECT COUNT(*) FROM mart_road_quality_summary"
).fetchone()[0]

print(f"Raw rows: {raw_rows:,}")
print(f"Staging rows: {staging_rows:,}")
print(f"Mart rows: {mart_rows:,}")
connection.close()
Automate the quality gates

Automated SQL checks turn warehouse expectations into repeatable pass-or-fail evidence. This file tests ingestion, uniqueness, reconciliation, required identifiers, and mart output.

  • Use PyCharm's project file tree to create quality_checks.py inside colorado-roads-pipeline.
  • Add the database connection plus the five check definitions by pasting this code into quality_checks.py:
import duckdb

from common import DB_PATH


connection = duckdb.connect(DB_PATH)
checks = [
    ("Raw rows loaded", "SELECT COUNT(*) > 0 FROM raw_local_roads"),
    (
        "Raw OBJECTID values are unique",
        "SELECT COUNT(*) = COUNT(DISTINCT OBJECTID) FROM raw_local_roads",
    ),
    (
        "Raw and staging row counts reconcile",
        """
        SELECT
            (SELECT COUNT(*) FROM raw_local_roads) =
            (SELECT COUNT(*) FROM staging_local_roads)
        """,
    ),
    (
        "Staging road IDs are not null",
        "SELECT COUNT(*) = 0 FROM staging_local_roads WHERE road_id IS NULL",
    ),
    (
        "County quality mart contains rows",
        "SELECT COUNT(*) > 0 FROM mart_road_quality_summary",
    ),
]

What do these checks protect?

  • The first check confirms that source ingestion produced at least one row.
  • The uniqueness check confirms that each raw OBJECTID appears once.
  • The reconciliation check confirms that raw plus staging row counts match.
  • The final checks protect required road identifiers plus the county mart output.
  • Save quality_checks.py.
  • Confirm that the check definitions load correctly by running this command:
python3 quality_checks.py

What does this run prove?

Python loads the database connection plus every query definition before returning control to the terminal. A clean return proves that the check names plus query strings are valid Python data.

You should return to the terminal prompt without a traceback. The next code block executes each stored query.

Does the check list fail to load?

Compare the parentheses plus commas around each tuple. Confirm that the closing square bracket appears after the mart check.

Use the traceback for focused help: help me fix the checks list in quality_checks.py

The execution loop runs each stored query as a Boolean test. A failed condition changes the process exit status so the quality gate cannot be overlooked.

  • Append the execution loop below the checks list in quality_checks.py by pasting this code:
all_passed = True
for name, query in checks:
    passed = connection.execute(query).fetchone()[0]
    status = "PASS" if passed else "FAIL"
    print(f"[{status}] {name}")
    all_passed = all_passed and passed

connection.close()

if not all_passed:
    raise SystemExit(1)

How does the quality gate work?

  • The loop executes each query before reading its Boolean result into passed.
  • The status value converts the result into a visible pass-or-fail label.
  • The all_passed value remembers whether every check succeeded.
  • A failed collection exits with a nonzero status for use in an automated pipeline.
  • Save quality_checks.py.

Before you run the checks, do you expect the earlier duplicate failure to remain after the raw replacement?

  • Execute all five quality checks by running this command:
python3 quality_checks.py

What does this result prove?

Each printed line represents one warehouse invariant evaluated directly against the database. Five passing results prove that the rebuilt layers satisfy the current quality rules.

You should see five lines beginning with [PASS]. The checks cover loaded rows, unique raw identifiers, reconciled counts, non-null staging identifiers, and mart output.

Does a quality check fail?

Read the failed check name before changing any code. That name identifies whether the problem belongs to raw ingestion, staging reconciliation, identifier conversion, or mart creation.

Share the failed check plus the latest pipeline counts for help: help me trace a failing warehouse quality check

✔️ Awesome, I've got everything!

Great. Save quality_checks.py before the final replay test.

ⓧ I'd like to double check the full code

import duckdb

from common import DB_PATH


connection = duckdb.connect(DB_PATH)
checks = [
    ("Raw rows loaded", "SELECT COUNT(*) > 0 FROM raw_local_roads"),
    (
        "Raw OBJECTID values are unique",
        "SELECT COUNT(*) = COUNT(DISTINCT OBJECTID) FROM raw_local_roads",
    ),
    (
        "Raw and staging row counts reconcile",
        """
        SELECT
            (SELECT COUNT(*) FROM raw_local_roads) =
            (SELECT COUNT(*) FROM staging_local_roads)
        """,
    ),
    (
        "Staging road IDs are not null",
        "SELECT COUNT(*) = 0 FROM staging_local_roads WHERE road_id IS NULL",
    ),
    (
        "County quality mart contains rows",
        "SELECT COUNT(*) > 0 FROM mart_road_quality_summary",
    ),
]

all_passed = True
for name, query in checks:
    passed = connection.execute(query).fetchone()[0]
    status = "PASS" if passed else "FAIL"
    print(f"[{status}] {name}")
    all_passed = all_passed and passed

connection.close()

if not all_passed:
    raise SystemExit(1)
Prove repeated runs stay stable

The decisive idempotency test repeats the same pipeline with the same input. Stable counts show that a batch replay no longer changes the warehouse state.

Before you run the sequence, do you think either pipeline execution will increase the raw count?

  • Run the complete pipeline twice before executing its quality checks by using these commands:
python3 pipeline.py
python3 pipeline.py
python3 quality_checks.py

What does this sequence prove?

  • The first pipeline run rebuilds every layer from the downloaded source.
  • The second pipeline run repeats that rebuild without appending another batch.
  • The final command evaluates all five quality rules against the repeated result.

You should see identical Raw rows:, Staging rows:, and Mart rows: values in both pipeline runs. Five lines should then begin with [PASS].

You have closed the replay gap. The warehouse now produces stable layers plus repeatable quality evidence from the same source batch.

Do repeated runs produce different counts?

Confirm that all three table statements use CREATE OR REPLACE TABLE. Check that you ran pipeline.py instead of the append-only naive_load.py.

Ask for help with both sets of counts if they still differ: help me find why my idempotent pipeline changes counts between runs

Your warehouse is repeatable plus tested. Next, you will turn its counts and county summaries into a recruiter-ready Markdown report.

Generate the Recruiter Demo Report

Your warehouse now rebuilds clean layers. Every automated quality check passes. A recruiter still needs an artifact that communicates those results quickly.

This step turns your verified DuckDB results into a Markdown report that you can open during the interview.

A README connects the local ETL design to the advertised role. Git prepares the finished evidence for review.

In this step, get ready to:
  • Generate a Markdown report from the warehouse results.
  • Document the architecture mapping and interview talk track.
  • Stage the finished project in Git.
Build the quality report

The report turns warehouse queries into evidence that someone can scan without running the pipeline. Its first section captures the reconciliation metrics that prove your repaired load is stable.

  • Return to the PyCharm project from earlier.
  • Right-click the colorado-roads-pipeline folder in the file sidebar.
  • Choose the option for a new Python file.
  • Name the file report.py.
  • Add the database setup and pipeline metrics query with this code:
import os

import duckdb

from common import DB_PATH, REPORT_PATH


os.makedirs("reports", exist_ok=True)
connection = duckdb.connect(DB_PATH, read_only=True)

raw_rows, unique_ids, staging_rows, mart_rows = connection.execute("""
    SELECT
        (SELECT COUNT(*) FROM raw_local_roads),
        (SELECT COUNT(DISTINCT OBJECTID) FROM raw_local_roads),
        (SELECT COUNT(*) FROM staging_local_roads),
        (SELECT COUNT(*) FROM mart_road_quality_summary)
""").fetchone()

What does this code do?

  • The connection opens DB_PATH in read-only mode. This protects the warehouse while the report queries it.
  • The query collects raw rows and unique road identifiers. It also collects staging rows and mart rows.
  • The four values become the headline metrics in the generated report.
  • Add the initial report structure below the metrics query with this code:
connection.close()

duplicate_rows = raw_rows - unique_ids
lines = [
    "# Colorado Local Roads Pipeline Report",
    "",
    "## Pipeline checks",
    "",
    f"- Raw rows: {raw_rows:,}",
    f"- Unique road IDs: {unique_ids:,}",
    f"- Duplicate rows: {duplicate_rows:,}",
    f"- Staging rows: {staging_rows:,}",
    f"- County mart rows: {mart_rows:,}",
    "",
    "## Ten largest county summaries",
    "",
    "| County code | Segments | Total segment length | Missing route names | Missing traffic counts |",
    "|---|---:|---:|---:|---:|",
]

with open(REPORT_PATH, "w", encoding="utf-8") as report_file:
    report_file.write("\n".join(lines) + "\n")

print(f"Report written to {REPORT_PATH}")

How is the report assembled?

  • The subtraction derives duplicate_rows from the raw count and unique identifier count.
  • The lines list stores each heading and metric in Markdown format.
  • The file-writing block joins those lines with newline characters. It writes the result to REPORT_PATH.
  • Save report.py.
  • Generate the first report version by running this command:
python3 report.py

You should see Report written to reports/quality_report.md in the terminal.

  • Select reports/quality_report.md in the PyCharm file sidebar.

You should see five pipeline metrics followed by the county summary table headings. The next query supplies the table rows.

Report file missing?

Check that REPORT_PATH is imported from common.py. Confirm that the terminal still uses the activated environment from earlier.

If the script reports a missing table, run the finished warehouse pipeline again before retrying the report.

Help me diagnose why report.py is not generating quality_report.md.

A recruiter also needs a compact view of county-level completeness. The mart already contains those measurements, so the report can present the ten largest county summaries.

  • In report.py, find the line containing connection.close().
  • Insert the county summary query immediately above that line with this code:
county_rows = connection.execute("""
    SELECT
        county_code,
        segment_count,
        total_segment_length,
        missing_route_names,
        missing_traffic_counts
    FROM mart_road_quality_summary
    ORDER BY segment_count DESC
    LIMIT 10
""").fetchall()

What does the county query collect?

  • The query reads the completeness measurements produced by mart_road_quality_summary.
  • The ordering puts counties with the most road segments first.
  • The limit keeps the recruiter report focused on ten summaries.
  • In report.py, find the line beginning with with open(REPORT_PATH.
  • Insert the table-row loop immediately above that line with this code:
for county_code, segment_count, total_length, missing_names, missing_traffic in county_rows:
    safe_length = 0 if total_length is None else total_length
    lines.append(
        f"| {county_code} | {segment_count:,} | {safe_length:,.2f} | "
        f"{missing_names:,} | {missing_traffic:,} |"
    )

How are table rows formatted?

  • The loop turns every result tuple into one Markdown table row.
  • The safe_length value prevents an empty aggregate from breaking numeric formatting.
  • The formatted strings add thousands separators and two decimal places for readability.

Before you rerun the report, do you expect the duplicate count to stay at zero while county rows appear beneath the table headings?

  • Save report.py.
  • Generate the complete report by running this command:
python3 report.py

You should see Report written to reports/quality_report.md again.

  • Return to reports/quality_report.md in PyCharm.
  • Confirm that the duplicate row metric shows 0.
  • Confirm that ten county summary rows appear beneath the table headings.

Strong result. Your passing checks now have a readable artifact that exposes counts and county-level completeness.

County table still empty?

Check that the county_rows query appears before connection.close(). Check that the loop appears before the file-writing block.

Help me find why my Markdown county table has headings but no data rows.

✔️ Awesome, I've got everything!

Great. Double-check that you saved report.py before moving on.

ⓧ I'd like to double check the full code

import os

import duckdb

from common import DB_PATH, REPORT_PATH


os.makedirs("reports", exist_ok=True)
connection = duckdb.connect(DB_PATH, read_only=True)

raw_rows, unique_ids, staging_rows, mart_rows = connection.execute("""
    SELECT
        (SELECT COUNT(*) FROM raw_local_roads),
        (SELECT COUNT(DISTINCT OBJECTID) FROM raw_local_roads),
        (SELECT COUNT(*) FROM staging_local_roads),
        (SELECT COUNT(*) FROM mart_road_quality_summary)
""").fetchone()

county_rows = connection.execute("""
    SELECT
        county_code,
        segment_count,
        total_segment_length,
        missing_route_names,
        missing_traffic_counts
    FROM mart_road_quality_summary
    ORDER BY segment_count DESC
    LIMIT 10
""").fetchall()
connection.close()

duplicate_rows = raw_rows - unique_ids
lines = [
    "# Colorado Local Roads Pipeline Report",
    "",
    "## Pipeline checks",
    "",
    f"- Raw rows: {raw_rows:,}",
    f"- Unique road IDs: {unique_ids:,}",
    f"- Duplicate rows: {duplicate_rows:,}",
    f"- Staging rows: {staging_rows:,}",
    f"- County mart rows: {mart_rows:,}",
    "",
    "## Ten largest county summaries",
    "",
    "| County code | Segments | Total segment length | Missing route names | Missing traffic counts |",
    "|---|---:|---:|---:|---:|",
]

for county_code, segment_count, total_length, missing_names, missing_traffic in county_rows:
    safe_length = 0 if total_length is None else total_length
    lines.append(
        f"| {county_code} | {segment_count:,} | {safe_length:,.2f} | "
        f"{missing_names:,} | {missing_traffic:,} |"
    )

with open(REPORT_PATH, "w", encoding="utf-8") as report_file:
    report_file.write("\n".join(lines) + "\n")

print(f"Report written to {REPORT_PATH}")
Document the demo story

The generated report proves what the pipeline produced. The README explains why each layer exists and gives you a concise interview story.

  • Right-click the colorado-roads-pipeline folder in the PyCharm file sidebar.
  • Choose the option for a new file.
  • Name the file README.md.
  • Add the project purpose and architecture with this content:
# Colorado Local Roads ETL Quality Pipeline

This project demonstrates a local ETL and data-quality workflow using official Colorado Department of Transportation Local Roads data.

## Why this project exists

It targets the core skills in Ledgent Technology Data Engineer job reference `JN -092026-430398`: SQL querying, ETL support, dataset comparison, discrepancy troubleshooting, data validation, warehouse concepts, Python, and technical documentation.

This is a local demonstration. It does not claim to deploy Azure Data Factory, Azure SQL, SSIS, or Snowflake.

## Architecture

1. `raw_local_roads` lands the source CSV with its original fields.
2. `staging_local_roads` deduplicates by `OBJECTID`, converts important fields to useful types, and standardizes empty text as null.
3. `mart_road_quality_summary` aggregates completeness metrics by county code.
4. `quality_checks.py` validates uniqueness, reconciliation, required identifiers, and mart output.
5. `report.py` creates a concise Markdown artifact for review.

Why document the architecture?

The architecture section assigns one responsibility to each warehouse layer. That structure gives you a clear explanation of ingestion and transformation.

The wording also states that this is a local demonstration. It keeps your interview claims accurate.

  • Save README.md.
  • Review the five numbered architecture points in README.md.

You should see separate descriptions for the raw layer and staging layer. You should also see the mart and validation responsibilities.

Architecture list looks merged?

Check that every numbered item begins on its own line. Keep a blank line before the Architecture heading.

Help me fix the Markdown formatting in the architecture section of my README.

The cloud-tool mapping shows how your local implementation transfers to the advertised platform stack. It communicates relevant understanding without claiming a cloud deployment.

  • Append the cloud-tool mapping to README.md with this content:
## Cloud-tool mapping

- `naive_load.py` and `pipeline.py` represent orchestration and transformation activities that could be implemented with Azure Data Factory or SSIS.
- DuckDB's local raw, staging, and mart tables represent warehouse layers that could live in Azure SQL or Snowflake.
- `quality_checks.py` represents validation and reconciliation gates around a production pipeline.
- `reports/quality_report.md` represents operational documentation and reporting support.

How does the cloud mapping stay honest?

The mapping connects local orchestration to Azure Data Factory and SSIS.

It connects local warehouse layers to Azure SQL and Snowflake. The wording describes transferable design decisions without claiming deployment.

  • Save README.md.
  • Confirm that each local component maps to one advertised responsibility.

Cloud mapping unclear?

Check that the local scripts remain the subject of each mapping statement. Keep the wording about possible implementation in the cloud tools.

Help me explain this local ETL project without implying that I deployed it to cloud platforms.

A reproducible run sequence lets another person replay the duplicate failure and verify the fix. The interview talk track turns that sequence into a six-part explanation.

  • Append the exact run sequence to README.md with this content:
## Run the project

```bash
source .venv/bin/activate
python3 naive_load.py
python3 naive_load.py
python3 audit_raw.py
python3 pipeline.py
python3 pipeline.py
python3 quality_checks.py
python3 report.py
```

Why preserve the full run sequence?

The sequence reproduces the original append-only failure before running the idempotent replacement. It finishes with passing checks and a refreshed report.

  • Save README.md.
  • Confirm that the run sequence starts with environment activation.
  • Confirm that the run sequence ends with report generation.

Run sequence formatting broken?

Check that the opening and closing code fences each use three backticks. Keep every command on a separate line inside the fenced block.

Help me fix the fenced command block in my README.

  • Append the interview talk track to README.md with this content:
## Interview talk track

1. I first built a naive append-only batch loader to get a visible result quickly.
2. I replayed the same batch and used SQL reconciliation to prove that duplicate `OBJECTID` values were introduced.
3. I replaced the append with an idempotent rebuild so repeated executions produce stable counts.
4. I separated raw ingestion, typed staging transformations, and a reporting mart to keep responsibilities clear.
5. I added automated quality gates and a report so failures are visible before downstream consumers use the data.
6. In Azure, I would map orchestration to Azure Data Factory, relational storage to Azure SQL or Snowflake, and package-style ETL work to SSIS.

How does the talk track flow?

The story begins with a fast loader and a reproducible failure. It then explains the diagnosis and idempotent repair.

The final points connect quality gates and warehouse layers to the advertised responsibilities. This gives the recruiter a problem-to-proof narrative.

  • Save README.md.
  • Read the six talk-track points aloud to confirm that the story moves from failure to repair.

Talk track feels disconnected?

Check that the duplicate replay appears before the idempotent repair. Keep the cloud mapping as the final point.

Help me practise this six-part ETL interview talk track as a concise spoken answer.

The final README section records where the data came from. Attribution makes the demo traceable to the official public source.

  • Append the data source attribution to README.md with this content:
## Data source and attribution

- Dataset: Local Roads
- Publisher: CDOT ArcGIS Online
- Catalog: `https://data.colorado.gov/dataset/Local-Roads/d8t8-kgsa`
- CSV: `https://geodata.colorado.gov/api/download/v1/items/f031695aeaeb42a0b86e4887108d03c8/csv?layers=14`
- License: CC BY-SA 4.0, `https://creativecommons.org/licenses/by-sa/4.0`

Why include attribution?

The attribution identifies the dataset and publisher. It also preserves the catalog link and license beside the demo.

  • Save README.md.
  • Confirm that the attribution section lists the dataset and publisher.
  • Confirm that the attribution section includes the catalog and license.

Links not displaying correctly?

Check that every URL remains inside one pair of backticks. Confirm that no URL wraps onto a second source line.

Help me fix the data source links in my Markdown README.

✔️ Awesome, I've got everything!

Your README now documents the purpose and architecture. It also contains the cloud mapping and run sequence. The talk track and attribution are included.

ⓧ I'd like to double check the full code

# Colorado Local Roads ETL Quality Pipeline

This project demonstrates a local ETL and data-quality workflow using official Colorado Department of Transportation Local Roads data.

## Why this project exists

It targets the core skills in Ledgent Technology Data Engineer job reference `JN -092026-430398`: SQL querying, ETL support, dataset comparison, discrepancy troubleshooting, data validation, warehouse concepts, Python, and technical documentation.

This is a local demonstration. It does not claim to deploy Azure Data Factory, Azure SQL, SSIS, or Snowflake.

## Architecture

1. `raw_local_roads` lands the source CSV with its original fields.
2. `staging_local_roads` deduplicates by `OBJECTID`, converts important fields to useful types, and standardizes empty text as null.
3. `mart_road_quality_summary` aggregates completeness metrics by county code.
4. `quality_checks.py` validates uniqueness, reconciliation, required identifiers, and mart output.
5. `report.py` creates a concise Markdown artifact for review.

## Cloud-tool mapping

- `naive_load.py` and `pipeline.py` represent orchestration and transformation activities that could be implemented with Azure Data Factory or SSIS.
- DuckDB's local raw, staging, and mart tables represent warehouse layers that could live in Azure SQL or Snowflake.
- `quality_checks.py` represents validation and reconciliation gates around a production pipeline.
- `reports/quality_report.md` represents operational documentation and reporting support.

## Run the project

```bash
source .venv/bin/activate
python3 naive_load.py
python3 naive_load.py
python3 audit_raw.py
python3 pipeline.py
python3 pipeline.py
python3 quality_checks.py
python3 report.py
```

## Interview talk track

1. I first built a naive append-only batch loader to get a visible result quickly.
2. I replayed the same batch and used SQL reconciliation to prove that duplicate `OBJECTID` values were introduced.
3. I replaced the append with an idempotent rebuild so repeated executions produce stable counts.
4. I separated raw ingestion, typed staging transformations, and a reporting mart to keep responsibilities clear.
5. I added automated quality gates and a report so failures are visible before downstream consumers use the data.
6. In Azure, I would map orchestration to Azure Data Factory, relational storage to Azure SQL or Snowflake, and package-style ETL work to SSIS.

## Data source and attribution

- Dataset: Local Roads
- Publisher: CDOT ArcGIS Online
- Catalog: `https://data.colorado.gov/dataset/Local-Roads/d8t8-kgsa`
- CSV: `https://geodata.colorado.gov/api/download/v1/items/f031695aeaeb42a0b86e4887108d03c8/csv?layers=14`
- License: CC BY-SA 4.0, `https://creativecommons.org/licenses/by-sa/4.0`
Stage the finished project

Staging creates one reviewable snapshot of the finished demo. Your existing ignore rules keep the downloaded CSV and local database outside that snapshot.

  • Stage the finished project by running this command:
git add .

Before you check the staging area, which generated artifact do you expect Git to include while the downloaded data remains absent?

  • Inspect the staged files by running this command:
git status

You should see the Python source files and README.md staged. You should also see requirements.txt and .gitignore staged.

The generated reports/quality_report.md should be staged. The files data/local_roads.csv and data/colorado_roads.duckdb should be absent.

Why are the data files absent?

Your .gitignore rules exclude downloaded CSV files and DuckDB database files inside data/. The code and generated report remain reproducible without staging those larger local artifacts.

Downloaded data appears in Git?

Return to .gitignore from earlier. Confirm that it still contains the patterns for CSV files and DuckDB files inside data/.

Help me diagnose why ignored data files appear in my staged Git changes.

That is your recruiter demo ready. You can now show the passing metrics and explain the failure-to-fix story from one local project.

Secret mission

Quarantine Invalid Road Records

Malformed source records deserve a traceable destination. Add a quarantine table for invalid road identifiers or segment lengths, then prove every raw row reaches an accountable outcome.

Clean Up Your Resources

Clean Up Your Resources

This project runs locally with no ongoing costs. Choose whether to keep it available, pause the active environment, or delete it entirely.

Resources you used:

  • The colorado-roads-pipeline folder on your Desktop. This contains your Python source files, README.md, requirements.txt, and .gitignore.
  • The .venv virtual environment inside the project folder. This contains DuckDB version 1.5.5.
  • The downloaded data/local_roads.csv file. The local database is stored at data/colorado_roads.duckdb.
  • The generated reports/quality_report.md report. Your local Git repository is also stored inside the project folder.

Keep everything running

No action is needed. Choose this if you want to rerun the pipeline before your interview.

  • Refresh your warehouse results and report by running these commands:
python3 pipeline.py
python3 quality_checks.py
python3 report.py

What Do These Commands Refresh?

  • The first command rebuilds the raw, rejected, staging, and mart layers.
  • The second command checks row accounting and staging data quality.
  • The third command refreshes reports/quality_report.md.

You should see every quality check print [PASS]. Your interview artifact is now refreshed with the latest warehouse results.

Pause - I'll come back to this later

Deactivate the virtual environment to return your terminal to its normal shell. Your project files remain available for your next session.

  • Pause the active virtual environment by running this command:
deactivate

What Does This Change?

The command leaves the active .venv session. It does not delete the environment or any project files.

Your shell prompt should no longer show .venv. The pipeline data and report remain on your Desktop.

Delete - I don't want to use this again

Deleting this folder is permanent. It removes only the local project resources created during this project.

  • Remove the complete colorado-roads-pipeline folder by running these commands:
deactivate
cd ~/Desktop
rm -rf colorado-roads-pipeline

What This Removes

  • The first command leaves the active virtual environment.
  • The second command moves your terminal outside the project folder.
  • The final command deletes the project folder. This includes every resource listed above.
  • Confirm the deletion by checking your Desktop.

You should no longer see the colorado-roads-pipeline folder. All project files and local data have now been removed.

Nice Work!

Nice Work!

What a result! You've built a local ETL quality pipeline over official CDOT road data. The generated report gives you a concise story for your Data Engineer interview.

You've learned how to:

  • Build a layered data warehouse with DuckDB. Land source records in a raw layer. Convert them into typed staging rows. Summarize county-level quality in a mart.
  • Reproduce a duplicate-load failure by replaying the same batch. Diagnose duplicate OBJECTID values with SQL reconciliation.
  • Apply idempotency so reruns produce stable counts. Prove the result with automated quality checks. Generate a Markdown report that maps the local workflow to cloud data responsibilities.
  • Secret Mission: Add a quarantine path that stores invalid road records with rejection reasons. Prove complete raw-row accounting across staging and rejected outputs.

Ready to quiz yourself?