Build a Dutch Energy Data Platform

Introduction

30 Second Summary

Every time you check your energy bill or see a headline about the Dutch energy transition, data pipelines are working behind the scenes to collect, transform, and deliver that information. Data engineers build those pipelines, and the Netherlands has one of Europe's highest demands for this role.

In this project, you will build a complete Dutch Energy Data Platform over 8 weeks. You'll go from advanced SQL foundations to a production-quality portfolio project using the exact tools that appear in Netherlands data engineering job postings.

What You'll Build

A complete data engineering platform that extracts real Dutch government energy data from the CBS Open Data API, transforms it through a medallion architecture using dbt, and orchestrates the full pipeline with Apache Airflow.

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

• A PostgreSQL database with advanced analytical queries using window functions and CTEs on real Dutch data.
• A complete dbt project with medallion architecture (bronze/silver/gold layers), data quality tests, macros, and full documentation.
• An Airflow DAG that orchestrates the full ELT pipeline with error handling, retries, and monitoring.
• A GitHub repository documenting the entire project as a portfolio piece for Dutch employers.
• Secret Mission: Build a Streamlit dashboard that visualizes Netherlands energy transition trends.

Are there any prerequisites?

You should have basic Python knowledge, basic SQL knowledge, and PostgreSQL installed and working on your machine. This roadmap builds on those foundations to take you from intermediate to job-ready.

Install Your Data Engineering Toolkit

Before you can build data pipelines, you need the right tools installed and ready. This project uses Docker for containerized orchestration, dbt for data transformations, and DuckDB for lightweight local analytics.

In this step, you will install all three tools alongside Git for version control, and set up the folder structure for your 8-week project.

In this step, get ready to:

• Install Docker Engine and Docker Compose on Linux.
• Install dbt Core with the PostgreSQL adapter and DuckDB in a Python virtual environment.
• Set up Git and create a project repository with the roadmap folder structure.

Install Docker Engine on Linux

Docker runs Apache Airflow in later weeks. You need Docker Engine and Docker Compose v2.14.0 or newer installed on your machine.

• Set up Docker's official apt repository by running these commands:

# Add Docker's official GPG key
sudo apt-get update
sudo apt-get install ca-certificates curl
sudo install -m 0755 -d /etc/apt/keyrings
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
sudo chmod a+r /etc/apt/keyrings/docker.asc

# Add the repository to apt sources
echo \
  "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/ubuntu \
  $(. /etc/os-release && echo "${UBUNTU_CODENAME:-$VERSION_CODENAME}") stable" | \
  sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
sudo apt-get update

What does this do?

These commands download Docker's official GPG signing key and register Docker's package repository with your system's package manager.

After this, apt knows where to find the official Docker packages.

• Install Docker Engine and Docker Compose by running:

sudo apt-get install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin

What does this install?

This installs the Docker daemon (docker-ce), the CLI client (docker-ce-cli), and the Docker Compose plugin (docker-compose-plugin). Docker Compose is what you will use to run Airflow's multi-container setup later.

• Allow your user to run Docker without sudo by adding yourself to the docker group:

sudo usermod -aG docker $USER
newgrp docker

Why do I need this?

By default, Docker commands require sudo. Adding your user to the docker group lets you run docker and docker compose directly.

The newgrp docker command activates the new group membership in your current shell session without requiring a logout.

• Verify Docker is installed by running:

docker --version

You should see a version number printed, confirming Docker Engine is installed.

• Verify Docker Compose meets the minimum version requirement by running:

docker compose version

✔️ I see version 2.14.0 or higher

Docker Compose is ready. You are all set to move on.

ⓧ I see a version lower than 2.14.0

You need Docker Compose v2.14.0 or newer for Airflow. Update your Docker packages by running:

sudo apt-get update
sudo apt-get install --only-upgrade docker-compose-plugin

Run docker compose version again to confirm the update.

ⓧ Command not found

The Docker Compose plugin was not installed. Install it by running:

sudo apt-get install docker-compose-plugin

Then run docker compose version to confirm it prints a version number.

Getting permission denied?

If docker --version works but docker compose version shows a permission error, make sure your newgrp docker command ran successfully. Try logging out and back in, then run the version commands again.

help me fix Docker permission errors on Linux

Install DuckDB and dbt Core with the PostgreSQL adapter

The dbt docs recommend installing dbt inside a dedicated Python virtual environment to avoid dependency conflicts.

• Move to your Desktop by running this command:

cd ~/Desktop

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

mkdir dutch-energy-platform
cd dutch-energy-platform

• Create a Python virtual environment by running:

python3 -m venv dbt-env

• Activate the virtual environment by running:

source dbt-env/bin/activate

Your terminal prompt should now show (dbt-env) at the beginning, confirming the virtual environment is active.

• Upgrade pip and install build dependencies by running:

python -m pip install --upgrade pip wheel setuptools

• Install dbt with the PostgreSQL adapter and DuckDB by running:

python -m pip install dbt-postgres duckdb

What does this install?

The dbt-postgres package installs dbt Core automatically along with the adapter for PostgreSQL. You do not need to install dbt-core separately.

DuckDB is a lightweight analytics database that you will use for fast SQL practice without needing a running PostgreSQL server for every exercise.

• Verify dbt installed correctly by running:

dbt --version

✔️ I see dbt version output

dbt is ready. You should see dbt-core and dbt-postgres listed with their version numbers.

ⓧ Command not found

Make sure your virtual environment is active. Your prompt should start with (dbt-env). If not, activate it again:

source dbt-env/bin/activate

Then try dbt --version again.

ⓧ I see an error during install

dbt Core requires Python 3.10 or newer. Check your Python version:

python3 --version

If your version is below 3.10, you will need to install a newer Python version before proceeding.

Still stuck on the dbt install?

Make sure you ran the pip install --upgrade pip wheel setuptools step first. Outdated build tools are the most common cause of install failures.

help me debug my dbt-postgres installation error

Set up Git and create your project repository

Your project needs version control from day one. You will use Git to track every change across the 8-week roadmap and push it to GitHub as a portfolio piece.

• Check if Git is already installed by running:

git --version

✔️ I see a version number

Git is ready. Move on to creating the folder structure.

ⓧ Command not found

Install Git by running:

sudo apt-get install git

Run git --version again to confirm it is installed.

• Initialize a Git repository in your project folder by running:

git init

• Create the folder structure for the roadmap project by running:

mkdir -p dbt airflow/dags extraction sql docs

What are these folders for?

• dbt/ holds your dbt project (models, tests, macros).
• airflow/dags/ stores your Airflow DAG definitions for pipeline orchestration.
• extraction/ contains Python scripts that pull data from the CBS API.
• sql/ is where you practice complex SQL queries in the early weeks.
• docs/ stores architecture diagrams, data dictionaries, and design decisions.

• Create a README file by running:

echo "# Dutch Energy Data Platform" > README.md

• Confirm all files and folders exist by running:

ls

You should see dbt, airflow, extraction, sql, docs, dbt-env, and README.md in the output.

• Open the project folder in VS Code by running:

code .

VS Code not opening from terminal?

If the code command is not recognized, open VS Code manually and use File > Open Folder to navigate to ~/Desktop/dutch-energy-platform.

help me open VS Code from the Linux terminal

Your toolkit is ready. Next up, you will connect to PostgreSQL and start writing advanced SQL queries on real Dutch government data.

Master Advanced SQL Foundations

Real data engineering starts with SQL that goes beyond simple SELECT statements. Dutch companies like ABN AMRO and KPMG expect you to write queries that rank, compare, and aggregate data across multiple dimensions in a single pass.

In this step, you will load real Dutch neighborhood data from CBS (Statistics Netherlands) into PostgreSQL and write analytical queries using Common Table Expressions (CTEs) and window functions.

In this step, get ready to:

• Load CBS neighborhood data into PostgreSQL using the cbsodata Python package.
• Write queries using CTEs to break complex logic into readable steps.
• Master window functions (ROW_NUMBER, RANK, DENSE_RANK, LAG, LEAD, SUM OVER, AVG OVER) to analyze Dutch population trends.

Load the CBS neighborhood dataset into PostgreSQL

Before you can practice advanced SQL, you need real data to query against. The CBS Open Data API provides free, no-authentication-required access to Dutch government statistics.

• Activate your virtual environment and install the additional packages you need by running:

source dbt-env/bin/activate
pip install cbsodata pandas sqlalchemy psycopg2-binary

What are these packages?

The cbsodata package is the official Python client for the CBS Open Data API. It fetches StatLine datasets directly into Python lists that Pandas can convert into DataFrames.

SQLAlchemy and psycopg2-binary together let Pandas write DataFrames directly to PostgreSQL tables using the to_sql() method.

• Create a dedicated database for this project by running:

createdb dataeng

Permission denied or role does not exist?

If your PostgreSQL instance uses a different user, add -U postgres to the command. Make sure your PostgreSQL service is running.

Help me troubleshoot my PostgreSQL connection.

• Create the data loading script by saving the following code in extraction/load_cbs_data.py:

import cbsodata
import pandas as pd
from sqlalchemy import create_engine

# Fetch CBS neighborhood key figures (table 83765NED)
# This contains population data for all Dutch municipalities, districts, and neighborhoods
print("Fetching CBS neighborhood data (this may take up to 30 seconds)...")
data = pd.DataFrame(cbsodata.get_data('83765NED'))

# Show what columns are available
print(f"Loaded {len(data)} rows with {len(data.columns)} columns")
print("Sample columns:", list(data.columns[:10]))

# Connect to PostgreSQL and load the data
engine = create_engine('postgresql+psycopg2://postgres:postgres@localhost:5432/dataeng')

# Write the full dataset to a table called cbs_neighborhoods
data.to_sql('cbs_neighborhoods', engine, if_exists='replace', index=False)
print("Data loaded into PostgreSQL table: cbs_neighborhoods")

What does this script do?

• cbsodata.get_data('83765NED') fetches the full "Kerncijfers wijken en buurten" (Key figures neighborhoods and districts) dataset from CBS. This includes population counts, household data, and demographics for every municipality in the Netherlands.
• to_sql() writes the entire DataFrame into PostgreSQL as a table called cbs_neighborhoods. The if_exists='replace' flag drops and recreates the table each time you run the script.

• Run the loading script by executing:

python extraction/load_cbs_data.py

• You should see output confirming the row count and that data was loaded into the cbs_neighborhoods table.

Connection refused or password authentication failed?

Update the connection string in the script to match your PostgreSQL credentials. If you use peer authentication on Linux, try postgresql+psycopg2:///dataeng (no user/password, connecting via Unix socket).

Help me fix my PostgreSQL connection string for the CBS loading script.

Now let's confirm the data is queryable. Connect to your database and run a quick count.

• Open a psql session by running:

psql dataeng

• Verify your data loaded by running this query:

-- Count total rows and check municipality-level records
SELECT COUNT(*) AS total_rows,
       COUNT(*) FILTER (WHERE "WijkenEnBuurten" LIKE 'GM%') AS municipalities
FROM cbs_neighborhoods;

You should see a total row count in the thousands, with several hundred municipality-level records (codes starting with GM).

Now write your first CTE query to practice breaking complex logic into readable steps.

• Create a new file sql/week1_window_functions.sql and add this first query that uses a CTE to isolate municipality data:

-- Query 1: Top 10 municipalities by population using a CTE
WITH municipalities AS (
    -- Filter to municipality-level records only (codes starting with GM)
    SELECT
        "WijkenEnBuurten" AS region_code,
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population
FROM municipalities
ORDER BY population DESC
LIMIT 10;

What is a CTE?

A Common Table Expression (CTE) is a temporary named result set defined with the WITH keyword. It makes complex queries readable by breaking them into logical steps. Think of it as naming an intermediate result so you can reference it later.

The CTE above isolates municipality-level records from the full dataset (which also contains district and neighborhood records). The main query then works with this clean subset.

• Test this query in your psql session by copying and pasting it. You should see Amsterdam, Rotterdam, and Den Haag near the top.

Master window functions on the CBS dataset

Window functions let you compute values across rows that are related to the current row, without collapsing the result set like GROUP BY does. They are essential for ranking, comparing adjacent rows, and calculating running totals.

• Add the following query to sql/week1_window_functions.sql that demonstrates ranking functions:

-- Query 2: Rank municipalities by population with multiple ranking methods
WITH municipalities AS (
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population,
    -- ROW_NUMBER: unique sequential number, no ties
    ROW_NUMBER() OVER (ORDER BY population DESC) AS row_num,
    -- RANK: same rank for ties, skips next number
    RANK() OVER (ORDER BY population DESC) AS rank_pos,
    -- DENSE_RANK: same rank for ties, no gaps
    DENSE_RANK() OVER (ORDER BY population DESC) AS dense_rank_pos
FROM municipalities
ORDER BY population DESC
LIMIT 20;

How do the three ranking functions differ?

ROW_NUMBER always assigns unique numbers (1, 2, 3) even when values tie. RANK gives tied rows the same number but skips ahead (1, 1, 3). DENSE_RANK gives tied rows the same number without gaps (1, 1, 2).

In Dutch municipality data, ties in population are rare at the top. You will see the differences more clearly in the lower ranks where smaller municipalities share similar population counts.

• Add the next query that uses LAG and LEAD to compare adjacent rows:

-- Query 3: Population comparison between adjacent municipalities (by rank)
WITH ranked_municipalities AS (
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population,
        ROW_NUMBER() OVER (ORDER BY "AantalInwoners_5" DESC) AS rank_pos
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population,
    rank_pos,
    -- LAG: look at the previous row (larger municipality)
    LAG(municipality_name) OVER (ORDER BY rank_pos) AS larger_neighbor,
    LAG(population) OVER (ORDER BY rank_pos) AS larger_pop,
    -- LEAD: look at the next row (smaller municipality)
    LEAD(municipality_name) OVER (ORDER BY rank_pos) AS smaller_neighbor,
    -- Calculate the gap to the municipality above
    population - LEAD(population) OVER (ORDER BY rank_pos) AS gap_to_next
FROM ranked_municipalities
ORDER BY rank_pos
LIMIT 15;

What do LAG and LEAD do?

LAG(column) OVER (ORDER BY ...) looks at the value from the previous row in the specified order. LEAD(column) looks at the next row. These are powerful for calculating differences between adjacent records, like year-over-year changes or gaps between ranked items.

In this query, you can see how much larger each municipality is compared to the one ranked just below it. The gap_to_next column reveals where the biggest population jumps occur.

• Add this query that demonstrates running totals with SUM OVER and AVG OVER using PARTITION BY:

-- Query 4: Running totals and averages with PARTITION BY
WITH municipality_data AS (
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population,
        CASE
            WHEN "AantalInwoners_5" >= 100000 THEN 'Large'
            WHEN "AantalInwoners_5" >= 25000 THEN 'Medium'
            ELSE 'Small'
        END AS size_category
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population,
    size_category,
    -- Running total within each size category
    SUM(population) OVER (
        PARTITION BY size_category
        ORDER BY population DESC
    ) AS running_total,
    -- Average population within each category
    AVG(population) OVER (
        PARTITION BY size_category
    ) AS category_avg,
    -- How this municipality compares to its category average
    population - AVG(population) OVER (
        PARTITION BY size_category
    ) AS diff_from_avg
FROM municipality_data
ORDER BY size_category, population DESC;

What does PARTITION BY do?

PARTITION BY divides rows into groups (partitions) before applying the window function. Without PARTITION BY, the window function operates across all rows. With it, the function resets for each partition.

Here, the running total resets for each size category. Amsterdam's running total only accumulates with other large cities, not with all municipalities combined. This is the key difference between window functions and GROUP BY: you keep all rows while computing group-level aggregates.

• Test each query in psql to verify they return results. The running total should grow as you scan down each partition.

Write 5 complex analytical queries combining CTEs and window functions

Time to put it all together. Each query you write must combine at least 1 CTE and 2 different window functions. The queries should analyze Dutch population trends from different angles.

• Add your fifth and final query to sql/week1_window_functions.sql. This one combines all the techniques, using percentile ranking, running totals, and LAG in a single query:

-- Query 5: Comprehensive analysis combining all window function types
WITH municipality_stats AS (
    -- CTE: isolate and clean municipality data
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population,
        CASE
            WHEN "AantalInwoners_5" >= 100000 THEN 'Large'
            WHEN "AantalInwoners_5" >= 25000 THEN 'Medium'
            ELSE 'Small'
        END AS size_category
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
),
ranked AS (
    -- CTE: add ranking within each category
    SELECT
        municipality_name,
        population,
        size_category,
        DENSE_RANK() OVER (
            PARTITION BY size_category
            ORDER BY population DESC
        ) AS rank_in_category,
        SUM(population) OVER (
            PARTITION BY size_category
            ORDER BY population DESC
        ) AS cumulative_pop,
        LAG(population) OVER (
            PARTITION BY size_category
            ORDER BY population DESC
        ) AS prev_municipality_pop
    FROM municipality_stats
)
SELECT
    municipality_name,
    population,
    size_category,
    rank_in_category,
    cumulative_pop,
    -- Percentage of category total accumulated so far
    ROUND(
        cumulative_pop * 100.0 / SUM(population) OVER (PARTITION BY size_category),
        1
    ) AS pct_of_category,
    -- Gap to the municipality ranked above (within same category)
    COALESCE(prev_municipality_pop - population, 0) AS gap_to_above
FROM ranked
WHERE rank_in_category <= 10
ORDER BY size_category, rank_in_category;

What makes this query complex?

• It chains two CTEs: one for data cleaning and one for adding window calculations.
• It uses DENSE_RANK, SUM OVER, and LAG across partitions.
• The final SELECT adds another window function (SUM for category total) on top of the already-computed results. This layered approach is how production analytical queries are structured.

• Save sql/week1_window_functions.sql.
• Run the final query in psql. You should see the top 10 municipalities in each size category, with their cumulative population percentage and the gap to the municipality ranked above them.

Column not found or relation does not exist?

CBS column names are case-sensitive in PostgreSQL. Make sure you wrap them in double quotes exactly as shown (e.g., "AantalInwoners_5"). Also confirm you ran the load_cbs_data.py script successfully before querying.

Help me fix my SQL query errors on the CBS data.

✔️ Awesome, I've got everything!

Great. Double check you have saved sql/week1_window_functions.sql with all 5 queries.

ⓧ I'd like to double check the full code

Here is the complete sql/week1_window_functions.sql file with all 5 queries:

-- Query 1: Top 10 municipalities by population using a CTE
WITH municipalities AS (
    -- Filter to municipality-level records only (codes starting with GM)
    SELECT
        "WijkenEnBuurten" AS region_code,
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population
FROM municipalities
ORDER BY population DESC
LIMIT 10;

-- Query 2: Rank municipalities by population with multiple ranking methods
WITH municipalities AS (
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population,
    -- ROW_NUMBER: unique sequential number, no ties
    ROW_NUMBER() OVER (ORDER BY population DESC) AS row_num,
    -- RANK: same rank for ties, skips next number
    RANK() OVER (ORDER BY population DESC) AS rank_pos,
    -- DENSE_RANK: same rank for ties, no gaps
    DENSE_RANK() OVER (ORDER BY population DESC) AS dense_rank_pos
FROM municipalities
ORDER BY population DESC
LIMIT 20;

-- Query 3: Population comparison between adjacent municipalities (by rank)
WITH ranked_municipalities AS (
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population,
        ROW_NUMBER() OVER (ORDER BY "AantalInwoners_5" DESC) AS rank_pos
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population,
    rank_pos,
    -- LAG: look at the previous row (larger municipality)
    LAG(municipality_name) OVER (ORDER BY rank_pos) AS larger_neighbor,
    LAG(population) OVER (ORDER BY rank_pos) AS larger_pop,
    -- LEAD: look at the next row (smaller municipality)
    LEAD(municipality_name) OVER (ORDER BY rank_pos) AS smaller_neighbor,
    -- Calculate the gap to the municipality above
    population - LEAD(population) OVER (ORDER BY rank_pos) AS gap_to_next
FROM ranked_municipalities
ORDER BY rank_pos
LIMIT 15;

-- Query 4: Running totals and averages with PARTITION BY
WITH municipality_data AS (
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population,
        CASE
            WHEN "AantalInwoners_5" >= 100000 THEN 'Large'
            WHEN "AantalInwoners_5" >= 25000 THEN 'Medium'
            ELSE 'Small'
        END AS size_category
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
)
SELECT
    municipality_name,
    population,
    size_category,
    -- Running total within each size category
    SUM(population) OVER (
        PARTITION BY size_category
        ORDER BY population DESC
    ) AS running_total,
    -- Average population within each category
    AVG(population) OVER (
        PARTITION BY size_category
    ) AS category_avg,
    -- How this municipality compares to its category average
    population - AVG(population) OVER (
        PARTITION BY size_category
    ) AS diff_from_avg
FROM municipality_data
ORDER BY size_category, population DESC;

-- Query 5: Comprehensive analysis combining all window function types
WITH municipality_stats AS (
    -- CTE: isolate and clean municipality data
    SELECT
        "Gemeentenaam_1" AS municipality_name,
        "AantalInwoners_5" AS population,
        CASE
            WHEN "AantalInwoners_5" >= 100000 THEN 'Large'
            WHEN "AantalInwoners_5" >= 25000 THEN 'Medium'
            ELSE 'Small'
        END AS size_category
    FROM cbs_neighborhoods
    WHERE "WijkenEnBuurten" LIKE 'GM%'
      AND "AantalInwoners_5" IS NOT NULL
),
ranked AS (
    -- CTE: add ranking within each category
    SELECT
        municipality_name,
        population,
        size_category,
        DENSE_RANK() OVER (
            PARTITION BY size_category
            ORDER BY population DESC
        ) AS rank_in_category,
        SUM(population) OVER (
            PARTITION BY size_category
            ORDER BY population DESC
        ) AS cumulative_pop,
        LAG(population) OVER (
            PARTITION BY size_category
            ORDER BY population DESC
        ) AS prev_municipality_pop
    FROM municipality_stats
)
SELECT
    municipality_name,
    population,
    size_category,
    rank_in_category,
    cumulative_pop,
    -- Percentage of category total accumulated so far
    ROUND(
        cumulative_pop * 100.0 / SUM(population) OVER (PARTITION BY size_category),
        1
    ) AS pct_of_category,
    -- Gap to the municipality ranked above (within same category)
    COALESCE(prev_municipality_pop - population, 0) AS gap_to_above
FROM ranked
WHERE rank_in_category <= 10
ORDER BY size_category, rank_in_category;

And here is the complete extraction/load_cbs_data.py:

import cbsodata
import pandas as pd
from sqlalchemy import create_engine

# Fetch CBS neighborhood key figures (table 83765NED)
# This contains population data for all Dutch municipalities, districts, and neighborhoods
print("Fetching CBS neighborhood data (this may take up to 30 seconds)...")
data = pd.DataFrame(cbsodata.get_data('83765NED'))

# Show what columns are available
print(f"Loaded {len(data)} rows with {len(data.columns)} columns")
print("Sample columns:", list(data.columns[:10]))

# Connect to PostgreSQL and load the data
engine = create_engine('postgresql+psycopg2://postgres:postgres@localhost:5432/dataeng')

# Write the full dataset to a table called cbs_neighborhoods
data.to_sql('cbs_neighborhoods', engine, if_exists='replace', index=False)
print("Data loaded into PostgreSQL table: cbs_neighborhoods")

You now have a solid foundation in CTEs and window functions on real Dutch data. Next up, you will push further into advanced SQL patterns like recursive CTEs, GROUPING SETS, and query performance optimization.

Optimize SQL Performance and Complex Patterns

Last week you built powerful analytical queries with window functions and CTEs. Those queries work, but as datasets grow from thousands to millions of rows, unoptimized queries can take minutes instead of milliseconds.

This week focuses on making your queries fast and adding advanced SQL patterns that data engineers use daily. You will learn to read query plans, create targeted indexes, and write complex multi-level reports in a single query using PostgreSQL features like recursive CTEs and GROUPING SETS.

In this step, get ready to:

• Measure and optimize query performance using EXPLAIN ANALYZE and three indexing strategies.
• Write advanced SQL patterns including recursive CTEs, GROUPING SETS, CUBE, ROLLUP, lateral joins, and array operations.
• Build a complete reporting query that combines recursive hierarchy, multi-dimensional aggregation, and demonstrable performance gains.

Measure and optimize query performance with EXPLAIN ANALYZE and indexing

When a query runs slowly, you need a way to see exactly what PostgreSQL is doing behind the scenes. The EXPLAIN ANALYZE command executes a query and reports the actual execution plan, including time spent at each step and how many rows were processed.

• Connect to your PostgreSQL database where the CBS neighborhood data is loaded.
• Run a baseline query against your CBS data without any indexes by running this in your PostgreSQL client:

-- Baseline: filter neighborhoods by municipality code without an index
EXPLAIN ANALYZE
SELECT *
FROM cbs_neighborhood_data
WHERE municipality_code = 'GM0363';

What does EXPLAIN ANALYZE show you?

The output shows each plan node with two key metrics: the estimated cost (in arbitrary units) and the actual time (in milliseconds). Look for the Seq Scan node type, which means PostgreSQL is reading every row in the table to find matches.

The most important numbers to compare are Planning Time and Execution Time at the bottom. Write down the Execution Time value for your before-index measurement.

• Note the Execution Time from the output. This is your "before" measurement.
• Create a B-tree index on the municipality code column by running:

-- B-tree index: speeds up equality and range lookups on municipality_code
CREATE INDEX idx_municipality_code
ON cbs_neighborhood_data (municipality_code);

What is a B-tree index?

A B-tree (balanced tree) index organizes column values into a sorted tree structure. PostgreSQL can jump directly to matching values instead of scanning every row. B-tree is the default index type and works best for equality checks and range comparisons.

• Run the same EXPLAIN ANALYZE query again to measure the "after" performance:

-- After index: same query, now with the B-tree index available
EXPLAIN ANALYZE
SELECT *
FROM cbs_neighborhood_data
WHERE municipality_code = 'GM0363';

You should now see Index Scan instead of Seq Scan in the plan output. The Execution Time should be significantly lower.

• Create a partial index that only covers rows with high population values by running:

-- Partial index: only indexes neighborhoods with population above 10000
CREATE INDEX idx_high_population
ON cbs_neighborhood_data (population)
WHERE population > 10000;

Why use a partial index?

A partial index only stores entries for rows that match the WHERE condition. This makes the index smaller and faster to update. Use partial indexes when your queries consistently filter on a specific subset of data.

• Create a composite index that covers multiple columns used together in queries:

-- Composite index: covers queries that filter by municipality AND sort by population
CREATE INDEX idx_municipality_population
ON cbs_neighborhood_data (municipality_code, population);

When to use composite indexes

Composite indexes cover queries that filter or sort on multiple columns together. Column order matters. Put the column you filter with equality first, and the column you use for ranges or sorting second.

• Verify your indexes were created by running:

-- List all indexes on the CBS table
\di cbs_neighborhood_data*

You should see three new indexes listed alongside any default primary key index.

Not seeing a performance improvement?

If the query planner still chooses a Seq Scan after creating an index, your table may be too small for the planner to prefer the index. PostgreSQL decides that scanning a small table is faster than the overhead of index lookups. Try running ANALYZE cbs_neighborhood_data; to update table statistics.

help me understand why PostgreSQL is not using my index

Write advanced SQL patterns for complex reporting

Production data engineering requires patterns beyond basic queries. Recursive CTEs let you traverse hierarchies. GROUPING SETS, CUBE, and ROLLUP produce multi-level aggregations in a single pass.

• Practice a recursive CTE that builds a hierarchical structure from the CBS data. Run this query to simulate building a region hierarchy:

-- Recursive CTE: build a hierarchy of regions
-- CBS data has province > municipality > neighborhood structure
WITH RECURSIVE region_hierarchy AS (
    -- Base case: start with provinces (top level)
    SELECT
        municipality_code AS region_code,
        municipality_name AS region_name,
        1 AS level,
        municipality_name AS path
    FROM cbs_neighborhood_data
    WHERE neighborhood_code IS NULL
    GROUP BY municipality_code, municipality_name

    UNION ALL

    -- Recursive case: add neighborhoods under their municipality
    SELECT
        n.neighborhood_code AS region_code,
        n.neighborhood_name AS region_name,
        rh.level + 1 AS level,
        rh.path || ' > ' || n.neighborhood_name AS path
    FROM cbs_neighborhood_data n
    INNER JOIN region_hierarchy rh
        ON n.municipality_code = rh.region_code
    WHERE n.neighborhood_code IS NOT NULL
)
SELECT region_code, region_name, level, path
FROM region_hierarchy
ORDER BY path
LIMIT 20;

How does a recursive CTE work?

A recursive CTE has two parts connected by UNION ALL. The base case runs once to produce the starting rows. The recursive case references the CTE itself and repeats until no new rows are produced.

This pattern is essential for hierarchical data like org charts, category trees, or geographic hierarchies. Each iteration adds one more level of depth.

• Practice GROUPING SETS to produce multiple aggregation levels in a single query:

-- GROUPING SETS: aggregate population at multiple levels simultaneously
SELECT
    municipality_code,
    neighborhood_code,
    GROUPING(municipality_code, neighborhood_code) AS grouping_level,
    SUM(population) AS total_population,
    COUNT(*) AS area_count
FROM cbs_neighborhood_data
WHERE municipality_code IN ('GM0363', 'GM0518', 'GM0599')
GROUP BY GROUPING SETS (
    (municipality_code, neighborhood_code),  -- detail level
    (municipality_code),                     -- municipality subtotal
    ()                                       -- grand total
)
ORDER BY municipality_code NULLS LAST, neighborhood_code NULLS LAST;

What do the GROUPING values mean?

The GROUPING function returns a bitmask showing which columns are NOT included in the current grouping set. A value of 0 means the row is grouped by all specified columns (detail level). A value of 1 means one column is excluded (subtotal). A value of 3 means both columns are excluded (grand total).

This lets you produce subtotals and grand totals alongside detail rows in a single query pass.

• Practice ROLLUP for hierarchical totals. This is a shorthand for the most common GROUPING SETS pattern:

-- ROLLUP: hierarchical aggregation (municipality > neighborhood > total)
SELECT
    municipality_code,
    neighborhood_code,
    SUM(population) AS total_population,
    AVG(population) AS avg_population
FROM cbs_neighborhood_data
WHERE municipality_code IN ('GM0363', 'GM0518')
GROUP BY ROLLUP (municipality_code, neighborhood_code)
ORDER BY municipality_code NULLS LAST, neighborhood_code NULLS LAST;

ROLLUP vs CUBE vs GROUPING SETS

ROLLUP produces hierarchical totals: it includes all prefixes of the column list plus the empty set. So ROLLUP(a, b) gives you groupings (a, b), (a), and ().

CUBE produces all possible subsets (the power set). CUBE(a, b) gives you (a, b), (a), (b), and (). Use CUBE when you need every combination for cross-dimensional analysis.

• Practice a lateral join to compute per-row calculations from a correlated subquery:

-- Lateral join: get the top 3 most populated neighborhoods per municipality
SELECT
    m.municipality_code,
    m.municipality_name,
    top_n.neighborhood_name,
    top_n.population
FROM (
    SELECT DISTINCT municipality_code, municipality_name
    FROM cbs_neighborhood_data
    WHERE municipality_code IN ('GM0363', 'GM0518', 'GM0599')
) m
CROSS JOIN LATERAL (
    SELECT neighborhood_name, population
    FROM cbs_neighborhood_data n
    WHERE n.municipality_code = m.municipality_code
        AND n.neighborhood_code IS NOT NULL
    ORDER BY population DESC
    LIMIT 3
) top_n
ORDER BY m.municipality_code, top_n.population DESC;

What is a lateral join?

A lateral join lets the right-hand subquery reference columns from the left-hand table. This enables "for each row, compute something" patterns that are impossible with regular joins.

Common use cases include top-N per group, running calculations per row, or unpacking complex values. Lateral joins often replace complex window function workarounds.

• Practice array operations to aggregate values into arrays for compact reporting:

-- Array operations: collect neighborhood names into an array per municipality
SELECT
    municipality_code,
    municipality_name,
    array_agg(neighborhood_name ORDER BY population DESC) AS neighborhoods_by_pop,
    array_length(array_agg(neighborhood_name), 1) AS neighborhood_count
FROM cbs_neighborhood_data
WHERE neighborhood_code IS NOT NULL
    AND municipality_code IN ('GM0363', 'GM0518')
GROUP BY municipality_code, municipality_name
ORDER BY municipality_code;

Why use array operations in SQL?

PostgreSQL arrays let you collect multiple values into a single cell. This is useful for building compact reports, passing structured data to application code, or performing set-based comparisons. The array_agg function collects grouped values into an array, and you can control the order of elements.

Build the Week 2 reporting query exercise

Now combine everything into a single exercise. You will write a reporting query that uses recursive CTEs for hierarchy building, GROUPING SETS for multi-dimensional analysis, and demonstrates performance improvement through the indexes you created.

• Create the file week2_advanced_patterns.sql in your project-repo/sql/ folder.
• Write your complete reporting query that meets these requirements:

• Uses a recursive CTE to build a municipality-to-neighborhood hierarchy.
• Uses GROUPING SETS (or ROLLUP/CUBE) for multi-dimensional population analysis.
• Demonstrates measurable performance improvement by including EXPLAIN ANALYZE output as SQL comments showing before-and-after index timings.
• Includes at least one lateral join or array operation for advanced data shaping.

• Document your query plan analysis as comments at the top of the file. Include the before-index Execution Time and after-index Execution Time:

-- =============================================================
-- Week 2: Advanced SQL Patterns & Performance Optimization
-- =============================================================
-- PERFORMANCE ANALYSIS:
-- Query: SELECT * FROM cbs_neighborhood_data WHERE municipality_code = 'GM0363'
-- Before index: Seq Scan, Execution Time: [YOUR BEFORE TIME] ms
-- After index:  Index Scan using idx_municipality_code, Execution Time: [YOUR AFTER TIME] ms
-- Improvement: [CALCULATE PERCENTAGE]%
-- =============================================================

-- Part 1: Recursive CTE - Build region hierarchy
WITH RECURSIVE region_hierarchy AS (
    -- Base case: municipalities as top level
    SELECT
        municipality_code AS region_code,
        municipality_name AS region_name,
        1 AS level,
        municipality_name AS path
    FROM cbs_neighborhood_data
    WHERE neighborhood_code IS NULL
    GROUP BY municipality_code, municipality_name

    UNION ALL

    -- Recursive case: neighborhoods under municipalities
    SELECT
        n.neighborhood_code AS region_code,
        n.neighborhood_name AS region_name,
        rh.level + 1 AS level,
        rh.path || ' > ' || n.neighborhood_name AS path
    FROM cbs_neighborhood_data n
    INNER JOIN region_hierarchy rh
        ON n.municipality_code = rh.region_code
    WHERE n.neighborhood_code IS NOT NULL
)
SELECT region_code, region_name, level, path
FROM region_hierarchy
ORDER BY path;

-- Part 2: GROUPING SETS - Multi-dimensional population analysis
SELECT
    municipality_code,
    neighborhood_code,
    GROUPING(municipality_code, neighborhood_code) AS grouping_level,
    SUM(population) AS total_population,
    COUNT(*) AS area_count,
    AVG(population) AS avg_population
FROM cbs_neighborhood_data
GROUP BY GROUPING SETS (
    (municipality_code, neighborhood_code),
    (municipality_code),
    ()
)
ORDER BY municipality_code NULLS LAST, neighborhood_code NULLS LAST;

-- Part 3: Lateral join - Top 3 neighborhoods per municipality
SELECT
    m.municipality_code,
    m.municipality_name,
    top_n.neighborhood_name,
    top_n.population,
    top_n.rank_in_municipality
FROM (
    SELECT DISTINCT municipality_code, municipality_name
    FROM cbs_neighborhood_data
) m
CROSS JOIN LATERAL (
    SELECT
        neighborhood_name,
        population,
        ROW_NUMBER() OVER (ORDER BY population DESC) AS rank_in_municipality
    FROM cbs_neighborhood_data n
    WHERE n.municipality_code = m.municipality_code
        AND n.neighborhood_code IS NOT NULL
    ORDER BY population DESC
    LIMIT 3
) top_n
ORDER BY m.municipality_code, top_n.rank_in_municipality;

-- Part 4: Array operations - Compact municipality summary
SELECT
    municipality_code,
    municipality_name,
    array_agg(neighborhood_name ORDER BY population DESC) AS neighborhoods_by_pop,
    array_length(array_agg(neighborhood_name), 1) AS neighborhood_count,
    SUM(population) AS total_population
FROM cbs_neighborhood_data
WHERE neighborhood_code IS NOT NULL
GROUP BY municipality_code, municipality_name
ORDER BY total_population DESC
LIMIT 10;

What does this exercise prove?

This query demonstrates four key data engineering skills in one file. The recursive CTE shows you can traverse hierarchies. The GROUPING SETS show you can produce multi-level reports without multiple queries. The lateral join shows you can do per-group top-N analysis. The documented performance analysis shows you understand query optimization.

These are exactly the patterns that Dutch companies like ABN AMRO and KPMG test for in data engineering interviews.

• Save week2_advanced_patterns.sql to your project-repo/sql/ directory.
• Run each part of the query in your PostgreSQL client to confirm it executes without errors.
• Fill in your actual EXPLAIN ANALYZE timing values in the performance analysis comments at the top of the file.

• Commit your work to Git by running:

cd ~/Desktop/project-repo
git add sql/week2_advanced_patterns.sql
git commit -m "Week 2: Advanced SQL patterns and performance optimization"

Getting a recursive CTE infinite loop?

If your recursive query runs forever, the join condition in the recursive case is not narrowing the result set. Make sure the WHERE clause excludes rows that were already selected in the base case. Adding a LIMIT to the outer query is a safety measure while developing.

help me debug my recursive CTE that runs forever

✔️ Awesome, I've got everything!

Great. Double check you have saved week2_advanced_patterns.sql with your actual performance measurements filled in.

ⓧ I'd like to double check the full code

Here is the complete week2_advanced_patterns.sql file for reference:

-- =============================================================
-- Week 2: Advanced SQL Patterns & Performance Optimization
-- =============================================================
-- PERFORMANCE ANALYSIS:
-- Query: SELECT * FROM cbs_neighborhood_data WHERE municipality_code = 'GM0363'
-- Before index: Seq Scan, Execution Time: [YOUR BEFORE TIME] ms
-- After index:  Index Scan using idx_municipality_code, Execution Time: [YOUR AFTER TIME] ms
-- Improvement: [CALCULATE PERCENTAGE]%
-- =============================================================

-- Part 1: Recursive CTE - Build region hierarchy
WITH RECURSIVE region_hierarchy AS (
    -- Base case: municipalities as top level
    SELECT
        municipality_code AS region_code,
        municipality_name AS region_name,
        1 AS level,
        municipality_name AS path
    FROM cbs_neighborhood_data
    WHERE neighborhood_code IS NULL
    GROUP BY municipality_code, municipality_name

    UNION ALL

    -- Recursive case: neighborhoods under municipalities
    SELECT
        n.neighborhood_code AS region_code,
        n.neighborhood_name AS region_name,
        rh.level + 1 AS level,
        rh.path || ' > ' || n.neighborhood_name AS path
    FROM cbs_neighborhood_data n
    INNER JOIN region_hierarchy rh
        ON n.municipality_code = rh.region_code
    WHERE n.neighborhood_code IS NOT NULL
)
SELECT region_code, region_name, level, path
FROM region_hierarchy
ORDER BY path;

-- Part 2: GROUPING SETS - Multi-dimensional population analysis
SELECT
    municipality_code,
    neighborhood_code,
    GROUPING(municipality_code, neighborhood_code) AS grouping_level,
    SUM(population) AS total_population,
    COUNT(*) AS area_count,
    AVG(population) AS avg_population
FROM cbs_neighborhood_data
GROUP BY GROUPING SETS (
    (municipality_code, neighborhood_code),
    (municipality_code),
    ()
)
ORDER BY municipality_code NULLS LAST, neighborhood_code NULLS LAST;

-- Part 3: Lateral join - Top 3 neighborhoods per municipality
SELECT
    m.municipality_code,
    m.municipality_name,
    top_n.neighborhood_name,
    top_n.population,
    top_n.rank_in_municipality
FROM (
    SELECT DISTINCT municipality_code, municipality_name
    FROM cbs_neighborhood_data
) m
CROSS JOIN LATERAL (
    SELECT
        neighborhood_name,
        population,
        ROW_NUMBER() OVER (ORDER BY population DESC) AS rank_in_municipality
    FROM cbs_neighborhood_data n
    WHERE n.municipality_code = m.municipality_code
        AND n.neighborhood_code IS NOT NULL
    ORDER BY population DESC
    LIMIT 3
) top_n
ORDER BY m.municipality_code, top_n.rank_in_municipality;

-- Part 4: Array operations - Compact municipality summary
SELECT
    municipality_code,
    municipality_name,
    array_agg(neighborhood_name ORDER BY population DESC) AS neighborhoods_by_pop,
    array_length(array_agg(neighborhood_name), 1) AS neighborhood_count,
    SUM(population) AS total_population
FROM cbs_neighborhood_data
WHERE neighborhood_code IS NOT NULL
GROUP BY municipality_code, municipality_name
ORDER BY total_population DESC
LIMIT 10;

Your SQL toolkit now includes performance profiling and advanced patterns that most junior engineers never learn. Next up, you will design a dimensional data model for the Dutch energy balance dataset and implement it in PostgreSQL.

Design a Dimensional Data Model

In the previous weeks you built advanced SQL skills with window functions, CTEs, and performance optimization. Now it is time to apply a structured approach to organizing data itself.

Raw data in a single flat table is hard to analyze from multiple angles. Dimensional modeling solves this by splitting data into fact tables (measurements) and dimension tables (context). In this step, you will design and implement a star schema for the CBS energy balance dataset.

In this step, get ready to:

• Learn dimensional modeling theory and the Kimball methodology.
• Design and implement a star schema for CBS table 83140ENG in PostgreSQL.
• Write analytical queries that demonstrate slice-and-dice analysis.

Learn dimensional modeling theory

Before writing any DDL, you need to understand the building blocks of a dimensional model. The Kimball methodology is the industry standard for designing analytical databases.

What is a star schema?

A star schema organizes data around a central fact table surrounded by dimension tables. The fact table holds numeric measurements (things you count or sum). Dimension tables hold descriptive context (the who, what, when, where).

It is called a star schema because when you diagram it, the fact table in the center with dimensions radiating outward looks like a star. This shape makes queries fast and intuitive because you join the fact table to only the dimensions you need.

How does this apply to CBS energy data?

For the CBS energy balance dataset (table 83140ENG), the fact is energy flow measured in terajoules. The dimensions are: what type of energy (carrier), which sector consumes it, and when the measurement was taken (time period).

The grain of your fact table is one row per energy carrier, per sector, per time period. Defining the grain first prevents you from accidentally mixing different levels of detail in one table.

Implement the star schema in PostgreSQL

Now you will create the dimension tables and fact table. Dimension tables are created first because the fact table references them with foreign keys.

• Create a new file called week3_dimensional_model.sql inside your project-repo/sql/ folder.
• Add the dimension table DDL by copying this code into the file:

-- Week 3: Dimensional Data Model - Dutch Energy Balance (CBS 83140ENG)
-- Star schema design following Kimball methodology

-- Drop existing tables (CASCADE removes dependent foreign keys)
DROP TABLE IF EXISTS fact_energy_consumption CASCADE;
DROP TABLE IF EXISTS dim_energy_carrier CASCADE;
DROP TABLE IF EXISTS dim_sector CASCADE;
DROP TABLE IF EXISTS dim_time CASCADE;

-- Dimension: Energy Carrier
-- Describes the type of energy source (coal, natural gas, wind, solar, etc.)
CREATE TABLE dim_energy_carrier (
    carrier_key SERIAL PRIMARY KEY,
    carrier_code VARCHAR(50) NOT NULL UNIQUE,
    carrier_name VARCHAR(200) NOT NULL,
    carrier_category VARCHAR(100),
    is_renewable BOOLEAN DEFAULT FALSE
);

-- Dimension: Sector
-- Describes which sector of the economy consumes the energy
CREATE TABLE dim_sector (
    sector_key SERIAL PRIMARY KEY,
    sector_code VARCHAR(50) NOT NULL UNIQUE,
    sector_name VARCHAR(200) NOT NULL,
    sector_group VARCHAR(100)
);

-- Dimension: Time
-- Describes the time period of the measurement
CREATE TABLE dim_time (
    time_key SERIAL PRIMARY KEY,
    period_code VARCHAR(20) NOT NULL UNIQUE,
    year INTEGER NOT NULL,
    month INTEGER,
    quarter INTEGER,
    is_annual BOOLEAN DEFAULT FALSE
);

What does this code do?

• Each dimension table has a surrogate key (SERIAL PRIMARY KEY) that the fact table will reference. This decouples the warehouse from source system IDs.
• The UNIQUE constraint on business codes prevents duplicate dimension entries.
• The is_renewable flag on dim_energy_carrier enables quick filtering for energy transition analysis.
• The dim_time table breaks period codes into queryable components (year, month, quarter).

• Add the fact table DDL below the dimension tables in the same file:

-- Fact: Energy Consumption
-- Grain: one row per energy carrier, per sector, per time period
-- Measures are in terajoules (TJ)
CREATE TABLE fact_energy_consumption (
    consumption_key SERIAL PRIMARY KEY,
    carrier_key INTEGER NOT NULL REFERENCES dim_energy_carrier(carrier_key),
    sector_key INTEGER NOT NULL REFERENCES dim_sector(sector_key),
    time_key INTEGER NOT NULL REFERENCES dim_time(time_key),
    supply_tj NUMERIC(15,2),
    consumption_tj NUMERIC(15,2),
    transformation_tj NUMERIC(15,2)
);

-- Indexes on foreign keys for fast joins
CREATE INDEX idx_fact_carrier ON fact_energy_consumption(carrier_key);
CREATE INDEX idx_fact_sector ON fact_energy_consumption(sector_key);
CREATE INDEX idx_fact_time ON fact_energy_consumption(time_key);

Why index the foreign keys?

PostgreSQL does not automatically index foreign key columns. Without these indexes, every join between the fact table and a dimension requires a full table scan. Adding indexes makes star schema queries fast even with millions of rows.

The three measure columns (supply_tj, consumption_tj, transformation_tj) store the numeric facts that analysts aggregate. Using NUMERIC(15,2) ensures precision for terajoule values.

• Save week3_dimensional_model.sql.
• Run the DDL in your PostgreSQL database to create all four tables. Connect to your database and execute the file:

psql -f project-repo/sql/week3_dimensional_model.sql

• Verify the tables were created by running:

psql -c "\dt dim_* fact_*"

You should see four tables listed: dim_energy_carrier, dim_sector, dim_time, and fact_energy_consumption.

Tables not showing up?

• Check that psql connected to the correct database. You may need to add -d your_database_name to the command.
• Look for SQL errors in the output. A missing semicolon or typo in the DDL will stop execution.
• If you see "relation already exists" errors, the DROP TABLE IF EXISTS statements at the top should prevent this. Make sure they are included.

Why aren't my tables showing up in PostgreSQL?

Load sample data and write analytical queries

A star schema proves its value when you query it. In this exercise, you will load sample data representing Dutch household energy consumption and write three queries that demonstrate how the schema enables flexible analysis.

• Add the sample data inserts below your DDL in week3_dimensional_model.sql:

-- ============================================================
-- SAMPLE DATA LOAD
-- Representative data for Dutch household energy analysis
-- ============================================================

-- Load energy carriers
INSERT INTO dim_energy_carrier (carrier_code, carrier_name, carrier_category, is_renewable) VALUES
('E001', 'Natural gas', 'Fossil', FALSE),
('E002', 'Electricity', 'Secondary', FALSE),
('E003', 'Solar energy', 'Renewable', TRUE),
('E004', 'Wind energy', 'Renewable', TRUE),
('E005', 'Coal and coal products', 'Fossil', FALSE),
('E006', 'Biomass', 'Renewable', TRUE),
('E007', 'Heat from district heating', 'Secondary', FALSE);

-- Load sectors
INSERT INTO dim_sector (sector_code, sector_name, sector_group) VALUES
('S001', 'Households', 'Final consumption'),
('S002', 'Industry', 'Final consumption'),
('S003', 'Transport', 'Final consumption'),
('S004', 'Agriculture', 'Final consumption'),
('S005', 'Services', 'Final consumption'),
('S006', 'Electricity production', 'Transformation');

-- Load time periods (annual data 2018-2023)
INSERT INTO dim_time (period_code, year, month, quarter, is_annual) VALUES
('2018JJ00', 2018, NULL, NULL, TRUE),
('2019JJ00', 2019, NULL, NULL, TRUE),
('2020JJ00', 2020, NULL, NULL, TRUE),
('2021JJ00', 2021, NULL, NULL, TRUE),
('2022JJ00', 2022, NULL, NULL, TRUE),
('2023JJ00', 2023, NULL, NULL, TRUE);

-- Load fact data (energy values in terajoules)
INSERT INTO fact_energy_consumption (carrier_key, sector_key, time_key, supply_tj, consumption_tj, transformation_tj)
SELECT
    c.carrier_key, s.sector_key, t.time_key,
    vals.supply_tj, vals.consumption_tj, vals.transformation_tj
FROM (VALUES
    ('E001', 'S001', '2018JJ00', 950.00, 920.50, 0.00),
    ('E001', 'S001', '2019JJ00', 930.00, 905.00, 0.00),
    ('E001', 'S001', '2020JJ00', 960.00, 940.75, 0.00),
    ('E001', 'S001', '2021JJ00', 910.00, 885.25, 0.00),
    ('E001', 'S001', '2022JJ00', 820.00, 795.00, 0.00),
    ('E001', 'S001', '2023JJ00', 780.00, 760.50, 0.00),
    ('E002', 'S001', '2018JJ00', 320.00, 315.00, 0.00),
    ('E002', 'S001', '2019JJ00', 325.00, 320.00, 0.00),
    ('E002', 'S001', '2020JJ00', 340.00, 335.50, 0.00),
    ('E002', 'S001', '2021JJ00', 345.00, 340.00, 0.00),
    ('E002', 'S001', '2022JJ00', 330.00, 325.75, 0.00),
    ('E002', 'S001', '2023JJ00', 335.00, 330.00, 0.00),
    ('E003', 'S001', '2018JJ00', 15.00, 14.50, 0.00),
    ('E003', 'S001', '2019JJ00', 20.00, 19.50, 0.00),
    ('E003', 'S001', '2020JJ00', 28.00, 27.00, 0.00),
    ('E003', 'S001', '2021JJ00', 38.00, 37.00, 0.00),
    ('E003', 'S001', '2022JJ00', 52.00, 50.50, 0.00),
    ('E003', 'S001', '2023JJ00', 68.00, 66.00, 0.00),
    ('E001', 'S002', '2018JJ00', 480.00, 465.00, 0.00),
    ('E001', 'S002', '2019JJ00', 460.00, 445.00, 0.00),
    ('E001', 'S002', '2020JJ00', 420.00, 405.50, 0.00),
    ('E001', 'S002', '2021JJ00', 440.00, 425.00, 0.00),
    ('E001', 'S002', '2022JJ00', 390.00, 375.50, 0.00),
    ('E001', 'S002', '2023JJ00', 370.00, 355.00, 0.00),
    ('E004', 'S006', '2018JJ00', 0.00, 0.00, 95.00),
    ('E004', 'S006', '2019JJ00', 0.00, 0.00, 125.00),
    ('E004', 'S006', '2020JJ00', 0.00, 0.00, 160.00),
    ('E004', 'S006', '2021JJ00', 0.00, 0.00, 190.00),
    ('E004', 'S006', '2022JJ00', 0.00, 0.00, 230.00),
    ('E004', 'S006', '2023JJ00', 0.00, 0.00, 275.00)
) AS vals(carrier_code, sector_code, period_code, supply_tj, consumption_tj, transformation_tj)
JOIN dim_energy_carrier c ON c.carrier_code = vals.carrier_code
JOIN dim_sector s ON s.sector_code = vals.sector_code
JOIN dim_time t ON t.period_code = vals.period_code;

What does this code do?

The INSERT INTO ... SELECT ... JOIN pattern for the fact table is important. Instead of hard-coding surrogate key values, it looks up the correct keys by joining on business codes. This is the same pattern you would use when loading real data from an extraction pipeline.

The sample data covers household natural gas consumption declining over 6 years, electricity holding steady, and solar energy growing rapidly. This mirrors the real Dutch energy transition trends visible in CBS data.

• Now add three analytical queries that demonstrate how the star schema enables different perspectives on the same data. Add these below the sample data inserts:

-- ============================================================
-- ANALYTICAL QUERIES: Slice-and-dice demonstration
-- Business scenario: Dutch household energy costs over time
-- ============================================================

-- Query 1: Year-over-year household consumption by energy carrier
-- Slice by: sector (Households only), all carriers, all years
SELECT
    t.year,
    c.carrier_name,
    c.carrier_category,
    f.consumption_tj,
    LAG(f.consumption_tj) OVER (
        PARTITION BY c.carrier_key ORDER BY t.year
    ) AS prev_year_tj,
    ROUND(
        (f.consumption_tj - LAG(f.consumption_tj) OVER (
            PARTITION BY c.carrier_key ORDER BY t.year
        )) / NULLIF(LAG(f.consumption_tj) OVER (
            PARTITION BY c.carrier_key ORDER BY t.year
        ), 0) * 100, 1
    ) AS yoy_change_pct
FROM fact_energy_consumption f
JOIN dim_energy_carrier c ON f.carrier_key = c.carrier_key
JOIN dim_sector s ON f.sector_key = s.sector_key
JOIN dim_time t ON f.time_key = t.time_key
WHERE s.sector_name = 'Households'
    AND t.is_annual = TRUE
ORDER BY c.carrier_name, t.year;

-- Query 2: Renewable vs fossil share in household consumption
-- Dice by: renewable flag, aggregated across carriers per year
SELECT
    t.year,
    c.is_renewable,
    CASE WHEN c.is_renewable THEN 'Renewable' ELSE 'Non-renewable' END AS energy_type,
    SUM(f.consumption_tj) AS total_consumption_tj,
    ROUND(
        SUM(f.consumption_tj) / SUM(SUM(f.consumption_tj)) OVER (PARTITION BY t.year) * 100, 1
    ) AS share_pct
FROM fact_energy_consumption f
JOIN dim_energy_carrier c ON f.carrier_key = c.carrier_key
JOIN dim_sector s ON f.sector_key = s.sector_key
JOIN dim_time t ON f.time_key = t.time_key
WHERE s.sector_name = 'Households'
    AND t.is_annual = TRUE
GROUP BY t.year, c.is_renewable
ORDER BY t.year, c.is_renewable;

-- Query 3: Cross-sector comparison with running totals
-- Drill across: all sectors, natural gas only, cumulative decline
WITH sector_yearly AS (
    SELECT
        s.sector_name,
        t.year,
        SUM(f.consumption_tj) AS yearly_consumption_tj
    FROM fact_energy_consumption f
    JOIN dim_energy_carrier c ON f.carrier_key = c.carrier_key
    JOIN dim_sector s ON f.sector_key = s.sector_key
    JOIN dim_time t ON f.time_key = t.time_key
    WHERE c.carrier_name = 'Natural gas'
        AND t.is_annual = TRUE
    GROUP BY s.sector_name, t.year
)
SELECT
    sector_name,
    year,
    yearly_consumption_tj,
    SUM(yearly_consumption_tj) OVER (
        PARTITION BY sector_name ORDER BY year
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS cumulative_tj,
    RANK() OVER (PARTITION BY year ORDER BY yearly_consumption_tj DESC) AS sector_rank
FROM sector_yearly
ORDER BY year, sector_rank;

What do these queries demonstrate?

• Query 1 slices the fact table to households only, then uses LAG to calculate year-over-year percentage change per energy carrier.
• Query 2 dices by the renewable flag on the carrier dimension, showing how the energy mix shifts over time with percentage shares.
• Query 3 drills across sectors using a CTE, then calculates cumulative totals and ranks sectors by gas consumption each year.

These three patterns (slice, dice, drill-across) are the core analytical operations that star schemas are optimized for. Each query joins the fact table to only the dimensions it needs, which is exactly why star schemas outperform flat tables for analytics.

• Save week3_dimensional_model.sql.
• Run the complete file again to load the sample data and execute the queries:

psql -f project-repo/sql/week3_dimensional_model.sql

You should see the INSERT confirmations followed by three result sets. Query 1 shows household gas consumption declining year over year. Query 2 shows the renewable share growing from roughly 1% to over 5%. Query 3 ranks sectors by natural gas usage with cumulative totals.

Seeing NULL values in the year-over-year column?

The first year for each carrier will always show NULL in the yoy_change_pct column because LAG has no previous row to compare. This is expected behavior, not an error.

If ALL values are NULL, check that your WHERE clause matches the data. The sector name must be exactly 'Households' (capital H) as inserted in the sample data.

Help me debug NULL values in my star schema analytical queries.

✔️ Awesome, I've got everything!

Great. Double check you have saved week3_dimensional_model.sql and all four tables exist in your PostgreSQL database.

ⓧ I'd like to double check the full code

-- Week 3: Dimensional Data Model - Dutch Energy Balance (CBS 83140ENG)
-- Star schema design following Kimball methodology

-- Drop existing tables (CASCADE removes dependent foreign keys)
DROP TABLE IF EXISTS fact_energy_consumption CASCADE;
DROP TABLE IF EXISTS dim_energy_carrier CASCADE;
DROP TABLE IF EXISTS dim_sector CASCADE;
DROP TABLE IF EXISTS dim_time CASCADE;

-- Dimension: Energy Carrier
-- Describes the type of energy source (coal, natural gas, wind, solar, etc.)
CREATE TABLE dim_energy_carrier (
    carrier_key SERIAL PRIMARY KEY,
    carrier_code VARCHAR(50) NOT NULL UNIQUE,
    carrier_name VARCHAR(200) NOT NULL,
    carrier_category VARCHAR(100),
    is_renewable BOOLEAN DEFAULT FALSE
);

-- Dimension: Sector
-- Describes which sector of the economy consumes the energy
CREATE TABLE dim_sector (
    sector_key SERIAL PRIMARY KEY,
    sector_code VARCHAR(50) NOT NULL UNIQUE,
    sector_name VARCHAR(200) NOT NULL,
    sector_group VARCHAR(100)
);

-- Dimension: Time
-- Describes the time period of the measurement
CREATE TABLE dim_time (
    time_key SERIAL PRIMARY KEY,
    period_code VARCHAR(20) NOT NULL UNIQUE,
    year INTEGER NOT NULL,
    month INTEGER,
    quarter INTEGER,
    is_annual BOOLEAN DEFAULT FALSE
);

-- Fact: Energy Consumption
-- Grain: one row per energy carrier, per sector, per time period
-- Measures are in terajoules (TJ)
CREATE TABLE fact_energy_consumption (
    consumption_key SERIAL PRIMARY KEY,
    carrier_key INTEGER NOT NULL REFERENCES dim_energy_carrier(carrier_key),
    sector_key INTEGER NOT NULL REFERENCES dim_sector(sector_key),
    time_key INTEGER NOT NULL REFERENCES dim_time(time_key),
    supply_tj NUMERIC(15,2),
    consumption_tj NUMERIC(15,2),
    transformation_tj NUMERIC(15,2)
);

-- Indexes on foreign keys for fast joins
CREATE INDEX idx_fact_carrier ON fact_energy_consumption(carrier_key);
CREATE INDEX idx_fact_sector ON fact_energy_consumption(sector_key);
CREATE INDEX idx_fact_time ON fact_energy_consumption(time_key);

-- ============================================================
-- SAMPLE DATA LOAD
-- Representative data for Dutch household energy analysis
-- ============================================================

-- Load energy carriers
INSERT INTO dim_energy_carrier (carrier_code, carrier_name, carrier_category, is_renewable) VALUES
('E001', 'Natural gas', 'Fossil', FALSE),
('E002', 'Electricity', 'Secondary', FALSE),
('E003', 'Solar energy', 'Renewable', TRUE),
('E004', 'Wind energy', 'Renewable', TRUE),
('E005', 'Coal and coal products', 'Fossil', FALSE),
('E006', 'Biomass', 'Renewable', TRUE),
('E007', 'Heat from district heating', 'Secondary', FALSE);

-- Load sectors
INSERT INTO dim_sector (sector_code, sector_name, sector_group) VALUES
('S001', 'Households', 'Final consumption'),
('S002', 'Industry', 'Final consumption'),
('S003', 'Transport', 'Final consumption'),
('S004', 'Agriculture', 'Final consumption'),
('S005', 'Services', 'Final consumption'),
('S006', 'Electricity production', 'Transformation');

-- Load time periods (annual data 2018-2023)
INSERT INTO dim_time (period_code, year, month, quarter, is_annual) VALUES
('2018JJ00', 2018, NULL, NULL, TRUE),
('2019JJ00', 2019, NULL, NULL, TRUE),
('2020JJ00', 2020, NULL, NULL, TRUE),
('2021JJ00', 2021, NULL, NULL, TRUE),
('2022JJ00', 2022, NULL, NULL, TRUE),
('2023JJ00', 2023, NULL, NULL, TRUE);

-- Load fact data (energy values in terajoules)
INSERT INTO fact_energy_consumption (carrier_key, sector_key, time_key, supply_tj, consumption_tj, transformation_tj)
SELECT
    c.carrier_key, s.sector_key, t.time_key,
    vals.supply_tj, vals.consumption_tj, vals.transformation_tj
FROM (VALUES
    ('E001', 'S001', '2018JJ00', 950.00, 920.50, 0.00),
    ('E001', 'S001', '2019JJ00', 930.00, 905.00, 0.00),
    ('E001', 'S001', '2020JJ00', 960.00, 940.75, 0.00),
    ('E001', 'S001', '2021JJ00', 910.00, 885.25, 0.00),
    ('E001', 'S001', '2022JJ00', 820.00, 795.00, 0.00),
    ('E001', 'S001', '2023JJ00', 780.00, 760.50, 0.00),
    ('E002', 'S001', '2018JJ00', 320.00, 315.00, 0.00),
    ('E002', 'S001', '2019JJ00', 325.00, 320.00, 0.00),
    ('E002', 'S001', '2020JJ00', 340.00, 335.50, 0.00),
    ('E002', 'S001', '2021JJ00', 345.00, 340.00, 0.00),
    ('E002', 'S001', '2022JJ00', 330.00, 325.75, 0.00),
    ('E002', 'S001', '2023JJ00', 335.00, 330.00, 0.00),
    ('E003', 'S001', '2018JJ00', 15.00, 14.50, 0.00),
    ('E003', 'S001', '2019JJ00', 20.00, 19.50, 0.00),
    ('E003', 'S001', '2020JJ00', 28.00, 27.00, 0.00),
    ('E003', 'S001', '2021JJ00', 38.00, 37.00, 0.00),
    ('E003', 'S001', '2022JJ00', 52.00, 50.50, 0.00),
    ('E003', 'S001', '2023JJ00', 68.00, 66.00, 0.00),
    ('E001', 'S002', '2018JJ00', 480.00, 465.00, 0.00),
    ('E001', 'S002', '2019JJ00', 460.00, 445.00, 0.00),
    ('E001', 'S002', '2020JJ00', 420.00, 405.50, 0.00),
    ('E001', 'S002', '2021JJ00', 440.00, 425.00, 0.00),
    ('E001', 'S002', '2022JJ00', 390.00, 375.50, 0.00),
    ('E001', 'S002', '2023JJ00', 370.00, 355.00, 0.00),
    ('E004', 'S006', '2018JJ00', 0.00, 0.00, 95.00),
    ('E004', 'S006', '2019JJ00', 0.00, 0.00, 125.00),
    ('E004', 'S006', '2020JJ00', 0.00, 0.00, 160.00),
    ('E004', 'S006', '2021JJ00', 0.00, 0.00, 190.00),
    ('E004', 'S006', '2022JJ00', 0.00, 0.00, 230.00),
    ('E004', 'S006', '2023JJ00', 0.00, 0.00, 275.00)
) AS vals(carrier_code, sector_code, period_code, supply_tj, consumption_tj, transformation_tj)
JOIN dim_energy_carrier c ON c.carrier_code = vals.carrier_code
JOIN dim_sector s ON s.sector_code = vals.sector_code
JOIN dim_time t ON t.period_code = vals.period_code;

-- ============================================================
-- ANALYTICAL QUERIES: Slice-and-dice demonstration
-- Business scenario: Dutch household energy costs over time
-- ============================================================

-- Query 1: Year-over-year household consumption by energy carrier
-- Slice by: sector (Households only), all carriers, all years
SELECT
    t.year,
    c.carrier_name,
    c.carrier_category,
    f.consumption_tj,
    LAG(f.consumption_tj) OVER (
        PARTITION BY c.carrier_key ORDER BY t.year
    ) AS prev_year_tj,
    ROUND(
        (f.consumption_tj - LAG(f.consumption_tj) OVER (
            PARTITION BY c.carrier_key ORDER BY t.year
        )) / NULLIF(LAG(f.consumption_tj) OVER (
            PARTITION BY c.carrier_key ORDER BY t.year
        ), 0) * 100, 1
    ) AS yoy_change_pct
FROM fact_energy_consumption f
JOIN dim_energy_carrier c ON f.carrier_key = c.carrier_key
JOIN dim_sector s ON f.sector_key = s.sector_key
JOIN dim_time t ON f.time_key = t.time_key
WHERE s.sector_name = 'Households'
    AND t.is_annual = TRUE
ORDER BY c.carrier_name, t.year;

-- Query 2: Renewable vs fossil share in household consumption
-- Dice by: renewable flag, aggregated across carriers per year
SELECT
    t.year,
    c.is_renewable,
    CASE WHEN c.is_renewable THEN 'Renewable' ELSE 'Non-renewable' END AS energy_type,
    SUM(f.consumption_tj) AS total_consumption_tj,
    ROUND(
        SUM(f.consumption_tj) / SUM(SUM(f.consumption_tj)) OVER (PARTITION BY t.year) * 100, 1
    ) AS share_pct
FROM fact_energy_consumption f
JOIN dim_energy_carrier c ON f.carrier_key = c.carrier_key
JOIN dim_sector s ON f.sector_key = s.sector_key
JOIN dim_time t ON f.time_key = t.time_key
WHERE s.sector_name = 'Households'
    AND t.is_annual = TRUE
GROUP BY t.year, c.is_renewable
ORDER BY t.year, c.is_renewable;

-- Query 3: Cross-sector comparison with running totals
-- Drill across: all sectors, natural gas only, cumulative decline
WITH sector_yearly AS (
    SELECT
        s.sector_name,
        t.year,
        SUM(f.consumption_tj) AS yearly_consumption_tj
    FROM fact_energy_consumption f
    JOIN dim_energy_carrier c ON f.carrier_key = c.carrier_key
    JOIN dim_sector s ON f.sector_key = s.sector_key
    JOIN dim_time t ON f.time_key = t.time_key
    WHERE c.carrier_name = 'Natural gas'
        AND t.is_annual = TRUE
    GROUP BY s.sector_name, t.year
)
SELECT
    sector_name,
    year,
    yearly_consumption_tj,
    SUM(yearly_consumption_tj) OVER (
        PARTITION BY sector_name ORDER BY year
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS cumulative_tj,
    RANK() OVER (PARTITION BY year ORDER BY yearly_consumption_tj DESC) AS sector_rank
FROM sector_yearly
ORDER BY year, sector_rank;

You now have a working dimensional model backed by real-world energy transition patterns. Next up, you will use dbt to build this same architecture as version-controlled, testable transformation models.

Build Your dbt Project with Medallion Architecture

Your dimensional model is built and loaded with CBS energy data. Now it is time to layer professional-grade transformation tooling on top of it.

In this step, you will initialize a dbt project connected to your PostgreSQL database and build a full medallion architecture with staging, intermediate, and mart models. You will also add automated tests and generate documentation to achieve production-quality standards.

In this step, get ready to:

• Initialize a dbt project and connect it to PostgreSQL.
• Build staging, intermediate, and mart models following the medallion architecture.
• Add schema tests, a custom generic test, and generate full documentation.

Initialize your dbt project with PostgreSQL

dbt needs two things to work: a project folder with your transformation logic, and a connection profile that tells it how to reach your database. The dbt init command scaffolds both.

• Activate your Python virtual environment and navigate into your project repository by running:

source dbt-env/bin/activate
cd ~/Desktop/project-repo

• The existing dbt/ folder is currently empty. Remove it so dbt init can create a fresh project in its place by running:

rmdir dbt

• Initialize your dbt project by running:

dbt init dbt

What will dbt init ask you?

The command prompts you to select a database adapter. Choose postgres (it may show as option 1). dbt then creates a new folder called dbt/ with sample files and writes a connection profile to ~/.dbt/profiles.yml.

When prompted for connection details (host, port, user, password, dbname, schema, threads), enter the values for your local PostgreSQL. If unsure, use the defaults in the next code block.

• Open ~/.dbt/profiles.yml in your editor and verify it matches your PostgreSQL credentials. Update it to look like this:

dbt:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      user: postgres
      password: your-postgres-password-here
      port: 5432
      dbname: postgres
      schema: public
      threads: 4

What does each field do?

• type tells dbt which adapter to use (PostgreSQL).
• host and port point to your local PostgreSQL server.
• dbname is the database where your energy tables live.
• schema is the default schema dbt writes models into.
• threads controls how many models dbt builds in parallel.

• Move into your new dbt project and verify the connection by running:

cd dbt
dbt debug

You should see All checks passed! at the bottom of the output, confirming dbt can reach your PostgreSQL database.

Connection test failed?

• Check that PostgreSQL is running on your machine.
• Verify the dbname in profiles.yml matches the database where you created your energy tables.
• Confirm your user and password are correct for your local PostgreSQL instance.

Help me troubleshoot my dbt debug connection failure to PostgreSQL

• Now update dbt_project.yml to configure model paths for the medallion architecture. Replace its contents with:

name: 'dbt'
version: '1.0.0'
config-version: 2

profile: 'dbt'

# Paths to models, macros, tests
model-paths: ["models"]
macro-paths: ["macros"]
test-paths: ["tests"]
analysis-paths: ["analyses"]
seed-paths: ["seeds"]
snapshot-paths: ["snapshots"]
clean-targets:
  - "target"
  - "dbt_packages"

# Medallion architecture materialization settings
models:
  dbt:
    staging:
      +materialized: view
    intermediate:
      +materialized: view
    marts:
      +materialized: table

Why different materializations?

Staging and intermediate models are materialized as views because they are lightweight transformations that always reflect the latest source data. Mart models are materialized as tables for faster query performance since they serve as the final business-facing layer.

• Create the folder structure for the medallion architecture by running:

mkdir -p models/staging models/intermediate models/marts macros

• Confirm the folders exist by running:

ls models/

You should see intermediate, marts, and staging listed.

Build your medallion architecture models

The medallion architecture organizes data into three layers. The bronze layer (staging) pulls raw data from sources with minimal transformation. The silver layer (intermediate) joins and enriches data. The gold layer (marts) delivers business-ready datasets.

• First, define your sources so dbt knows where the raw data lives. Create a file called models/staging/sources.yml with this content:

version: 2

sources:
  - name: raw_energy
    description: Raw CBS energy balance data loaded into PostgreSQL
    schema: public
    tables:
      - name: fact_energy_consumption
        description: Fact table with CBS energy balance measurements
      - name: dim_energy_carrier
        description: Dimension table for energy carrier types
      - name: dim_sector
        description: Dimension table for consumption sectors
      - name: dim_time
        description: Dimension table for time periods

What is a source in dbt?

Sources define the raw tables that already exist in your database. By declaring them in YAML, dbt can track data lineage from source to mart.

You reference sources in SQL using the {{ source() }} function instead of hardcoding table names. This makes your project portable across environments.

• Create your first staging model. Save this as models/staging/stg_energy_balance.sql:

-- Staging model: pulls raw energy balance facts with minimal transformation
with source as (
    select * from {{ source('raw_energy', 'fact_energy_consumption') }}
),

renamed as (
    select
        id as energy_fact_id,
        energy_carrier_id,
        sector_id,
        time_id,
        value as energy_value_tj,
        unit,
        measurement_type
    from source
)

select * from renamed

• Create the energy carriers staging model. Save this as models/staging/stg_energy_carriers.sql:

-- Staging model: standardizes energy carrier dimension
with source as (
    select * from {{ source('raw_energy', 'dim_energy_carrier') }}
),

renamed as (
    select
        energy_carrier_id,
        carrier_name,
        carrier_category,
        is_renewable
    from source
)

select * from renamed

• Create the sectors staging model. Save this as models/staging/stg_sectors.sql:

-- Staging model: standardizes sector dimension
with source as (
    select * from {{ source('raw_energy', 'dim_sector') }}
),

renamed as (
    select
        sector_id,
        sector_name,
        sector_category
    from source
)

select * from renamed

What does the staging layer accomplish?

Each staging model does one job: pull from a single source table and rename columns into a consistent format. No complex joins or business logic happens here. This keeps the bronze layer simple and auditable.

• Now build the intermediate (silver) layer. Create models/intermediate/int_energy_by_carrier.sql:

-- Intermediate model: joins energy facts with carrier details
with energy as (
    select * from {{ ref('stg_energy_balance') }}
),

carriers as (
    select * from {{ ref('stg_energy_carriers') }}
),

joined as (
    select
        e.energy_fact_id,
        e.time_id,
        e.sector_id,
        e.energy_value_tj,
        e.measurement_type,
        c.carrier_name,
        c.carrier_category,
        c.is_renewable
    from energy e
    inner join carriers c
        on e.energy_carrier_id = c.energy_carrier_id
)

select * from joined

• Create the second intermediate model. Save this as models/intermediate/int_energy_by_sector.sql:

-- Intermediate model: joins energy facts with sector details
with energy as (
    select * from {{ ref('stg_energy_balance') }}
),

sectors as (
    select * from {{ ref('stg_sectors') }}
),

joined as (
    select
        e.energy_fact_id,
        e.time_id,
        e.energy_carrier_id,
        e.energy_value_tj,
        e.measurement_type,
        s.sector_name,
        s.sector_category
    from energy e
    inner join sectors s
        on e.sector_id = s.sector_id
)

select * from joined

What is the ref() function?

The {{ ref() }} function tells dbt to reference another model by name. dbt uses these references to build a dependency graph and automatically runs models in the correct order.

• Build the first mart (gold) model. Create models/marts/mart_energy_consumption.sql:

-- Mart model: comprehensive energy consumption view for analysts
with energy_carrier as (
    select * from {{ ref('int_energy_by_carrier') }}
),

sectors as (
    select * from {{ ref('stg_sectors') }}
),

time_dim as (
    select * from {{ source('raw_energy', 'dim_time') }}
),

final as (
    select
        ec.energy_fact_id,
        t.year,
        t.period_label,
        ec.carrier_name,
        ec.carrier_category,
        ec.is_renewable,
        s.sector_name,
        s.sector_category,
        ec.energy_value_tj,
        ec.measurement_type
    from energy_carrier ec
    inner join sectors s
        on ec.sector_id = s.sector_id
    inner join time_dim t
        on ec.time_id = t.time_id
)

select * from final

• Create the second mart model for trend analysis. Save as models/marts/mart_energy_trends.sql:

-- Mart model: energy trends over time for reporting
with consumption as (
    select * from {{ ref('mart_energy_consumption') }}
),

aggregated as (
    select
        year,
        carrier_category,
        is_renewable,
        sector_category,
        measurement_type,
        sum(energy_value_tj) as total_energy_tj,
        count(*) as record_count
    from consumption
    group by
        year,
        carrier_category,
        is_renewable,
        sector_category,
        measurement_type
)

select * from aggregated

• Save all files, then compile and execute your models by running:

dbt run

You should see each model listed with a green OK status, confirming that staging views and mart tables were created in PostgreSQL.

Seeing a database error?

• Check that your source table names in sources.yml match the exact table names in PostgreSQL (case-sensitive).
• Verify the column names in your staging models match what exists in the actual tables. Run \d fact_energy_consumption in psql to check.
• Ensure the schema in profiles.yml matches where your source tables live.

Help me debug my dbt run database error

Add schema tests, a custom test, and a date spine macro

Models without tests are just SQL files. The dbt testing framework lets you assert data quality rules that run automatically every time you build. You will also create a reusable macro for generating date sequences.

• Create the schema file with column descriptions and tests. Save this as models/schema.yml:

version: 2

models:
  - name: stg_energy_balance
    description: Staged energy balance facts from CBS raw data
    columns:
      - name: energy_fact_id
        description: Unique identifier for each energy measurement
        tests:
          - unique
          - not_null
      - name: energy_carrier_id
        description: Foreign key to energy carrier dimension
        tests:
          - not_null
          - relationships:
              to: ref('stg_energy_carriers')
              field: energy_carrier_id
      - name: sector_id
        description: Foreign key to sector dimension
        tests:
          - not_null
      - name: energy_value_tj
        description: Energy measurement value in terajoules
        tests:
          - not_null

  - name: stg_energy_carriers
    description: Staged energy carrier dimension from CBS data
    columns:
      - name: energy_carrier_id
        description: Unique identifier for each energy carrier
        tests:
          - unique
          - not_null
      - name: carrier_name
        description: Human-readable name of the energy carrier
        tests:
          - not_null
      - name: is_renewable
        description: Whether this carrier is a renewable energy source
        tests:
          - accepted_values:
              values: [true, false]

  - name: stg_sectors
    description: Staged sector dimension from CBS data
    columns:
      - name: sector_id
        description: Unique identifier for each sector
        tests:
          - unique
          - not_null
      - name: sector_name
        description: Name of the energy consumption sector
        tests:
          - not_null

  - name: int_energy_by_carrier
    description: Energy facts enriched with carrier details
    columns:
      - name: energy_fact_id
        description: Unique energy measurement identifier
        tests:
          - unique
          - not_null
      - name: carrier_name
        description: Name of the energy carrier
        tests:
          - not_null

  - name: int_energy_by_sector
    description: Energy facts enriched with sector details
    columns:
      - name: energy_fact_id
        description: Unique energy measurement identifier
        tests:
          - unique
          - not_null
      - name: sector_name
        description: Name of the consumption sector
        tests:
          - not_null

  - name: mart_energy_consumption
    description: Comprehensive energy consumption view combining all dimensions
    columns:
      - name: energy_fact_id
        description: Unique identifier for each energy record
        tests:
          - unique
          - not_null
      - name: year
        description: Year of the energy measurement
        tests:
          - not_null
      - name: carrier_name
        description: Name of the energy carrier
        tests:
          - not_null
      - name: sector_name
        description: Name of the consumption sector
        tests:
          - not_null
      - name: energy_value_tj
        description: Energy value in terajoules
        tests:
          - not_null
          - positive_value

  - name: mart_energy_trends
    description: Aggregated energy trends for reporting and dashboards
    columns:
      - name: year
        description: Year of the aggregated measurement
        tests:
          - not_null
      - name: total_energy_tj
        description: Sum of energy values in terajoules for the group
        tests:
          - not_null

What do these tests check?

• unique ensures no duplicate values exist in a column.
• not_null ensures every row has a value (no NULLs).
• accepted_values restricts a column to a defined set of allowed values.
• relationships verifies that every foreign key value exists in the referenced model.
• positive_value is a custom generic test you will create next.

• Create the folder for your custom generic test by running:

mkdir -p tests/generic

• Create the custom generic test that checks for positive values. Save this as tests/generic/test_positive_value.sql:

-- Custom generic test: ensures all values in a column are positive
{% test positive_value(model, column_name) %}

select
    {{ column_name }} as failing_value
from {{ model }}
where {{ column_name }} <= 0

{% endtest %}

How do custom generic tests work?

A custom generic test is a Jinja-templated SQL query that returns rows which fail the test. If the query returns zero rows, the test passes. If it returns any rows, those are the violations.

You can reference this test by name in your schema.yml just like the built-in tests.

• Now create the date spine macro. Save this as macros/date_spine.sql:

-- Macro: generates a sequence of dates between start and end
{% macro date_spine(start_date, end_date, datepart='day') %}

    with date_spine as (
        select
            generate_series(
                '{{ start_date }}'::date,
                '{{ end_date }}'::date,
                '1 {{ datepart }}'::interval
            )::date as date_day
    )

    select * from date_spine

{% endmacro %}

When would you use a date spine?

A date spine generates a complete sequence of dates even when your source data has gaps. This is essential for time-series analysis where you need to show zero values for missing periods instead of skipping them entirely.

• Run the full dbt build (models and tests together) by running:

dbt build

You should see all models compile and run successfully, followed by all tests passing. The final output shows Completed successfully with zero errors.

Tests failing?

• If unique tests fail, your source data may have duplicates. Check with a SELECT COUNT(*) vs SELECT COUNT(DISTINCT id) query in psql.
• If relationships tests fail, some foreign key values in the fact table do not exist in the dimension table. Check for orphaned records.
• If positive_value fails, your energy data contains zero or negative values. You may need to adjust the test to < 0 if zero is a valid measurement.

Help me fix failing dbt tests on my CBS energy data

• Generate the project documentation by running:

dbt docs generate

This creates catalog.json in the target/ directory, containing metadata for every model, column, and test in your project.

• Optionally, serve the docs locally to explore them in your browser by running:

dbt docs serve

This starts a local webserver on port 8080 where you can browse model lineage, column descriptions, and test results.

✔️ Awesome, I've got everything!

Great. Double check that dbt build runs with zero errors and dbt docs generate completes successfully.

ⓧ I'd like to double check the full code

Here is the complete dbt_project.yml:

name: 'dbt'
version: '1.0.0'
config-version: 2

profile: 'dbt'

# Paths to models, macros, tests
model-paths: ["models"]
macro-paths: ["macros"]
test-paths: ["tests"]
analysis-paths: ["analyses"]
seed-paths: ["seeds"]
snapshot-paths: ["snapshots"]
clean-targets:
  - "target"
  - "dbt_packages"

# Medallion architecture materialization settings
models:
  dbt:
    staging:
      +materialized: view
    intermediate:
      +materialized: view
    marts:
      +materialized: table

Here is the complete models/schema.yml:

version: 2

models:
  - name: stg_energy_balance
    description: Staged energy balance facts from CBS raw data
    columns:
      - name: energy_fact_id
        description: Unique identifier for each energy measurement
        tests:
          - unique
          - not_null
      - name: energy_carrier_id
        description: Foreign key to energy carrier dimension
        tests:
          - not_null
          - relationships:
              to: ref('stg_energy_carriers')
              field: energy_carrier_id
      - name: sector_id
        description: Foreign key to sector dimension
        tests:
          - not_null
      - name: energy_value_tj
        description: Energy measurement value in terajoules
        tests:
          - not_null

  - name: stg_energy_carriers
    description: Staged energy carrier dimension from CBS data
    columns:
      - name: energy_carrier_id
        description: Unique identifier for each energy carrier
        tests:
          - unique
          - not_null
      - name: carrier_name
        description: Human-readable name of the energy carrier
        tests:
          - not_null
      - name: is_renewable
        description: Whether this carrier is a renewable energy source
        tests:
          - accepted_values:
              values: [true, false]

  - name: stg_sectors
    description: Staged sector dimension from CBS data
    columns:
      - name: sector_id
        description: Unique identifier for each sector
        tests:
          - unique
          - not_null
      - name: sector_name
        description: Name of the energy consumption sector
        tests:
          - not_null

  - name: int_energy_by_carrier
    description: Energy facts enriched with carrier details
    columns:
      - name: energy_fact_id
        description: Unique energy measurement identifier
        tests:
          - unique
          - not_null
      - name: carrier_name
        description: Name of the energy carrier
        tests:
          - not_null

  - name: int_energy_by_sector
    description: Energy facts enriched with sector details
    columns:
      - name: energy_fact_id
        description: Unique energy measurement identifier
        tests:
          - unique
          - not_null
      - name: sector_name
        description: Name of the consumption sector
        tests:
          - not_null

  - name: mart_energy_consumption
    description: Comprehensive energy consumption view combining all dimensions
    columns:
      - name: energy_fact_id
        description: Unique identifier for each energy record
        tests:
          - unique
          - not_null
      - name: year
        description: Year of the energy measurement
        tests:
          - not_null
      - name: carrier_name
        description: Name of the energy carrier
        tests:
          - not_null
      - name: sector_name
        description: Name of the consumption sector
        tests:
          - not_null
      - name: energy_value_tj
        description: Energy value in terajoules
        tests:
          - not_null
          - positive_value

  - name: mart_energy_trends
    description: Aggregated energy trends for reporting and dashboards
    columns:
      - name: year
        description: Year of the aggregated measurement
        tests:
          - not_null
      - name: total_energy_tj
        description: Sum of energy values in terajoules for the group
        tests:
          - not_null

Here is the complete macros/date_spine.sql:

-- Macro: generates a sequence of dates between start and end
{% macro date_spine(start_date, end_date, datepart='day') %}

    with date_spine as (
        select
            generate_series(
                '{{ start_date }}'::date,
                '{{ end_date }}'::date,
                '1 {{ datepart }}'::interval
            )::date as date_day
    )

    select * from date_spine

{% endmacro %}

Here is the custom test tests/generic/test_positive_value.sql:

-- Custom generic test: ensures all values in a column are positive
{% test positive_value(model, column_name) %}

select
    {{ column_name }} as failing_value
from {{ model }}
where {{ column_name }} <= 0

{% endtest %}

Your dbt project now has a production-quality medallion architecture with automated data quality testing. Next up, you will extend this project with advanced dbt features like snapshots for SCD Type 2, incremental models, and custom macros for CBS-specific transformations.

Implement SCD Type 2, Macros, and Data Quality

Your dbt project now has a working medallion architecture with staging, intermediate, and mart models. But in production data engineering, data changes over time. Energy carrier classifications get reclassified. Sector definitions shift.

Without tracking these changes, you lose the ability to answer historical questions accurately. In this step, you will implement Slowly Changing Dimensions Type 2 (SCD Type 2) to preserve that history. You will also write reusable macros and convert a model to incremental processing for efficiency.

In this step, get ready to:

• Create snapshot models that track historical changes in dimension tables using SCD Type 2.
• Write custom macros for reusable transformation logic including CBS period code standardization.
• Convert a model to incremental with a merge strategy and add data quality tests.

Create SCD Type 2 snapshots for dimension tables

In a real data pipeline, dimension tables change over time. An energy carrier might be reclassified from "fossil" to "transitional," or a sector definition might be updated when CBS revises its methodology.

SCD Type 2 preserves every version of a record by adding validity timestamps. When a record changes, the old row gets an end date and a new row is inserted with the updated values.

What is SCD Type 2?

SCD Type 2 keeps a full history of changes. Each row has a dbt_valid_from and dbt_valid_to timestamp. The current row has dbt_valid_to set to NULL.

This lets you join a fact table to a dimension "as of" any point in time, answering questions like "what was the energy carrier classification in 2019?" even if it changed in 2021.

• Create a snapshots directory inside your project-repo/dbt/ folder by running:

mkdir -p project-repo/dbt/snapshots

• Create the first snapshot file for energy carrier classifications by running:

touch project-repo/dbt/snapshots/snap_energy_carrier.sql

• Open snap_energy_carrier.sql and add the following snapshot configuration:

snapshots:
  - name: snap_energy_carrier
    relation: source('cbs_energy', 'dim_energy_carrier')
    config:
      schema: snapshots
      unique_key: carrier_id
      strategy: check
      check_cols:
        - carrier_name
        - carrier_category
        - carrier_unit

What does this snapshot do?

The check strategy compares the values in check_cols each time dbt runs. If any of those columns change for a given carrier_id, dbt invalidates the old row and inserts a new one with the updated values.

The check strategy works well here because our CBS dimension tables do not have a reliable updated_at timestamp column. When no such column exists, dbt compares actual column values to detect changes.

• Create the second snapshot file for sector definitions by running:

touch project-repo/dbt/snapshots/snap_sector.sql

• Open snap_sector.sql and add the following snapshot configuration:

snapshots:
  - name: snap_sector
    relation: source('cbs_energy', 'dim_sector')
    config:
      schema: snapshots
      unique_key: sector_id
      strategy: check
      check_cols:
        - sector_name
        - sector_category

Why the check strategy instead of timestamp?

dbt offers two snapshot strategies. The timestamp strategy is recommended when your source has a reliable updated_at column. The check strategy compares actual column values and works when no reliable timestamp exists.

CBS data is published as point-in-time releases without per-row update timestamps. The check strategy detects any change to the dimension attributes regardless of when the source was last modified.

• Run your snapshots for the first time by running:

cd project-repo/dbt
dbt snapshot

• You should see output confirming both snapshots were created successfully.

Snapshot command failing?

Make sure your dbt_project.yml includes snapshot-paths: ['snapshots'] so dbt knows where to find your snapshot files. Also verify that your source definition in schema.yml includes the dim_energy_carrier and dim_sector tables.

Help me troubleshoot my dbt snapshot error. I have two snapshots using the check strategy on my CBS energy dimension tables.

Write custom macros for reusable transformation logic

CBS period codes follow a specific format like 2023JJ00 (annual), 2023KW01 (quarterly), or 2023MM01 (monthly). Standardizing these into proper date formats is logic you will reuse across many models.

Macros in dbt are reusable pieces of Jinja-templated SQL. They work like functions: define once, call from any model.

• Create the period standardization macro by running:

touch project-repo/dbt/macros/standardize_period.sql

• Open standardize_period.sql and add the following macro:

{%- macro standardize_cbs_period(period_column) -%}
{#
  Converts CBS period codes into proper date values.
  Handles:
    - Annual: 2023JJ00 -> 2023-01-01
    - Quarterly: 2023KW01 -> 2023-01-01, 2023KW02 -> 2023-04-01, etc.
    - Monthly: 2023MM01 -> 2023-01-01, 2023MM12 -> 2023-12-01
#}
case
  -- Annual periods (e.g., 2023JJ00)
  when {{ period_column }} ~ '^[0-9]{4}JJ00$' then
    make_date(
      cast(left({{ period_column }}, 4) as integer),
      1,
      1
    )
  -- Quarterly periods (e.g., 2023KW01)
  when {{ period_column }} ~ '^[0-9]{4}KW[0-9]{2}$' then
    make_date(
      cast(left({{ period_column }}, 4) as integer),
      (cast(right({{ period_column }}, 2) as integer) - 1) * 3 + 1,
      1
    )
  -- Monthly periods (e.g., 2023MM01)
  when {{ period_column }} ~ '^[0-9]{4}MM[0-9]{2}$' then
    make_date(
      cast(left({{ period_column }}, 4) as integer),
      cast(right({{ period_column }}, 2) as integer),
      1
    )
  else null
end
{%- endmacro -%}

What does this macro do?

• It accepts a column name as input and generates a SQL CASE statement that parses CBS period codes.
• Annual codes like 2023JJ00 become 2023-01-01.
• Quarterly codes map to the first day of each quarter.
• Monthly codes map to the first day of each month.
• Any unrecognized format returns NULL so your pipeline does not break on unexpected data.

• Now create a third macro for generating a clean column name from a CBS identifier. You already have date_spine.sql from before. Create a new macro by running:

touch project-repo/dbt/macros/clean_cbs_identifier.sql

• Open clean_cbs_identifier.sql and add the following:

{%- macro clean_cbs_identifier(column_name) -%}
{#
  Strips trailing whitespace and underscores from CBS identifiers.
  CBS OData often pads identifiers with trailing spaces.
#}
trim(both ' ' from {{ column_name }})
{%- endmacro -%}

Why build custom macros?

CBS data has consistent quirks: padded identifiers, non-standard period codes, and specific naming conventions. By abstracting these patterns into macros, you call {{ standardize_cbs_period('perioden') }} in any model instead of rewriting the same CASE statement everywhere.

You now have 3 macros total: date_spine.sql, standardize_period.sql, and clean_cbs_identifier.sql.

Convert a model to incremental and add data quality tests

As your energy dataset grows with new CBS data releases, rebuilding every model from scratch becomes slow. Incremental models process only new or changed data. The merge strategy specifically updates existing rows and inserts new ones.

• Choose one of your intermediate models to convert to incremental. Open the model file and update its config block to use the incremental materialization with a merge strategy. The config should look like this:

{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='energy_record_id'
) }}

select
    -- your existing model columns
    *
from {{ ref('stg_energy_balance') }}

{% if is_incremental() %}
  -- only process rows newer than the latest in our current table
  where loaded_at > (select max(loaded_at) from {{ this }})
{% endif %}

What does this incremental config do?

• materialized='incremental' tells dbt to only process new data on subsequent runs instead of rebuilding the entire table.
• incremental_strategy='merge' inserts new records and updates existing records that match the unique_key.
• is_incremental() returns true on subsequent runs. On the first run, dbt builds the full table. On later runs, only new rows are processed.
• Use dbt run --full-refresh to force a complete rebuild if you ever need to reset the incremental state.

Now add data quality tests to your schema.yml to catch real data issues.

• Open project-repo/dbt/models/schema.yml and add the following test configurations to your energy models:

models:
  - name: int_energy_consumption
    columns:
      - name: energy_value
        tests:
          - not_null:
              config:
                severity: warn
                warn_if: ">100"
      - name: period_code
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: "'1946JJ00'"
              max_value: "'2030JJ00'"
      - name: carrier_id
        tests:
          - not_null
          - relationships:
              to: ref('stg_dim_energy_carrier')
              field: carrier_id

What do these data quality tests catch?

• not_null on energy_value catches null energy measurements that would corrupt aggregations.
• The range test on period_code catches invalid period formats that fall outside CBS's data range.
• relationships on carrier_id catches referential integrity violations where a fact record references an energy carrier that does not exist in the dimension table.

• Run the full build to verify everything passes by running:

dbt build

• You should see all models, snapshots, and tests passing. The output will show your two snapshots, your incremental model, and the data quality tests all completing successfully.

Tests failing on null values?

If your not_null tests fail, this means your CBS data actually contains null energy values. This is expected for certain carrier/sector combinations. Use severity: warn to allow the build to pass while still flagging the issue, or add a where clause to exclude known-null combinations.

Help me debug my failing dbt data quality tests on CBS energy data.

Your dbt project now tracks dimension history, processes data incrementally, and validates data quality automatically. Next up, you will orchestrate all of this with Apache Airflow so it runs on a schedule without manual intervention.

Orchestrate Pipelines with Apache Airflow

Your dbt project transforms raw data into clean, tested models. But right now, you run each step manually. In production, data pipelines need to run automatically on a schedule, handle failures gracefully, and pass information between steps.

This is where Apache Airflow comes in. Airflow is a workflow orchestration platform that lets you define pipelines as code, schedule them with cron expressions, and monitor everything through a web UI. In this step, you will launch Airflow locally using Docker Compose and write a DAG that orchestrates your entire CBS energy pipeline end-to-end.

In this step, get ready to:

• Launch Apache Airflow locally using Docker Compose.
• Write your first DAG with PythonOperator, BashOperator, XComs, and retry logic.
• Build and test a complete daily pipeline that extracts CBS data, loads it into PostgreSQL, and triggers dbt transformations.

Launch Apache Airflow with Docker Compose

Airflow runs multiple services (scheduler, webserver, workers, database) that work together. Docker Compose lets you spin up all of these with a single command using a pre-built configuration file from the Airflow project.

• Navigate to the airflow directory in your project repository by running this command:

cd ~/Desktop/project-repo/airflow

• Fetch the official Airflow 3.2.2 Docker Compose file by running this command:

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/3.2.2/docker-compose.yaml'

What does this command do?

This downloads the official docker-compose.yaml file maintained by the Apache Airflow project. It defines all the services Airflow needs: a scheduler, a DAG processor, an API server (web UI), workers, a triggerer, a PostgreSQL database, and a Redis broker.

The -LfO flags tell curl to follow redirects, fail silently on HTTP errors, and save the file with its original name.

• Create the required directories and set your user ID by running these commands:

mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env

Why set AIRFLOW_UID?

On Linux, Airflow containers need to know your host user ID so that files created in the dags, logs, and plugins directories have the correct ownership. Without this, files get created as root and become hard to manage.

• Initialize the Airflow database by running:

docker compose up airflow-init

Wait until you see a message ending with airflow-init-1 exited with code 0. This means the database migrations completed and the default admin user was created.

• Start all Airflow services by running:

docker compose up -d

What does docker compose up -d do?

The -d flag runs the containers in detached mode so you get your terminal back. Airflow needs at least 4 GB of memory allocated to Docker to run properly.

• Verify all containers are healthy by running:

docker ps

You should see several containers with (healthy) in their status column, including the API server, scheduler, and worker.

• Open your browser and navigate to http://localhost:8080.
• Log in with username airflow and password airflow.

You should see the Airflow web UI with a list of example DAGs. Your orchestration platform is running.

Containers keep restarting?

If the webserver or scheduler keeps restarting, Docker likely does not have enough memory allocated. Airflow requires at least 4 GB. Check Docker settings and increase the memory limit.

If you see permission errors in the logs, verify your .env file contains AIRFLOW_UID set to the output of id -u.

help me troubleshoot Airflow Docker Compose containers that keep restarting

Write your first DAG with operators, XComs, and retries

A DAG (Directed Acyclic Graph) defines the order and dependencies of your pipeline tasks. Each task uses an operator that determines what kind of work it performs.

What is a DAG?

In Airflow, a DAG is a Python file that defines tasks and their execution order. "Directed" means tasks flow in one direction. "Acyclic" means there are no circular dependencies. Airflow reads DAG files from the dags/ folder and automatically detects them.

You will use three core concepts: PythonOperator (runs a Python function), BashOperator (runs a shell command), and XComs (passes small data between tasks).

• Create the DAG file by running:

touch ~/Desktop/project-repo/airflow/dags/cbs_energy_pipeline.py

• Open cbs_energy_pipeline.py in VS Code and add the imports and default arguments:

"""CBS Energy Pipeline DAG - Orchestrates extraction, loading, and transformation."""
import json
import logging
from datetime import datetime, timedelta

import cbsodata
import pendulum
import psycopg2

from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG

# Configure logging for pipeline visibility
logger = logging.getLogger(__name__)

# Default arguments applied to every task in the DAG
default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "execution_timeout": timedelta(minutes=30),
}

What does this code do?

The default_args dictionary sets behaviour for every task in the DAG. Each task automatically retries 3 times with 5 minutes between attempts if it fails. The execution_timeout kills any task that runs longer than 30 minutes.

The imports pull in Airflow's operator classes from airflow.providers.standard (the Airflow 3.x import path) and the CBS data library you installed earlier.

• Below the imports, add the extraction function that pulls data from the CBS API:

def extract_cbs_data(**kwargs):
    """Extract energy balance data from CBS OData API."""
    logger.info("Starting CBS data extraction for table 83140ENG")

    # Pull the full energy balance dataset from CBS
    data = cbsodata.get_data("83140ENG")
    logger.info(f"Extracted {len(data)} records from CBS")

    # Push record count to XCom for downstream tasks to reference
    kwargs["ti"].xcom_push(key="record_count", value=len(data))

    # Return the data as JSON string (stored in XCom automatically)
    return json.dumps(data[:100])  # Sample for XCom size limits

What are XComs?

XComs (cross-communications) let tasks pass small pieces of data to each other. When a function returns a value, Airflow automatically stores it as an XCom. You can also explicitly push values with ti.xcom_push() and pull them in downstream tasks with ti.xcom_pull().

XComs are designed for small metadata (record counts, file paths, status messages). Never pass large datasets through XComs.

• Add the load function that writes extracted data into PostgreSQL:

def load_to_postgres(**kwargs):
    """Load extracted CBS data into PostgreSQL bronze layer."""
    ti = kwargs["ti"]

    # Pull the record count from the extraction task via XCom
    record_count = ti.xcom_pull(task_ids="extract_cbs_data", key="record_count")
    logger.info(f"Loading {record_count} records into bronze layer")

    # Connect to PostgreSQL and insert the raw data
    conn = psycopg2.connect(
        host="host.docker.internal",
        database="cbs_energy",
        user="postgres",
        password="postgres",
        port=5432,
    )
    cursor = conn.cursor()

    # Create bronze table if it does not exist
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS bronze_energy_balance (
            id SERIAL PRIMARY KEY,
            raw_data JSONB,
            loaded_at TIMESTAMP DEFAULT NOW()
        )
    """)

    # Pull the sample data from XCom and insert
    raw_data = ti.xcom_pull(task_ids="extract_cbs_data")
    cursor.execute(
        "INSERT INTO bronze_energy_balance (raw_data) VALUES (%s)",
        (raw_data,)
    )

    conn.commit()
    cursor.close()
    conn.close()

    logger.info("Bronze layer load complete")
    return {"status": "success", "records_loaded": record_count}

Why host.docker.internal?

Since Airflow runs inside Docker containers, it cannot reach your host machine's PostgreSQL using localhost. The special hostname host.docker.internal resolves to the host machine from inside a container. This lets the Airflow worker connect to your local PostgreSQL instance.

• Add the DAG definition that wires everything together with scheduling and task dependencies:

# Define the DAG with daily schedule
with DAG(
    dag_id="cbs_energy_pipeline",
    default_args=default_args,
    description="Daily CBS energy data extraction, load, and dbt transformation",
    schedule="0 6 * * *",  # Run daily at 6:00 AM
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["cbs", "energy", "elt"],
) as dag:

    # Task 1: Extract data from CBS OData API
    extract_task = PythonOperator(
        task_id="extract_cbs_data",
        python_callable=extract_cbs_data,
    )

    # Task 2: Load raw data into PostgreSQL bronze layer
    load_task = PythonOperator(
        task_id="load_to_postgres",
        python_callable=load_to_postgres,
    )

    # Task 3: Run dbt transformations (bronze -> silver -> gold)
    dbt_run_task = BashOperator(
        task_id="run_dbt_transformations",
        bash_command="cd /opt/airflow/dbt && dbt build --profiles-dir .",
    )

    # Define task dependencies: extract -> load -> transform
    extract_task >> load_task >> dbt_run_task

What does this DAG definition do?

The schedule="0 6 * * *" is a cron expression that runs the pipeline daily at 6:00 AM UTC. The catchup=False parameter prevents Airflow from running backfill jobs for past dates when you first enable the DAG.

The >> operator defines task dependencies. Extract runs first, then load, then dbt transformations. If any task fails, Airflow retries it up to 3 times before marking the DAG run as failed.

The BashOperator runs dbt build which executes all models, tests, and snapshots in dependency order.

• Save cbs_energy_pipeline.py.

DAG not appearing in the Airflow UI?

Airflow scans the dags/ folder every 30 seconds. If your DAG does not appear, check for Python syntax errors by running python cbs_energy_pipeline.py locally. Import errors prevent the DAG from loading.

Also verify that the file is saved inside ~/Desktop/project-repo/airflow/dags/ since that is the directory mounted into the Airflow container.

help me debug why my Airflow DAG is not showing up in the web UI

Test your pipeline end-to-end

A DAG that parses without errors is a good start, but you need to verify the full pipeline runs successfully from extraction through transformation.

• Return to the Airflow UI at http://localhost:8080.
• Find cbs_energy_pipeline in the DAG list.
• Toggle the DAG to Active using the toggle switch on the left.
• Click the Trigger DAG button (play icon) to manually run the pipeline.

Watch the DAG run in the grid view. Each task shows a green box when it succeeds. You should see extract_cbs_data complete first, then load_to_postgres, then run_dbt_transformations.

• Click on any task to view its logs. Verify the extraction task shows a record count and the load task confirms data was inserted.

What if a task fails?

Click the failed task (shown in red) to see its logs. The default_args retry setting means Airflow automatically retries 3 times before marking it as failed. Common failure causes include network timeouts to the CBS API or incorrect PostgreSQL connection details.

You can check XCom values in the Airflow UI by navigating to a task instance and clicking the XCom tab. This shows what data was passed between tasks.

• Verify the complete pipeline ran successfully by checking that the extract_cbs_data task shows a green status in the grid view.

Extract task fails with a connection error?

The CBS API is publicly accessible without authentication, but the Airflow container needs internet access. Verify your Docker network settings allow outbound connections.

If the load task fails with a connection refused error, ensure your PostgreSQL is configured to accept connections from Docker containers. You may need to update pg_hba.conf to allow connections from the Docker subnet.

help me fix connection errors in my Airflow DAG tasks

✔️ Awesome, I've got everything!

Great. Double check you have saved cbs_energy_pipeline.py and that your DAG ran successfully in the Airflow UI.

ⓧ I'd like to double check the full code

Here is the complete cbs_energy_pipeline.py file:

"""CBS Energy Pipeline DAG - Orchestrates extraction, loading, and transformation."""
import json
import logging
from datetime import datetime, timedelta

import cbsodata
import pendulum
import psycopg2

from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG

# Configure logging for pipeline visibility
logger = logging.getLogger(__name__)

# Default arguments applied to every task in the DAG
default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "execution_timeout": timedelta(minutes=30),
}


def extract_cbs_data(**kwargs):
    """Extract energy balance data from CBS OData API."""
    logger.info("Starting CBS data extraction for table 83140ENG")

    # Pull the full energy balance dataset from CBS
    data = cbsodata.get_data("83140ENG")
    logger.info(f"Extracted {len(data)} records from CBS")

    # Push record count to XCom for downstream tasks to reference
    kwargs["ti"].xcom_push(key="record_count", value=len(data))

    # Return the data as JSON string (stored in XCom automatically)
    return json.dumps(data[:100])  # Sample for XCom size limits


def load_to_postgres(**kwargs):
    """Load extracted CBS data into PostgreSQL bronze layer."""
    ti = kwargs["ti"]

    # Pull the record count from the extraction task via XCom
    record_count = ti.xcom_pull(task_ids="extract_cbs_data", key="record_count")
    logger.info(f"Loading {record_count} records into bronze layer")

    # Connect to PostgreSQL and insert the raw data
    conn = psycopg2.connect(
        host="host.docker.internal",
        database="cbs_energy",
        user="postgres",
        password="postgres",
        port=5432,
    )
    cursor = conn.cursor()

    # Create bronze table if it does not exist
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS bronze_energy_balance (
            id SERIAL PRIMARY KEY,
            raw_data JSONB,
            loaded_at TIMESTAMP DEFAULT NOW()
        )
    """)

    # Pull the sample data from XCom and insert
    raw_data = ti.xcom_pull(task_ids="extract_cbs_data")
    cursor.execute(
        "INSERT INTO bronze_energy_balance (raw_data) VALUES (%s)",
        (raw_data,)
    )

    conn.commit()
    cursor.close()
    conn.close()

    logger.info("Bronze layer load complete")
    return {"status": "success", "records_loaded": record_count}


# Define the DAG with daily schedule
with DAG(
    dag_id="cbs_energy_pipeline",
    default_args=default_args,
    description="Daily CBS energy data extraction, load, and dbt transformation",
    schedule="0 6 * * *",  # Run daily at 6:00 AM
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["cbs", "energy", "elt"],
) as dag:

    # Task 1: Extract data from CBS OData API
    extract_task = PythonOperator(
        task_id="extract_cbs_data",
        python_callable=extract_cbs_data,
    )

    # Task 2: Load raw data into PostgreSQL bronze layer
    load_task = PythonOperator(
        task_id="load_to_postgres",
        python_callable=load_to_postgres,
    )

    # Task 3: Run dbt transformations (bronze -> silver -> gold)
    dbt_run_task = BashOperator(
        task_id="run_dbt_transformations",
        bash_command="cd /opt/airflow/dbt && dbt build --profiles-dir .",
    )

    # Define task dependencies: extract -> load -> transform
    extract_task >> load_task >> dbt_run_task

Your data pipeline is now fully orchestrated. Airflow handles scheduling, retries, and monitoring automatically. Next up, you will add production patterns like CI/CD, data freshness checks, and pipeline monitoring.

Implement CI/CD, Monitoring, and Production Patterns

Your data pipeline works end-to-end. Extraction, loading, and transformation all run on schedule through Airflow. But production pipelines need more than correct logic.

They need guardrails that catch data drift before it reaches downstream consumers, branching strategies that prevent untested code from hitting production, and idempotency guarantees that let you safely re-run any task without duplicating rows.

In this step, get ready to:

• Set up a Git branching strategy with pre-commit hooks for automated quality gates.
• Add data freshness monitoring and a post-load validation DAG with SLAs.
• Publish your project to GitHub with a production-quality README.

Set up Git branching and pre-commit hooks

A branching strategy keeps experimental work isolated from stable code. Pre-commit hooks act as automated gatekeepers that block broken code before it enters your repository.

• Switch back to your terminal in the project repository directory.
• Create your branching structure by running these commands:

# Create the develop branch for integration work
git checkout -b develop
git push -u origin develop

# Create a feature branch for this week's work
git checkout -b feature/week7-production-patterns

What is this branching strategy?

The main branch holds production-ready code only. The develop branch is where you merge completed features before promoting to main. Feature branches isolate individual tasks so broken work never touches develop.

This mirrors how Dutch companies like ABN AMRO and KPMG manage data pipeline code in their repositories.

• Install the pre-commit framework by running:

pip install pre-commit sqlfluff

• Create the pre-commit configuration file by running:

touch .pre-commit-config.yaml

• Open .pre-commit-config.yaml in VS Code and paste the following configuration:

repos:
  # SQL linting for dbt models
  - repo: https://github.com/sqlfluff/sqlfluff
    rev: 3.4.0
    hooks:
      - id: sqlfluff-lint
        additional_dependencies: ['dbt-postgres', 'sqlfluff-templater-dbt']

  # Run dbt compile to catch broken references
  - repo: local
    hooks:
      - id: dbt-compile
        name: dbt compile
        entry: bash -c 'cd dbt && dbt compile'
        language: system
        pass_filenames: false
        files: ^dbt/

What does this configuration do?

The sqlfluff-lint hook lints your SQL files against style rules every time you commit. It catches formatting issues and potential errors before they enter the repository.

The dbt-compile local hook runs dbt compile on any commit that touches the dbt/ folder. This verifies that all model references resolve correctly and catches broken ref() calls instantly.

• Activate the pre-commit hooks by running:

pre-commit install

• Verify the hooks work by running them against all files:

pre-commit run --all-files

You should see output showing each hook running. Some SQL linting warnings are expected at this stage.

Seeing dbt compile errors?

Make sure your virtual environment is activated before running pre-commit. The hooks need access to the dbt command in your PATH.

If you see database connection errors, ensure PostgreSQL is running. The compile step checks references but does not execute queries.

help me fix my pre-commit hooks for dbt

Add data freshness checks and a quality validation DAG

Freshness checks define how recent your source data should be. If the CBS extraction fails silently, your mart models would serve stale data without anyone noticing. A dedicated validation DAG catches these issues automatically.

• Open your dbt/models/schema.yml file and add a sources block with freshness configuration. Place this at the top of the file:

sources:
  - name: cbs_raw
    description: Raw CBS energy data loaded by Airflow extraction
    schema: public
    config:
      freshness:
        warn_after:
          count: 36
          period: hour
        error_after:
          count: 48
          period: hour
      loaded_at_field: _loaded_at

    tables:
      - name: energy_balance_raw
        description: Raw CBS energy balance data (table 83140ENG)

What do warn_after and error_after mean?

The warn_after threshold raises a warning if data is older than 36 hours. The error_after threshold fails the freshness check entirely if data is older than 48 hours.

The loaded_at_field tells dbt which column to check. Your extraction scripts should add a _loaded_at timestamp column when loading data into PostgreSQL.

• Test the freshness configuration by running:

cd dbt && dbt source freshness

You should see output showing the freshness status of your source tables. A warning or error is expected if you have not run the extraction recently.

• Now create the post-load data quality validation DAG. Create a new file by running:

touch ../airflow/dags/data_quality_validation.py

• Open airflow/dags/data_quality_validation.py and add the following DAG definition:

from datetime import datetime, timedelta
from airflow.sdk import DAG, task
from airflow.providers.postgres.hooks.postgres import PostgresHook
import logging

logger = logging.getLogger(__name__)

# Default arguments with retry logic
default_args = {
    'owner': 'data_engineering',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

# DAG definition with SLA
with DAG(
    dag_id='data_quality_validation',
    default_args=default_args,
    description='Post-load data quality checks for CBS energy pipeline',
    schedule='0 8 * * *',  # Run daily at 8 AM, after the main pipeline
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['data-quality', 'monitoring'],
) as dag:

    @task(sla=timedelta(minutes=30))
    def check_row_counts():
        """Verify tables have expected minimum row counts."""
        hook = PostgresHook(postgres_conn_id='postgres_default')
        checks = {
            'fact_energy_consumption': 1000,
            'dim_energy_carrier': 5,
            'dim_sector': 3,
            'dim_time': 10,
        }
        for table, min_rows in checks.items():
            count = hook.get_first(f"SELECT COUNT(*) FROM {table}")[0]
            if count < min_rows:
                raise ValueError(
                    f"Table {table} has {count} rows, expected >= {min_rows}"
                )
            logger.info(f"PASS: {table} has {count} rows (min: {min_rows})")

    @task(sla=timedelta(minutes=30))
    def check_null_values():
        """Check critical columns for unexpected NULL values."""
        hook = PostgresHook(postgres_conn_id='postgres_default')
        null_checks = [
            ('fact_energy_consumption', 'energy_carrier_key'),
            ('fact_energy_consumption', 'time_key'),
            ('fact_energy_consumption', 'sector_key'),
        ]
        for table, column in null_checks:
            null_count = hook.get_first(
                f"SELECT COUNT(*) FROM {table} WHERE {column} IS NULL"
            )[0]
            if null_count > 0:
                raise ValueError(
                    f"Found {null_count} NULL values in {table}.{column}"
                )
            logger.info(f"PASS: No NULLs in {table}.{column}")

    @task(sla=timedelta(minutes=30))
    def check_referential_integrity():
        """Verify foreign key relationships are intact."""
        hook = PostgresHook(postgres_conn_id='postgres_default')
        # Check that all fact table keys exist in dimension tables
        orphan_count = hook.get_first("""
            SELECT COUNT(*)
            FROM fact_energy_consumption f
            LEFT JOIN dim_energy_carrier d ON f.energy_carrier_key = d.energy_carrier_key
            WHERE d.energy_carrier_key IS NULL
        """)[0]
        if orphan_count > 0:
            raise ValueError(
                f"Found {orphan_count} orphan records in fact_energy_consumption"
            )
        logger.info("PASS: Referential integrity check passed")

    # Define task dependencies
    check_row_counts() >> check_null_values() >> check_referential_integrity()

What are SLAs in Airflow?

The sla=timedelta(minutes=30) parameter on each task means Airflow will flag a missed SLA if any task takes longer than 30 minutes. This is how production teams set performance contracts on their pipelines.

The three checks cover the most common data quality failures: missing data (row counts), corrupt data (NULL foreign keys), and broken relationships (referential integrity).

Why idempotency matters

Your pipeline is already idempotent because of two design decisions you made earlier. Your incremental dbt models use a merge strategy that updates existing rows instead of inserting duplicates. Your extraction scripts use a DELETE + INSERT pattern (or TRUNCATE + LOAD) for the bronze layer, so re-running never creates duplicate data.

This means you can safely trigger any DAG run multiple times. Failed runs can be retried without manual cleanup.

• Save data_quality_validation.py.
• Return to the Airflow UI at localhost:8080 and refresh the page. You should see the data_quality_validation DAG appear in the list.

DAG not appearing in Airflow?

Make sure the file is saved in airflow/dags/ and has no Python syntax errors. Check the Airflow scheduler logs for import errors. A common cause is a missing import statement or an indentation mistake.

help me debug why my Airflow DAG is not showing up

Publish to GitHub with a production-quality README

A well-documented GitHub repository is your portfolio piece. Dutch hiring managers at companies like ilionx and EDSN review candidate repositories before interviews. Your README needs to communicate architecture decisions, not just setup steps.

• Open README.md in your project root and replace its contents with a structured README:

# Dutch Energy Data Platform

A production-grade ELT pipeline that extracts real energy consumption data from
the CBS (Statistics Netherlands) Open Data API, transforms it through a medallion
architecture using dbt, and orchestrates everything with Apache Airflow.

## Architecture

```
CBS OData API --> [Airflow Extraction] --> PostgreSQL (Bronze)
                                              |
                                     [dbt Transformations]
                                              |
                              Staging (Bronze) --> Intermediate (Silver) --> Marts (Gold)
                                              |
                                     [Data Quality Checks]
                                              |
                              Airflow Validation DAG + dbt Tests
```

## Tech Stack

- **Orchestration**: Apache Airflow 3.2.2 (Docker Compose)
- **Transformation**: dbt Core with dbt-postgres adapter
- **Database**: PostgreSQL
- **Data Source**: CBS Open Data (table 83140ENG - Energy Balance)
- **Quality**: dbt tests, source freshness checks, Airflow SLAs
- **CI/CD**: pre-commit hooks (sqlfluff linting + dbt compile)

## Project Structure

```
├── airflow/
│   ├── dags/
│   │   ├── cbs_energy_pipeline.py      # Main ELT orchestration
│   │   └── data_quality_validation.py  # Post-load quality checks
│   └── docker-compose.yaml
├── dbt/
│   ├── models/
│   │   ├── staging/        # Bronze layer (3 models)
│   │   ├── intermediate/   # Silver layer (2+ models, incremental)
│   │   └── marts/          # Gold layer (2+ models)
│   ├── snapshots/          # SCD Type 2 tracking
│   ├── macros/             # Reusable transformation logic
│   └── dbt_project.yml
├── extraction/             # CBS API extraction scripts
├── sql/                    # Analytical SQL exercises
├── docs/                   # Architecture documentation
└── .pre-commit-config.yaml
```

## Design Decisions

### Medallion Architecture
Bronze (raw) -> Silver (cleaned, validated) -> Gold (business-ready).
This separation allows reprocessing at any layer without affecting downstream
consumers.

### SCD Type 2 Snapshots
Energy carrier classifications and sector definitions change over time.
dbt snapshots track these changes with valid_from/valid_to columns.

### Idempotent Pipeline
Every task is safe to re-run. Bronze uses TRUNCATE+LOAD, incremental
models use merge strategies, and quality checks are stateless.

## Setup Instructions

1. Clone this repository
2. Create a Python virtual environment: `python3 -m venv dbt-env`
3. Activate it: `source dbt-env/bin/activate`
4. Install dependencies: `pip install dbt-postgres duckdb cbsodata pre-commit`
5. Start Airflow: `cd airflow && docker compose up -d`
6. Install pre-commit hooks: `pre-commit install`
7. Configure dbt: update `dbt/profiles.yml` with your PostgreSQL credentials
8. Run the pipeline: trigger the `cbs_energy_pipeline` DAG in Airflow UI

## Data Quality

- 3+ automated checks run post-load (row counts, NULL detection, referential integrity)
- dbt source freshness alerts when data is older than 36 hours
- Pipeline SLAs flag tasks that exceed 30 minutes
- All models have schema tests (unique, not_null, accepted_values, relationships)

Why this README structure matters

Dutch data engineering interviews often start with a repository review. Hiring managers look for three signals: architectural thinking (the diagram), trade-off awareness (design decisions section), and operational maturity (quality and monitoring sections).

A README that only lists setup steps tells the reviewer you followed a tutorial. A README with architecture and design decisions tells them you think like an engineer.

• Save README.md.
• Commit your week 7 work and push to GitHub by running:

git add -A
git commit -m "feat: add CI/CD, monitoring, and production patterns"
git push -u origin feature/week7-production-patterns

• Merge your feature branch into develop:

git checkout develop
git merge feature/week7-production-patterns
git push origin develop

• If you have not already created a remote repository on GitHub, do so now and push your develop branch. Your repository should be publicly visible.

You should see your complete project structure on GitHub with the README rendered on the repository landing page.

Push rejected or authentication issues?

If git push fails with an authentication error, you may need to set up a personal access token or SSH key with GitHub. Check that your remote URL is correct with git remote -v.

help me push my project to GitHub

Your pipeline now has proper version control, automated quality gates, and comprehensive monitoring. Next up, you will bring everything together into a polished capstone deliverable.

Assemble the Dutch Energy Data Platform Capstone

Over the past seven weeks, you built advanced SQL queries, dimensional models, a full dbt project with medallion architecture, and orchestrated everything with Apache Airflow.

Now it is time to assemble these pieces into a single, polished portfolio project. In this step, you will refactor your work into a cohesive data platform, generate a data catalog with lineage visualization, write professional documentation, and review the final result against real Dutch job postings.

In this step, get ready to:

• Refactor your project into a cohesive, production-quality data platform with organized extraction scripts.
• Generate a data catalog with lineage visualization and write a data dictionary.
• Polish your GitHub repository into a portfolio piece and review it against Dutch job postings.

Refactor into a production-quality data platform

A capstone project is more than the sum of its weekly exercises. Employers reviewing your GitHub want to see a coherent system where data flows cleanly from extraction to consumption.

Your first task is organizing the CBS extraction logic into a dedicated script and confirming the full pipeline runs end-to-end.

• In your terminal, navigate to your project repository by running:

cd ~/Desktop/project-repo

• Create the extraction directory and extraction script file by running:

mkdir -p extraction
touch extraction/cbs_extract.py

• Open extraction/cbs_extract.py in VS Code and add the following extraction script:

"""CBS OData API extraction script for the Dutch Energy Data Platform.

Extracts energy balance data from Statistics Netherlands (CBS) and
loads it into the PostgreSQL bronze layer.
"""
import cbsodata
import pandas as pd
from sqlalchemy import create_engine
import logging
import sys

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# CBS dataset identifiers
ENERGY_BALANCE_TABLE = '83140ENG'

# PostgreSQL connection
DB_CONNECTION = 'postgresql://localhost:5432/energy_platform'

def extract_cbs_data(table_id: str) -> pd.DataFrame:
    """Extract data from CBS OData API."""
    logger.info(f"Extracting data from CBS table: {table_id}")
    data = cbsodata.get_data(table_id)
    df = pd.DataFrame(data)
    logger.info(f"Extracted {len(df)} rows from {table_id}")
    return df

def extract_cbs_metadata(table_id: str) -> dict:
    """Extract metadata (column definitions) from CBS."""
    logger.info(f"Extracting metadata for table: {table_id}")
    meta = cbsodata.get_meta(table_id, 'DataProperties')
    return meta

def load_to_bronze(df: pd.DataFrame, table_name: str, engine) -> None:
    """Load extracted data into PostgreSQL bronze layer."""
    logger.info(f"Loading {len(df)} rows into bronze.{table_name}")
    df.to_sql(
        name=table_name,
        con=engine,
        schema='bronze',
        if_exists='replace',
        index=False
    )
    logger.info(f"Successfully loaded bronze.{table_name}")

def main():
    """Main extraction pipeline."""
    logger.info("Starting CBS energy data extraction")

    # Create database connection
    engine = create_engine(DB_CONNECTION)

    # Extract energy balance data
    energy_df = extract_cbs_data(ENERGY_BALANCE_TABLE)

    # Load into bronze layer
    load_to_bronze(energy_df, 'raw_energy_balance', engine)

    # Extract and store metadata for documentation
    metadata = extract_cbs_metadata(ENERGY_BALANCE_TABLE)
    logger.info(f"Retrieved {len(metadata)} data properties")

    logger.info("CBS extraction complete")

if __name__ == '__main__':
    main()

What does this script do?

This script consolidates your CBS data extraction into a single, reusable module. It uses the cbsodata package to pull data from table 83140ENG (the English-language energy balance dataset).

The script separates concerns into three functions: extract, load to bronze, and metadata retrieval. This is the same extraction logic your Airflow DAG calls, now organized as an importable module.

• Save extraction/cbs_extract.py.
• Run the full dbt pipeline to confirm all layers still build successfully:

cd dbt
dbt build

• You should see all models, snapshots, and tests pass. Confirm the output shows 10+ models across staging, intermediate, and marts layers.

Seeing model failures?

If any models fail, check that your PostgreSQL database is running and that the bronze layer tables contain data. Run dbt debug to verify your connection profile is correct.

help me debug dbt build failures in my capstone project

Generate a data catalog and write documentation

A data catalog makes your project self-documenting. Recruiters and hiring managers can browse your models, see how data flows through the pipeline, and understand your design decisions without reading every SQL file.

• From inside the dbt directory, regenerate the data catalog by running:

dbt docs generate

• Serve the documentation locally to explore lineage by running:

dbt docs serve --port 8001

Why port 8001?

Your Airflow instance is already running on port 8080. The --port flag tells dbt to serve the documentation site on a different port so both services can run simultaneously.

• Your browser should open automatically. Click the lineage graph icon (bottom-right corner) to see your full DAG visualization showing how data flows from sources through staging, intermediate, and mart layers.
• Press Ctrl+C in the terminal to stop the docs server once you have explored the lineage.

Next, create a data dictionary that documents every model and its key fields. This is what analysts and stakeholders reference when using your gold layer tables.

• Return to the project root and create the docs directory:

cd ~/Desktop/project-repo
mkdir -p docs

• Create docs/data_dictionary.md and add the following structure:

# Data Dictionary - Dutch Energy Data Platform

## Source: CBS Open Data (Statistics Netherlands)

- **Table**: 83140ENG (Energy balance sheet; supply, transformation and consumption)
- **Granularity**: Energy carrier × sector × time period
- **Update frequency**: Annual
- **API**: https://datasets.cbs.nl/odata/v1/CBS/83140ENG

## Bronze Layer (Staging Models)

### stg_energy_balance
- Raw energy balance data loaded from CBS API
- No transformations applied
- Columns preserve original CBS naming

### stg_energy_carriers
- Energy carrier dimension extracted from CBS metadata
- Maps carrier codes to human-readable names (e.g., coal, natural gas, renewable)

### stg_sectors
- Sector dimension from CBS metadata
- Maps sector codes to readable names (e.g., households, industry, transport)

## Silver Layer (Intermediate Models)

### int_energy_balance_cleaned
- Cleaned and standardized energy values
- CBS period codes converted to proper date format
- Null and placeholder values handled
- Incremental model: processes only new periods on each run

### int_energy_carriers_enriched
- Carriers classified into categories (fossil, renewable, nuclear)
- Used for energy transition analysis in gold layer

## Gold Layer (Mart Models)

### mart_energy_consumption
- Final fact table: energy consumption by carrier, sector, and time period
- Grain: one row per carrier × sector × year
- Measures: total_supply_pj, total_consumption_pj, transformation_losses_pj

### mart_energy_transition
- Aggregated view showing renewable vs fossil share over time
- Pre-calculated percentages for dashboard consumption

## Snapshots (SCD Type 2)

### snapshot_energy_carriers
- Tracks historical changes in energy carrier classifications
- Valid_from / valid_to date columns for point-in-time queries

### snapshot_sectors
- Tracks sector definition changes over time
- Enables historical reporting with correct sector mappings

## Key Metrics

| Metric | Definition | Unit |
|--------|-----------|------|
| total_supply_pj | Total energy supply | Petajoules (PJ) |
| total_consumption_pj | Final energy consumption | Petajoules (PJ) |
| renewable_share_pct | Renewable energy as % of total | Percentage |
| yoy_change_pct | Year-over-year consumption change | Percentage |

What makes a good data dictionary?

A data dictionary should answer three questions for every model: what data does it contain, what is the grain (one row per what), and what are the key measures or attributes. This is what data analysts need to write correct queries against your gold layer.

• Now create docs/architecture.md to document your technical decisions:

# Architecture Documentation - Dutch Energy Data Platform

## Overview

This platform extracts energy balance data from the CBS (Statistics Netherlands)
Open Data API, transforms it through a medallion architecture, and serves
analytical datasets for energy transition analysis.

## Architecture Decisions

### Why Medallion Architecture (Bronze/Silver/Gold)?

The medallion pattern separates concerns clearly:
- **Bronze**: Raw data preservation (audit trail, replayability)
- **Silver**: Cleaned, validated, business-logic-free data
- **Gold**: Business-specific aggregations ready for consumption

This matches the pattern used at companies like ABN AMRO and Eneco, where
data teams operate on shared silver layers while building domain-specific
gold layers.

### Why dbt over Custom Python Transformations?

dbt provides:
- Declarative SQL transformations (accessible to analysts)
- Built-in testing framework for data quality
- Automatic documentation and lineage
- Incremental processing for efficiency

The Netherlands market strongly favors dbt (required by EDSN, Eneco, KPMG).

### Why Apache Airflow for Orchestration?

Airflow is the dominant orchestrator in Dutch data engineering:
- Industry standard at ABN AMRO, KPMG, Eneco
- DAG-based scheduling matches the ELT dependency graph
- Rich operator ecosystem for diverse data sources
- Production-proven at enterprise scale

Dagster and Prefect are alternatives but have less market presence in NL.

### Why SCD Type 2 for Dimension Tracking?

Energy carrier and sector classifications change over time as the
Netherlands energy transition progresses. SCD Type 2 preserves history
so analysts can answer questions like "What was the renewable share
under the pre-2020 classification?"

## Data Flow

```
CBS OData API --> cbs_extract.py --> PostgreSQL (bronze schema)
    --> dbt staging models (bronze layer)
    --> dbt intermediate models (silver layer)
    --> dbt mart models (gold layer)
```

Orchestrated by Airflow DAG running on daily schedule.

## Trade-offs and Limitations

1. **Local PostgreSQL vs cloud warehouse**: Chose local for $0 cost.
   The patterns transfer directly to Snowflake/Databricks.
2. **Daily schedule vs real-time**: CBS updates annually, so daily is
   more than sufficient. The pattern supports higher frequency.
3. **Single-node Airflow vs distributed**: Local Docker Compose is
   suitable for this data volume. Production would use Kubernetes.

## Technology Stack

| Component | Tool | Version |
|-----------|------|--------|
| Warehouse | PostgreSQL | 15+ |
| Transformation | dbt Core | 1.11 |
| Orchestration | Apache Airflow | 3.2.2 |
| Extraction | Python + cbsodata | 3.10+ |
| Version Control | Git + GitHub | - |
| Containers | Docker Compose | 2.14.0+ |

Why document trade-offs?

Hiring managers look for engineers who can explain WHY they chose a specific approach, not just that they used a tool. Documenting trade-offs shows you understand the decision space and can evaluate alternatives.

• Save both docs/data_dictionary.md and docs/architecture.md.
• Confirm the files exist by running:

ls docs/

• You should see architecture.md and data_dictionary.md in the output.

Polish your README and review against Dutch job postings

Your README is the first thing a recruiter sees. It needs to immediately communicate what you built, what technologies you used, and how to run it.

• Open README.md in VS Code and update it to include these sections:

# Dutch Energy Data Platform

A production-quality ELT pipeline that extracts energy balance data from
Statistics Netherlands (CBS), transforms it through a medallion architecture
using dbt, and orchestrates everything with Apache Airflow.

## Architecture

```
CBS OData API --> Python Extraction --> PostgreSQL
    --> dbt (staging → intermediate → marts)
    --> Airflow Orchestration
    --> Data Catalog + Lineage
```

## Tech Stack

- **Extraction**: Python, cbsodata, SQLAlchemy
- **Warehouse**: PostgreSQL
- **Transformation**: dbt Core 1.11 (medallion architecture)
- **Orchestration**: Apache Airflow 3.2.2 (Docker Compose)
- **Data Modeling**: Star schema, SCD Type 2 snapshots
- **Quality**: dbt tests, Airflow data quality DAG, freshness checks
- **CI/CD**: Git branching, pre-commit hooks (dbt compile + lint)

## Key Features

- **10+ dbt models** across bronze/silver/gold layers
- **SCD Type 2 tracking** for energy carrier and sector changes
- **Incremental processing** for efficient data loads
- **Custom macros** for CBS period code parsing
- **Data catalog** with full lineage visualization (dbt docs)
- **Automated data quality** checks with Airflow DAG
- **Idempotent pipeline** safe to re-run without duplicates

## Project Structure

```
project-repo/
├── airflow/
│   ├── dags/
│   │   ├── cbs_energy_pipeline.py
│   │   └── data_quality_validation.py
│   └── docker-compose.yaml
├── dbt/
│   ├── models/
│   │   ├── staging/          # Bronze layer (3+ models)
│   │   ├── intermediate/     # Silver layer (2+ models)
│   │   └── marts/            # Gold layer (2+ models)
│   ├── snapshots/            # SCD Type 2 (2 models)
│   ├── macros/               # Reusable transformation logic
│   └── dbt_project.yml
├── extraction/
│   └── cbs_extract.py
├── docs/
│   ├── architecture.md
│   └── data_dictionary.md
├── sql/
│   ├── week1_window_functions.sql
│   ├── week2_advanced_patterns.sql
│   └── week3_dimensional_model.sql
└── README.md
```

## Quick Start

1. Clone this repository
2. Create a Python virtual environment: `python3 -m venv dbt-env`
3. Activate: `source dbt-env/bin/activate`
4. Install dependencies: `pip install dbt-postgres cbsodata`
5. Start PostgreSQL and load bronze data: `python extraction/cbs_extract.py`
6. Run dbt: `cd dbt && dbt build`
7. Start Airflow: `cd airflow && docker compose up`
8. View data catalog: `cd dbt && dbt docs generate && dbt docs serve --port 8001`

## Skills Demonstrated

- Advanced SQL (window functions, CTEs, recursive queries, GROUPING SETS)
- Dimensional data modeling (star schema, SCD Type 2)
- dbt (models, tests, macros, incremental, snapshots, documentation)
- Apache Airflow (DAGs, operators, scheduling, error handling)
- Data quality engineering (automated testing, freshness monitoring)
- Production patterns (CI/CD, idempotency, monitoring)

What does this README do?

This README follows a structure that technical hiring managers expect: a one-line summary, architecture overview, technology list, features, project structure, and quick start. It makes your project immediately runnable by anyone reviewing it.

The "Skills Demonstrated" section maps directly to requirements in Dutch data engineering job postings at companies like ABN AMRO, Eneco, KPMG, and EDSN.

• Save README.md.
• Commit and push all changes by running:

git add .
git commit -m "Add capstone documentation: data dictionary, architecture, polished README"
git push origin main

• You should see the commit pushed to your GitHub repository successfully.

Finally, review your project against the skills that Dutch employers are actively hiring for.

Dutch job posting alignment check

Open 2-3 data engineering job postings from Dutch companies (search LinkedIn or Indeed for "Data Engineer Netherlands"). Compare your project against their requirements list. Your capstone should demonstrate:

• Python and advanced SQL (check your sql/ folder and extraction scripts).
• dbt with medallion architecture (check your models, tests, and macros).
• Airflow orchestration (check your DAGs with scheduling and error handling).
• Dimensional modeling with SCD Type 2 (check your snapshots and star schema).
• CI/CD and data quality (check your pre-commit hooks and quality DAG).

Missing something from job postings?

If you notice a specific tool or pattern in job postings that your project does not cover (like Snowflake or Azure), add a section to your architecture.md explaining how the patterns you built transfer to that tool. For example, your local PostgreSQL + dbt setup is identical to a Snowflake + dbt setup in production.

help me identify gaps between my capstone project and Dutch data engineering job requirements

Secret mission

Build a Streamlit Dashboard

You have a fully orchestrated data platform with gold-layer tables ready to query. Now turn those insights into something visual. Build an interactive Streamlit dashboard that tells the story of the Netherlands energy transition, complete with filters and a province-level map.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing cloud costs.

Resources you used:

• Airflow Docker containers and volumes (running at localhost:8080).
• PostgreSQL databases and tables (CBS neighborhood data, dimensional model tables, dbt models).
• Python virtual environment (with dbt-postgres, DuckDB, cbsodata, and Streamlit installed).
• Local project files (the project-repo/ folder on your Desktop).
• GitHub repository (published portfolio project).

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend your project soon.

• Your Airflow containers will continue running and consuming memory until you stop them manually.
• PostgreSQL continues to store your CBS data and dimensional models.
• Your Python virtual environment and project files remain on disk.
• Docker Engine remains installed on your machine.

Pause - I'll come back to this later

Shut down running processes to free up memory, but keep all files and data intact so you can resume later.

• Stop the Streamlit app by pressing Ctrl+C in the terminal where it is running.
• Stop the Airflow containers by navigating to your project-repo/airflow/ directory and running the command below.

docker compose down

• This stops and removes the containers but keeps your volumes (Airflow metadata) intact for next time.
• Deactivate your Python virtual environment by running:

deactivate

• To resume later, reactivate your virtual environment with source dbt-env/bin/activate and run docker compose up from the project-repo/airflow/ directory.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is irreversible.

• Stop the Streamlit app by pressing Ctrl+C in its terminal.
• Stop and remove the Airflow containers, networks, and volumes by navigating to project-repo/airflow/ and running:

docker compose down --volumes

• Drop the PostgreSQL databases you created for this project. Connect to the default postgres database and run:

psql -d postgres

Then inside the psql prompt, drop the project database:

DROP DATABASE IF EXISTS cbs_energy WITH (FORCE);

• Exit psql with \q.
• Remove your Python virtual environment:

rm -rf dbt-env

• Remove the entire project folder from your Desktop:

rm -rf ~/Desktop/project-repo

• Optionally delete the GitHub repository. Go to your repository on GitHub, click Settings, scroll to the bottom of the General page, and click Delete this repository in the Danger Zone section.

What stays on your machine

Docker Engine, PostgreSQL, and Git remain installed. These are general-purpose tools you will likely use in future projects. If you want to remove Docker entirely, follow the official uninstall guide for your Linux distribution at docs.docker.com.

Nice Work!

Nice Work!

You have built a complete data engineering skillset optimized for the Netherlands job market. Your portfolio project demonstrates exactly the tools and patterns that Dutch companies like ABN AMRO, Eneco, KPMG, and ilionx are actively hiring for.

You learned how to:

• Mastered advanced SQL (window functions, CTEs, recursive queries, performance optimization) and built dimensional data models with SCD Type 2 tracking in PostgreSQL.
• Created a full dbt project with a medallion architecture (bronze/silver/gold layers), custom macros, data quality tests, and generated documentation.
• Orchestrated end-to-end ELT pipelines with Apache Airflow using real Dutch government data from CBS Open Data, complete with CI/CD, monitoring, and idempotency patterns.
• Secret Mission: Built a Streamlit dashboard visualizing the Netherlands energy transition trends with interactive filters and a province-level map.

Ready to quiz yourself?