Geospatial ETL Pipeline with PostGIS

Build a geospatial ETL pipeline from OpenStreetMap data to a star schema warehouse.

Introduction

30 Second Summary

Every ride-share route, city zoning decision, and delivery estimate depends on spatial data that has been cleaned, structured, and made queryable. Behind these tools sit data pipelines that transform messy map data into organized dimensional warehouses ready for analytics.

In this project, you will build a geospatial ETL pipeline that extracts real OpenStreetMap data for a city neighborhood, transforms it through layered processing stages, loads it into a dimensional warehouse, and publishes interactive dashboards.

What You'll Build

A complete geospatial data platform that ingests raw OpenStreetMap data, processes it through dbt-style staging/intermediate/mart layers in Python, stores it in a PostgreSQL/PostGIS star schema warehouse, and exports analytics to Tableau Public dashboards.

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

  • A modular Python ETL pipeline with config-driven layered architecture that extracts, cleans, and loads geospatial data.
  • A PostgreSQL/PostGIS dimensional warehouse with a star schema containing fact and dimension tables for urban land-use features.
  • A Tableau Public dashboard with spatial visualizations and land-use analytics consuming exported GeoJSON and CSV files.
  • Secret Mission: Orchestrate the entire pipeline with an Apache Airflow DAG running in Docker Compose.

Are there any prerequisites?

You should have Python and Docker installed on your machine. Familiarity with SQL and basic data modeling concepts (tables, schemas, joins) will help you follow along smoothly.

Set Up Your Environment

Before you can build an ETL pipeline, you need the right tools in place. PostgreSQL with PostGIS gives you a spatial database that can store and query geographic data natively.

In this step, you will install PostgreSQL 17 with the PostGIS extension, set up a Python virtual environment with geospatial libraries, and scaffold the project folder structure that mirrors a professional dbt-style pipeline architecture.

In this step, get ready to:
  • Install PostgreSQL 17 and enable the PostGIS spatial extension.
  • Create a Python virtual environment and install the geospatial library stack.
  • Scaffold the project folder structure and configuration file.
Install PostgreSQL 17 and PostGIS

PostgreSQL is the database engine that will house your dimensional warehouse. PostGIS is an extension that adds spatial data types and functions so PostgreSQL can understand geometry and geography.

  • Check whether PostgreSQL is already installed on your machine by opening a terminal and running:
psql --version

What if psql is not in my PATH?

The EnterpriseDB installer does not always add psql to your system PATH by default. If you know PostgreSQL is installed but the command is not recognized, you can check the version later from pgAdmin instead. Pick the tab below that matches what you see.

✔️ I see version 17 or higher

PostgreSQL 17 is ready. Skip ahead to the PostGIS installation section below.

ⓧ I see an older version

You need PostgreSQL 17 for this project. Download the latest PostgreSQL 17 installer from the official site and install it alongside or in place of your current version.

  • Visit postgresql.org/download/windows and click Download the installer.
  • Select the latest PostgreSQL 17 version for Windows x86-64.
  • Run the downloaded installer.
  • Accept the default install directory and components.
  • When prompted for a password, choose a password for the postgres superuser and write it down.
  • Accept the default port 5432 and finish the installation.

When the installer finishes, StackBuilder will launch automatically. Continue to the PostGIS section below.

ⓧ Command not found

PostgreSQL is not installed yet. You will install it using the EnterpriseDB installer, which bundles the server, pgAdmin, and the StackBuilder package manager.

  • Visit postgresql.org/download/windows and click Download the installer.
  • Select the latest PostgreSQL 17 version for Windows x86-64.
  • Run the downloaded installer.
  • Accept the default install directory and components (PostgreSQL Server, pgAdmin 4, StackBuilder, Command Line Tools).
  • When prompted for a password, choose a password for the postgres superuser and write it down. You will need this later for your pipeline configuration.
  • Accept the default port 5432.
  • Accept the default locale and click Next until the installation begins.

When the installer finishes, it will ask if you want to launch StackBuilder. Leave the checkbox checked and click Finish.

Now install the PostGIS spatial extension using StackBuilder.

  • In StackBuilder, select your PostgreSQL 17 server from the dropdown and click Next.
  • Expand the Spatial Extensions category.
  • Check the box for PostGIS Bundle and click Next.
  • Accept the license agreement and follow the prompts to complete the PostGIS installation.
  • When asked whether to register GDAL_DATA environment variables, select Yes.

StackBuilder not launching?

If StackBuilder did not open automatically, press the Windows key, type StackBuilder, and press Enter to open it manually.

Now verify that PostGIS is working by running a query in pgAdmin.

  • Press the Windows key, type pgAdmin, and press Enter to open it.
  • Enter the postgres superuser password you set during installation when prompted.
  • In the left sidebar, expand Servers then expand your PostgreSQL 17 server.
  • Right-click Databases and select Create then Database.
  • Enter geo_warehouse as the database name and click Save.
  • Click on the geo_warehouse database in the sidebar, then click Tools in the top menu and select Query Tool.
  • Run the following SQL to enable PostGIS and verify the installation:
CREATE EXTENSION postgis;
SELECT postgis_full_version();

What does this SQL do?

CREATE EXTENSION postgis; loads the PostGIS spatial functions and data types into your database.

SELECT postgis_full_version(); confirms that PostGIS is active by returning version and dependency information.

  • Click the Execute button (or press F5) to run the query.

You should see a result row containing PostGIS version details, library paths, and GEOS/PROJ versions.

Not seeing PostGIS version output?

Make sure you selected the geo_warehouse database (not the default postgres database) before opening the Query Tool. PostGIS must be enabled per database.

If you see an error that the extension does not exist, StackBuilder may not have completed the PostGIS installation. Reopen StackBuilder and confirm the PostGIS Bundle is checked under Spatial Extensions.

Help me troubleshoot PostGIS installation in pgAdmin

Create a virtual environment and install geospatial libraries

A virtual environment isolates this project's Python packages from your system Python. This prevents version conflicts between projects.

  • Open a terminal by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir geo-etl-pipeline
cd geo-etl-pipeline
  • Create a Python virtual environment by running:
python -m venv venv
  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

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

  • Install the geospatial library stack by running:
pip install osmnx geopandas psycopg2-binary sqlalchemy shapely

What are these libraries?

OSMnx downloads geospatial features (buildings, roads, parks) from OpenStreetMap. GeoPandas extends Pandas with geometry columns so you can manipulate spatial data in DataFrames.

psycopg2-binary is the PostgreSQL adapter for Python. SQLAlchemy provides a high-level interface for database connections. Shapely handles geometric operations like area and intersection calculations.

  • Verify OSMnx installed correctly by running:
python -c "import osmnx; print(osmnx.__version__)"

You should see a version number printed (e.g. 2.0.1 or similar).

Installation failing?

If pip install fails with build errors for shapely or geopandas, make sure your virtual environment is activated (you see (venv) in your prompt) and that you are using Python 3.9 or higher.

Help me fix pip install errors for geospatial libraries

Create the project folder structure and configuration

A well-organized folder structure makes the pipeline modular and easy to maintain. Each ETL layer gets its own directory, mirroring dbt-style project layouts.

  • Create the directory structure by running these commands:
mkdir pipeline
mkdir pipeline/extract
mkdir pipeline/staging
mkdir pipeline/intermediate
mkdir pipeline/mart
mkdir config
mkdir tests
mkdir exports

Why this folder structure?

Each folder maps to a layer in the ETL pipeline. pipeline/extract/ handles raw data ingestion. pipeline/staging/ cleans and standardizes. pipeline/intermediate/ enriches with computed metrics. pipeline/mart/ builds the final dimensional model.

config/ holds settings. tests/ holds unit tests. exports/ is where Tableau-ready files will land.

  • Confirm the directories exist by running:
ls

You should see config, exports, pipeline, tests, and venv listed in the output.

Now create the configuration file that stores database credentials, the target city, and schema names.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type the name of your code editor and press Enter to open it.
  • Open the geo-etl-pipeline folder from your Desktop in your editor.
  • Create a new file at config/settings.py and paste the following code:
"""Pipeline configuration settings."""

DATABASE = {
    "host": "localhost",
    "port": 5432,
    "dbname": "geo_warehouse",
    "user": "postgres",
    "password": "your-postgres-password-here",
}

SQLALCHEMY_URL = (
    f"postgresql://{DATABASE['user']}:{DATABASE['password']}"
    f"@{DATABASE['host']}:{DATABASE['port']}/{DATABASE['dbname']}"
)

TARGET_PLACE = "Chinatown, San Francisco, California"

OSM_TAGS = {
    "buildings": {"building": True},
    "roads": {"highway": True},
    "parks": {"leisure": "park"},
    "landuse": {"landuse": True},
}

RAW_DATA_DIR = "data/raw"
EXPORT_DIR = "exports"

SCHEMAS = {
    "raw": "raw",
    "staging": "staging",
    "intermediate": "intermediate",
    "mart": "mart",
}

What does this configuration do?

DATABASE stores your PostgreSQL connection details. Replace your-postgres-password-here with the password you set during PostgreSQL installation.

TARGET_PLACE tells OSMnx which geographic area to download. OSM_TAGS defines which feature types to extract (buildings, roads, parks, land use polygons).

SCHEMAS maps each pipeline layer to a PostgreSQL schema, keeping your data organized inside the database.

  • Replace your-postgres-password-here in the DATABASE dictionary with the password you chose during PostgreSQL installation.
  • Save config/settings.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file with your actual PostgreSQL password in place.

ⓧ I'd like to double check the full code

Here is the complete config/settings.py for comparison:

"""Pipeline configuration settings."""

DATABASE = {
    "host": "localhost",
    "port": 5432,
    "dbname": "geo_warehouse",
    "user": "postgres",
    "password": "your-postgres-password-here",
}

SQLALCHEMY_URL = (
    f"postgresql://{DATABASE['user']}:{DATABASE['password']}"
    f"@{DATABASE['host']}:{DATABASE['port']}/{DATABASE['dbname']}"
)

TARGET_PLACE = "Chinatown, San Francisco, California"

OSM_TAGS = {
    "buildings": {"building": True},
    "roads": {"highway": True},
    "parks": {"leisure": "park"},
    "landuse": {"landuse": True},
}

RAW_DATA_DIR = "data/raw"
EXPORT_DIR = "exports"

SCHEMAS = {
    "raw": "raw",
    "staging": "staging",
    "intermediate": "intermediate",
    "mart": "mart",
}

Seeing a connection error later?

If later steps fail to connect to PostgreSQL, check that the password field in config/settings.py matches exactly what you set during installation. Also verify the dbname is geo_warehouse and that you created this database in pgAdmin earlier.

Help me fix my PostgreSQL connection settings

Your environment is fully configured. Next up, you will use OSMnx to extract real geospatial data from OpenStreetMap and load it into your PostgreSQL database.

Extract Raw Geospatial Data

With your environment configured, it's time to build the extraction layer of your ETL pipeline. Every data pipeline begins by pulling raw data from its source into a format you can work with downstream.

In this step, you'll use OSMnx to download real geospatial features from OpenStreetMap. You'll grab buildings, roads, parks, and land-use polygons for Chinatown in San Francisco, save them as local GeoPackage files, then load everything into a PostgreSQL database with PostGIS enabled.

In this step, get ready to:
  • Download OpenStreetMap features using OSMnx.
  • Create the geo_warehouse database with PostGIS enabled.
  • Load raw geospatial data into PostgreSQL tables.
Download OSM data with OSMnx

The extract module connects to OpenStreetMap's Overpass API through OSMnx and downloads four categories of geospatial features: buildings, roads, parks, and land-use polygons. Each category comes back as a GeoDataFrame that you'll save locally as a GeoPackage file for reproducibility.

  • Create a new file called extract.py inside the pipeline/extract/ folder.
  • Add the imports, logger setup, and the extract_osm_data() function by pasting the following code:
"""Extract module: downloads OSM data and loads into raw schema."""

import logging
import os

import geopandas as gpd
import osmnx as ox
from sqlalchemy import create_engine, text

from config.settings import (
    OSM_TAGS,
    RAW_DATA_DIR,
    SCHEMAS,
    SQLALCHEMY_URL,
    TARGET_PLACE,
)

logger = logging.getLogger(__name__)


def extract_osm_data():
    """Download OSM features for the target place."""
    os.makedirs(RAW_DATA_DIR, exist_ok=True)
    datasets = {}

    for name, tags in OSM_TAGS.items():
        logger.info(f"Extracting {name} for {TARGET_PLACE}")
        # Download features matching the tag from OpenStreetMap
        gdf = ox.features.features_from_place(TARGET_PLACE, tags=tags)
        gdf = gdf.reset_index()
        # Save each dataset locally as a GeoPackage file
        filepath = os.path.join(RAW_DATA_DIR, f"{name}.gpkg")
        gdf.to_file(filepath, driver="GPKG")
        datasets[name] = gdf
        logger.info(f"Extracted {len(gdf)} {name} features")

    return datasets

What does this code do?

The function loops through each tag category defined in your config/settings.py (buildings, roads, parks, landuse). For each one, it calls ox.features.features_from_place() which queries OpenStreetMap's Overpass API and returns matching geospatial features as a GeoDataFrame.

Each GeoDataFrame is saved to a .gpkg file in data/raw/ so you have a local backup of the raw data. The logger records timestamps and row counts for every extraction, which is essential for debugging data pipelines in production.

  • Save pipeline/extract/extract.py.

Why save to GeoPackage files?

Saving raw data locally before loading it into the database gives you a checkpoint. If your database connection fails or you need to re-run the pipeline, you can reload from the local files without hitting the OpenStreetMap API again. This is a standard data engineering practice called "landing zone" storage.

Create the database and load raw data into PostgreSQL

Before the pipeline can load data, you need a PostgreSQL database to receive it. You'll create the geo_warehouse database and enable PostGIS for spatial data support.

  • Open pgAdmin (you used this in the previous step to verify PostGIS).
  • In the left sidebar tree, expand your server connection (PostgreSQL 17).
  • Right-click the Databases node.
  • Select Create, then Database.
  • In the Database field on the General tab, enter geo_warehouse.
  • Click Save.

Now enable PostGIS on this new database.

  • In the left sidebar, expand geo_warehouse to confirm it appears.
  • Right-click geo_warehouse and select Query Tool.
  • Enable PostGIS by running this SQL:
CREATE EXTENSION postgis;
  • Click the Execute button (play icon) or press F5.

What does CREATE EXTENSION postgis do?

This command adds geospatial data types (like geometry and geography) and spatial functions (like ST_Area, ST_IsValid) to this specific database. PostGIS must be enabled per-database, not just installed on the server.

You should see a success message in the query output panel confirming the extension was created.

Seeing an error about the extension not existing?

Make sure you installed the PostGIS bundle via StackBuilder in Step 1. If you skipped that, re-run StackBuilder from your Start menu, select Spatial Extensions, and install the PostGIS bundle for PostgreSQL 17.

Help me fix a PostGIS extension error in pgAdmin.

Now add the load_raw_to_postgres() function that pushes the extracted GeoDataFrames into PostgreSQL.

  • Return to pipeline/extract/extract.py in your code editor.
  • Add the following function below extract_osm_data():

def load_raw_to_postgres(datasets):
    """Load raw GeoDataFrames into PostgreSQL raw schema."""
    engine = create_engine(SQLALCHEMY_URL)
    schema = SCHEMAS["raw"]

    # Create the raw schema if it doesn't exist yet
    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema}"))
        conn.commit()

    for name, gdf in datasets.items():
        table_name = f"raw_{name}"
        # Keep geometry plus any relevant attribute columns
        gdf_clean = gdf[["geometry"]].copy()
        if "name" in gdf.columns:
            gdf_clean["name"] = gdf["name"]
        if "building" in gdf.columns:
            gdf_clean["building"] = gdf["building"]
        if "highway" in gdf.columns:
            gdf_clean["highway"] = gdf["highway"]
        if "landuse" in gdf.columns:
            gdf_clean["landuse"] = gdf["landuse"]
        if "leisure" in gdf.columns:
            gdf_clean["leisure"] = gdf["leisure"]

        # Drop rows with null geometry before loading
        gdf_clean = gdf_clean[gdf_clean.geometry.notnull()]
        gdf_clean.to_postgis(
            table_name, engine, schema=schema, if_exists="replace", index=False
        )
        logger.info(f"Loaded {len(gdf_clean)} rows into {schema}.{table_name}")

What does this code do?

This function creates a raw schema in PostgreSQL to hold the unprocessed data. For each dataset, it selects only the columns needed (geometry plus the relevant attribute like building or highway), drops any rows with null geometry, and loads the result into a PostGIS-enabled table.

The to_postgis() method from GeoPandas handles converting Python geometry objects into PostGIS-compatible spatial data automatically. The if_exists="replace" parameter means you can re-run the pipeline without manually dropping tables first.

  • Save pipeline/extract/extract.py.

Connection refused when loading to PostgreSQL?

Make sure the password in config/settings.py matches the password you set during PostgreSQL installation. Also confirm PostgreSQL is running by checking Services (Windows) for the postgresql-x64-17 service.

Help me troubleshoot my PostgreSQL connection from Python.

Run the extraction pipeline

With both functions written, you need a way to execute them together. The if __name__ == "__main__" block lets you run the extraction module directly from the terminal.

  • Add the following block at the bottom of pipeline/extract/extract.py:

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    datasets = extract_osm_data()
    load_raw_to_postgres(datasets)

What does this code do?

The logging.basicConfig(level=logging.INFO) line activates the logger so you can see timestamps and row counts in your terminal. The two function calls run the full extraction: first downloading OSM data, then loading it into PostgreSQL.

  • Save pipeline/extract/extract.py.
  • Make sure your config/settings.py has the correct PostgreSQL password before running.
  • Run the extraction module from your project root directory by running this command:
python -m pipeline.extract.extract

The extraction takes 30 to 60 seconds depending on your internet speed. You should see log output showing each feature type being extracted and loaded.

What should I see?

Your terminal should display lines like:

INFO:pipeline.extract.extract:Extracting buildings for Chinatown, San Francisco, California
INFO:pipeline.extract.extract:Extracted 1842 buildings features
INFO:pipeline.extract.extract:Loaded 1842 rows into raw.raw_buildings

The exact row counts will vary because OpenStreetMap is a live dataset that gets updated by contributors. Any non-zero counts mean the extraction worked.

  • Confirm the GeoPackage files were saved by running:
ls data/raw/

You should see buildings.gpkg, roads.gpkg, parks.gpkg, and landuse.gpkg listed.

  • Validate the data landed in PostgreSQL by switching back to the Query Tool in pgAdmin (connected to geo_warehouse) and running:
SELECT table_name FROM information_schema.tables WHERE table_schema = 'raw';

You should see four tables listed: raw_buildings, raw_roads, raw_parks, and raw_landuse.

Seeing a ModuleNotFoundError?

Make sure you're running the command from your project root folder (the one containing the pipeline/ and config/ directories). Also confirm your virtual environment is activated. You should see its name in your terminal prompt.

Help me fix a ModuleNotFoundError when running my extract module.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file and confirmed the raw tables exist in pgAdmin.

ⓧ I'd like to double check the full code

"""Extract module: downloads OSM data and loads into raw schema."""

import logging
import os

import geopandas as gpd
import osmnx as ox
from sqlalchemy import create_engine, text

from config.settings import (
    OSM_TAGS,
    RAW_DATA_DIR,
    SCHEMAS,
    SQLALCHEMY_URL,
    TARGET_PLACE,
)

logger = logging.getLogger(__name__)


def extract_osm_data():
    """Download OSM features for the target place."""
    os.makedirs(RAW_DATA_DIR, exist_ok=True)
    datasets = {}

    for name, tags in OSM_TAGS.items():
        logger.info(f"Extracting {name} for {TARGET_PLACE}")
        gdf = ox.features.features_from_place(TARGET_PLACE, tags=tags)
        gdf = gdf.reset_index()
        filepath = os.path.join(RAW_DATA_DIR, f"{name}.gpkg")
        gdf.to_file(filepath, driver="GPKG")
        datasets[name] = gdf
        logger.info(f"Extracted {len(gdf)} {name} features")

    return datasets


def load_raw_to_postgres(datasets):
    """Load raw GeoDataFrames into PostgreSQL raw schema."""
    engine = create_engine(SQLALCHEMY_URL)
    schema = SCHEMAS["raw"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema}"))
        conn.commit()

    for name, gdf in datasets.items():
        table_name = f"raw_{name}"
        gdf_clean = gdf[["geometry"]].copy()
        if "name" in gdf.columns:
            gdf_clean["name"] = gdf["name"]
        if "building" in gdf.columns:
            gdf_clean["building"] = gdf["building"]
        if "highway" in gdf.columns:
            gdf_clean["highway"] = gdf["highway"]
        if "landuse" in gdf.columns:
            gdf_clean["landuse"] = gdf["landuse"]
        if "leisure" in gdf.columns:
            gdf_clean["leisure"] = gdf["leisure"]

        gdf_clean = gdf_clean[gdf_clean.geometry.notnull()]
        gdf_clean.to_postgis(
            table_name, engine, schema=schema, if_exists="replace", index=False
        )
        logger.info(f"Loaded {len(gdf_clean)} rows into {schema}.{table_name}")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    datasets = extract_osm_data()
    load_raw_to_postgres(datasets)

Your raw geospatial data is now safely stored in both local GeoPackage files and PostgreSQL tables. Next up, you'll build the staging and intermediate layers that clean and enrich this data for the dimensional model.

Build Staging and Intermediate Layers

Your raw data is sitting in PostgreSQL, but it's messy. Column names are inconsistent, some geometries are invalid, and there are no computed metrics to analyze.

In professional data engineering, you never query raw data directly. Instead, you pass it through transformation layers that progressively clean, enrich, and reshape it. In this step, you will build two of those layers: staging (cleaning and standardization) and intermediate (enrichment with spatial metrics).

In this step, get ready to:
  • Build the staging layer to clean geometries and standardize column names.
  • Build the intermediate layer to compute spatial metrics like area and length.
  • Write unit tests to validate data quality across both layers.
Build the staging layer

The staging layer is the first line of defense against bad data. It standardizes geometry types, filters out invalid shapes, fills in missing values with defaults, and assigns surrogate keys for downstream joins.

  • Create the file pipeline/staging/staging.py inside your project folder.
  • Add the following staging module code:
"""Staging module: clean and standardize raw data."""

import logging

import geopandas as gpd
from sqlalchemy import create_engine, text

from config.settings import SCHEMAS, SQLALCHEMY_URL

logger = logging.getLogger(__name__)


def run_staging():
    """Create staging tables with cleaned, standardized data."""
    engine = create_engine(SQLALCHEMY_URL)
    raw = SCHEMAS["raw"]
    stg = SCHEMAS["staging"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {stg}"))
        conn.commit()

    staging_queries = {
        "stg_buildings": f"""
            DROP TABLE IF EXISTS {stg}.stg_buildings;
            CREATE TABLE {stg}.stg_buildings AS
            SELECT
                ROW_NUMBER() OVER () AS building_id,
                COALESCE(name, 'Unknown') AS building_name,
                COALESCE(building, 'yes') AS building_type,
                ST_Multi(ST_MakeValid(geometry))::geometry(MultiPolygon, 4326) AS geometry
            FROM {raw}.raw_buildings
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_Polygon', 'ST_MultiPolygon');
        """,
        "stg_roads": f"""
            DROP TABLE IF EXISTS {stg}.stg_roads;
            CREATE TABLE {stg}.stg_roads AS
            SELECT
                ROW_NUMBER() OVER () AS road_id,
                COALESCE(name, 'Unnamed') AS road_name,
                COALESCE(highway, 'unclassified') AS road_type,
                ST_MakeValid(geometry)::geometry(Geometry, 4326) AS geometry
            FROM {raw}.raw_roads
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_LineString', 'ST_MultiLineString');
        """,
        "stg_parks": f"""
            DROP TABLE IF EXISTS {stg}.stg_parks;
            CREATE TABLE {stg}.stg_parks AS
            SELECT
                ROW_NUMBER() OVER () AS park_id,
                COALESCE(name, 'Unnamed Park') AS park_name,
                ST_Multi(ST_MakeValid(geometry))::geometry(MultiPolygon, 4326) AS geometry
            FROM {raw}.raw_parks
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_Polygon', 'ST_MultiPolygon');
        """,
        "stg_landuse": f"""
            DROP TABLE IF EXISTS {stg}.stg_landuse;
            CREATE TABLE {stg}.stg_landuse AS
            SELECT
                ROW_NUMBER() OVER () AS landuse_id,
                COALESCE(name, 'Unnamed') AS landuse_name,
                COALESCE(landuse, 'unclassified') AS landuse_type,
                ST_Multi(ST_MakeValid(geometry))::geometry(MultiPolygon, 4326) AS geometry
            FROM {raw}.raw_landuse
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_Polygon', 'ST_MultiPolygon');
        """,
    }

    with engine.connect() as conn:
        for name, query in staging_queries.items():
            conn.execute(text(query))
            logger.info(f"Created {stg}.{name}")
        conn.commit()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run_staging()

What does this code do?

This module creates four staging tables in a dedicated staging schema, each cleaning one raw table:

  • ROW_NUMBER() OVER () generates surrogate keys (building_id, road_id, etc.) so every row has a unique identifier for downstream joins.
  • ST_MakeValid() repairs broken geometries, and ST_IsValid() filters out anything that cannot be repaired.
  • ST_Multi() forces all polygon features to MultiPolygon so the geometry column has a consistent type.
  • COALESCE() fills null names and types with sensible defaults like 'Unknown' or 'unclassified'.
  • Save pipeline/staging/staging.py.
  • Run the staging layer by executing this command from your project root:
python -m pipeline.staging.staging

You should see log messages confirming each table was created in the staging schema.

Seeing a permission or schema error?

  • Make sure your config/settings.py has the correct password for your postgres user.
  • Confirm the raw schema tables exist by checking pgAdmin. If they are empty, re-run the extraction step first.
  • Check that PostGIS is enabled in your geo_warehouse database. Functions like ST_MakeValid require it.

Help me debug a schema or permission error when running my staging layer.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline/staging/staging.py and that you saw log output confirming all four staging tables were created.

ⓧ I'd like to double check the full code

"""Staging module: clean and standardize raw data."""

import logging

import geopandas as gpd
from sqlalchemy import create_engine, text

from config.settings import SCHEMAS, SQLALCHEMY_URL

logger = logging.getLogger(__name__)


def run_staging():
    """Create staging tables with cleaned, standardized data."""
    engine = create_engine(SQLALCHEMY_URL)
    raw = SCHEMAS["raw"]
    stg = SCHEMAS["staging"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {stg}"))
        conn.commit()

    staging_queries = {
        "stg_buildings": f"""
            DROP TABLE IF EXISTS {stg}.stg_buildings;
            CREATE TABLE {stg}.stg_buildings AS
            SELECT
                ROW_NUMBER() OVER () AS building_id,
                COALESCE(name, 'Unknown') AS building_name,
                COALESCE(building, 'yes') AS building_type,
                ST_Multi(ST_MakeValid(geometry))::geometry(MultiPolygon, 4326) AS geometry
            FROM {raw}.raw_buildings
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_Polygon', 'ST_MultiPolygon');
        """,
        "stg_roads": f"""
            DROP TABLE IF EXISTS {stg}.stg_roads;
            CREATE TABLE {stg}.stg_roads AS
            SELECT
                ROW_NUMBER() OVER () AS road_id,
                COALESCE(name, 'Unnamed') AS road_name,
                COALESCE(highway, 'unclassified') AS road_type,
                ST_MakeValid(geometry)::geometry(Geometry, 4326) AS geometry
            FROM {raw}.raw_roads
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_LineString', 'ST_MultiLineString');
        """,
        "stg_parks": f"""
            DROP TABLE IF EXISTS {stg}.stg_parks;
            CREATE TABLE {stg}.stg_parks AS
            SELECT
                ROW_NUMBER() OVER () AS park_id,
                COALESCE(name, 'Unnamed Park') AS park_name,
                ST_Multi(ST_MakeValid(geometry))::geometry(MultiPolygon, 4326) AS geometry
            FROM {raw}.raw_parks
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_Polygon', 'ST_MultiPolygon');
        """,
        "stg_landuse": f"""
            DROP TABLE IF EXISTS {stg}.stg_landuse;
            CREATE TABLE {stg}.stg_landuse AS
            SELECT
                ROW_NUMBER() OVER () AS landuse_id,
                COALESCE(name, 'Unnamed') AS landuse_name,
                COALESCE(landuse, 'unclassified') AS landuse_type,
                ST_Multi(ST_MakeValid(geometry))::geometry(MultiPolygon, 4326) AS geometry
            FROM {raw}.raw_landuse
            WHERE ST_IsValid(ST_MakeValid(geometry))
              AND ST_GeometryType(geometry) IN ('ST_Polygon', 'ST_MultiPolygon');
        """,
    }

    with engine.connect() as conn:
        for name, query in staging_queries.items():
            conn.execute(text(query))
            logger.info(f"Created {stg}.{name}")
        conn.commit()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run_staging()
Build the intermediate layer

The intermediate layer takes clean staged data and enriches it with computed spatial metrics. This is where your geospatial data becomes analytically useful.

Why a separate intermediate layer?

In dbt-style architectures, staging handles cleaning while intermediate handles enrichment. Keeping them separate means you can re-run spatial calculations without re-cleaning raw data.

The intermediate layer computes area in square meters, perimeter, road segment lengths, and centroid coordinates. These are the building blocks for analytics in the mart layer.

  • Create the file pipeline/intermediate/intermediate.py inside your project folder.
  • Add the following intermediate module code:
"""Intermediate module: enrich staged data with computed metrics."""

import logging

from sqlalchemy import create_engine, text

from config.settings import SCHEMAS, SQLALCHEMY_URL

logger = logging.getLogger(__name__)


def run_intermediate():
    """Create intermediate tables with spatial metrics and classifications."""
    engine = create_engine(SQLALCHEMY_URL)
    stg = SCHEMAS["staging"]
    int_schema = SCHEMAS["intermediate"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {int_schema}"))
        conn.commit()

    intermediate_queries = {
        "int_buildings": f"""
            DROP TABLE IF EXISTS {int_schema}.int_buildings;
            CREATE TABLE {int_schema}.int_buildings AS
            SELECT
                building_id,
                building_name,
                building_type,
                ST_Area(geometry::geography) AS area_sqm,
                ST_Perimeter(geometry::geography) AS perimeter_m,
                ST_Y(ST_Centroid(geometry)) AS centroid_lat,
                ST_X(ST_Centroid(geometry)) AS centroid_lon,
                'building' AS feature_category,
                geometry
            FROM {stg}.stg_buildings;
        """,
        "int_roads": f"""
            DROP TABLE IF EXISTS {int_schema}.int_roads;
            CREATE TABLE {int_schema}.int_roads AS
            SELECT
                road_id,
                road_name,
                road_type,
                ST_Length(geometry::geography) AS length_m,
                ST_Y(ST_Centroid(geometry)) AS centroid_lat,
                ST_X(ST_Centroid(geometry)) AS centroid_lon,
                'road' AS feature_category,
                geometry
            FROM {stg}.stg_roads;
        """,
        "int_parks": f"""
            DROP TABLE IF EXISTS {int_schema}.int_parks;
            CREATE TABLE {int_schema}.int_parks AS
            SELECT
                park_id,
                park_name,
                ST_Area(geometry::geography) AS area_sqm,
                ST_Perimeter(geometry::geography) AS perimeter_m,
                ST_Y(ST_Centroid(geometry)) AS centroid_lat,
                ST_X(ST_Centroid(geometry)) AS centroid_lon,
                'park' AS feature_category,
                geometry
            FROM {stg}.stg_parks;
        """,
    }

    with engine.connect() as conn:
        for name, query in intermediate_queries.items():
            conn.execute(text(query))
            logger.info(f"Created {int_schema}.{name}")
        conn.commit()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run_intermediate()

What does this code do?

This module creates three intermediate tables, each enriching staged data with spatial computations:

  • ST_Area(geometry::geography) calculates area in square meters by casting to a geography type. Without the cast, you would get area in degrees (not useful).
  • ST_Length(geometry::geography) computes road segment length in meters using the same geography cast.
  • ST_Y(ST_Centroid(geometry)) and ST_X(ST_Centroid(geometry)) extract latitude and longitude of each feature's center point.
  • A feature_category column classifies each row for use in the dimensional model later.
  • Save pipeline/intermediate/intermediate.py.
  • Run the intermediate layer by executing this command:
python -m pipeline.intermediate.intermediate

You should see log messages confirming three intermediate tables were created.

Seeing an error about stg_buildings not existing?

  • The intermediate layer depends on staging tables. Make sure you successfully ran the staging module first.
  • Open pgAdmin and check that the staging schema exists with tables like stg_buildings inside it.

Help me debug why my intermediate layer cannot find the staging tables.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline/intermediate/intermediate.py and that you saw log output confirming all three intermediate tables were created.

ⓧ I'd like to double check the full code

"""Intermediate module: enrich staged data with computed metrics."""

import logging

from sqlalchemy import create_engine, text

from config.settings import SCHEMAS, SQLALCHEMY_URL

logger = logging.getLogger(__name__)


def run_intermediate():
    """Create intermediate tables with spatial metrics and classifications."""
    engine = create_engine(SQLALCHEMY_URL)
    stg = SCHEMAS["staging"]
    int_schema = SCHEMAS["intermediate"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {int_schema}"))
        conn.commit()

    intermediate_queries = {
        "int_buildings": f"""
            DROP TABLE IF EXISTS {int_schema}.int_buildings;
            CREATE TABLE {int_schema}.int_buildings AS
            SELECT
                building_id,
                building_name,
                building_type,
                ST_Area(geometry::geography) AS area_sqm,
                ST_Perimeter(geometry::geography) AS perimeter_m,
                ST_Y(ST_Centroid(geometry)) AS centroid_lat,
                ST_X(ST_Centroid(geometry)) AS centroid_lon,
                'building' AS feature_category,
                geometry
            FROM {stg}.stg_buildings;
        """,
        "int_roads": f"""
            DROP TABLE IF EXISTS {int_schema}.int_roads;
            CREATE TABLE {int_schema}.int_roads AS
            SELECT
                road_id,
                road_name,
                road_type,
                ST_Length(geometry::geography) AS length_m,
                ST_Y(ST_Centroid(geometry)) AS centroid_lat,
                ST_X(ST_Centroid(geometry)) AS centroid_lon,
                'road' AS feature_category,
                geometry
            FROM {stg}.stg_roads;
        """,
        "int_parks": f"""
            DROP TABLE IF EXISTS {int_schema}.int_parks;
            CREATE TABLE {int_schema}.int_parks AS
            SELECT
                park_id,
                park_name,
                ST_Area(geometry::geography) AS area_sqm,
                ST_Perimeter(geometry::geography) AS perimeter_m,
                ST_Y(ST_Centroid(geometry)) AS centroid_lat,
                ST_X(ST_Centroid(geometry)) AS centroid_lon,
                'park' AS feature_category,
                geometry
            FROM {stg}.stg_parks;
        """,
    }

    with engine.connect() as conn:
        for name, query in intermediate_queries.items():
            conn.execute(text(query))
            logger.info(f"Created {int_schema}.{name}")
        conn.commit()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run_intermediate()
Write unit tests for data quality

Automated tests catch data quality issues before they reach your mart layer. You will write tests that verify staging produced no null geometries and that intermediate area calculations are positive.

  • Install pytest by running this command:
pip install pytest
  • Create the file tests/test_pipeline.py inside your project folder.
  • Add the following test code:
"""Unit tests for the geospatial ETL pipeline."""

import pytest
from sqlalchemy import create_engine, text

from config.settings import SCHEMAS, SQLALCHEMY_URL


@pytest.fixture
def engine():
    return create_engine(SQLALCHEMY_URL)


def test_staging_no_null_geometries(engine):
    """Staging tables should have no null geometries."""
    stg = SCHEMAS["staging"]
    with engine.connect() as conn:
        result = conn.execute(
            text(f"SELECT COUNT(*) FROM {stg}.stg_buildings WHERE geometry IS NULL")
        )
        null_count = result.scalar()
    assert null_count == 0, f"Found {null_count} null geometries in stg_buildings"


def test_intermediate_positive_areas(engine):
    """Intermediate building areas should all be positive."""
    int_schema = SCHEMAS["intermediate"]
    with engine.connect() as conn:
        result = conn.execute(
            text(f"SELECT COUNT(*) FROM {int_schema}.int_buildings WHERE area_sqm <= 0")
        )
        bad_count = result.scalar()
    assert bad_count == 0, f"Found {bad_count} buildings with non-positive area"


def test_fact_table_populated(engine):
    """Fact table should contain data after pipeline runs."""
    mart = SCHEMAS["mart"]
    with engine.connect() as conn:
        result = conn.execute(
            text(f"SELECT COUNT(*) FROM {mart}.fact_land_features")
        )
        row_count = result.scalar()
    assert row_count > 0, "Fact table is empty"

What do these tests check?

  • test_staging_no_null_geometries queries the staging buildings table and asserts zero rows have a null geometry. This confirms ST_IsValid filtering worked correctly.
  • test_intermediate_positive_areas checks that every building in the intermediate layer has a positive area. A zero or negative area would indicate a geometry casting problem.
  • test_fact_table_populated will be used later to verify the mart layer has data. It will fail for now since the mart layer does not exist yet.
  • Save tests/test_pipeline.py.
  • Run the first two tests by executing this command:
pytest tests/test_pipeline.py::test_staging_no_null_geometries tests/test_pipeline.py::test_intermediate_positive_areas -v

You should see both tests pass with green PASSED indicators.

Why not run all three tests?

The third test (test_fact_table_populated) checks the mart layer, which you have not built yet. Running it now would produce an expected failure. You will run the full test suite after building the mart in the next step.

Tests failing unexpectedly?

  • Make sure both the staging and intermediate modules ran successfully before testing.
  • Verify your config/settings.py password matches your actual PostgreSQL password. The tests connect to the live database.
  • If you see ModuleNotFoundError, make sure you are running pytest from your project root folder (the one containing config/ and pipeline/).

Help me figure out why my pipeline tests are failing.

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_pipeline.py and that both tests showed PASSED.

ⓧ I'd like to double check the full code

"""Unit tests for the geospatial ETL pipeline."""

import pytest
from sqlalchemy import create_engine, text

from config.settings import SCHEMAS, SQLALCHEMY_URL


@pytest.fixture
def engine():
    return create_engine(SQLALCHEMY_URL)


def test_staging_no_null_geometries(engine):
    """Staging tables should have no null geometries."""
    stg = SCHEMAS["staging"]
    with engine.connect() as conn:
        result = conn.execute(
            text(f"SELECT COUNT(*) FROM {stg}.stg_buildings WHERE geometry IS NULL")
        )
        null_count = result.scalar()
    assert null_count == 0, f"Found {null_count} null geometries in stg_buildings"


def test_intermediate_positive_areas(engine):
    """Intermediate building areas should all be positive."""
    int_schema = SCHEMAS["intermediate"]
    with engine.connect() as conn:
        result = conn.execute(
            text(f"SELECT COUNT(*) FROM {int_schema}.int_buildings WHERE area_sqm <= 0")
        )
        bad_count = result.scalar()
    assert bad_count == 0, f"Found {bad_count} buildings with non-positive area"


def test_fact_table_populated(engine):
    """Fact table should contain data after pipeline runs."""
    mart = SCHEMAS["mart"]
    with engine.connect() as conn:
        result = conn.execute(
            text(f"SELECT COUNT(*) FROM {mart}.fact_land_features")
        )
        row_count = result.scalar()
    assert row_count > 0, "Fact table is empty"

Your data now flows cleanly from raw through staging to intermediate, with automated tests guarding quality at each layer. Next up, you will build the mart layer with a star schema dimensional model that turns this processed data into analytics-ready tables.

Build the Mart Layer and Dimensional Model

Your intermediate layer now holds enriched geospatial features with computed metrics like area, perimeter, and centroids. The next challenge is structuring this data for analytical consumption.

In a star schema, you separate your data into a central fact table (measurable events) surrounded by dimension tables (descriptive context). This is the mart layer in the dbt-style architecture. It transforms your enriched geodata into a shape optimized for BI tools like Tableau.

In this step, get ready to:
  • Design and create a star schema with fact and dimension tables.
  • Populate the fact table from your intermediate layer data.
  • Export mart data as GeoJSON and CSV for Tableau consumption.
Design the star schema dimensional model

A star schema puts one fact table at the center, surrounded by dimension tables that give context to each fact row. Your fact table will store every land feature (buildings, roads, parks) with its measurements. The dimension tables describe what type of land use each feature represents, and when it was extracted.

  • Create a new file at pipeline/mart/mart.py inside your project folder.
  • Add the imports and the create_dimensional_model() function by pasting this code:
"""Mart module: create dimensional model and export for BI."""

import logging
import os

import geopandas as gpd
import pandas as pd
from sqlalchemy import create_engine, text

from config.settings import EXPORT_DIR, SCHEMAS, SQLALCHEMY_URL

logger = logging.getLogger(__name__)


def create_dimensional_model():
    """Create star schema tables in the mart schema."""
    engine = create_engine(SQLALCHEMY_URL)
    mart = SCHEMAS["mart"]
    int_schema = SCHEMAS["intermediate"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {mart}"))
        conn.commit()

    ddl = f"""
        DROP TABLE IF EXISTS {mart}.dim_land_use_type CASCADE;
        CREATE TABLE {mart}.dim_land_use_type (
            land_use_type_key SERIAL PRIMARY KEY,
            category VARCHAR(50) NOT NULL,
            subcategory VARCHAR(100),
            description VARCHAR(255)
        );

        INSERT INTO {mart}.dim_land_use_type (category, subcategory, description)
        VALUES
            ('building', 'residential', 'Residential buildings'),
            ('building', 'commercial', 'Commercial buildings'),
            ('building', 'yes', 'Unclassified buildings'),
            ('road', 'primary', 'Primary roads'),
            ('road', 'secondary', 'Secondary roads'),
            ('road', 'residential', 'Residential streets'),
            ('road', 'service', 'Service roads'),
            ('road', 'unclassified', 'Unclassified roads'),
            ('park', 'park', 'Public parks and recreation'),
            ('landuse', 'residential', 'Residential land use'),
            ('landuse', 'commercial', 'Commercial land use'),
            ('landuse', 'unclassified', 'Unclassified land use');

        DROP TABLE IF EXISTS {mart}.dim_date CASCADE;
        CREATE TABLE {mart}.dim_date (
            date_key SERIAL PRIMARY KEY,
            extraction_date DATE NOT NULL DEFAULT CURRENT_DATE,
            year INTEGER,
            month INTEGER,
            day_of_week INTEGER
        );

        INSERT INTO {mart}.dim_date (extraction_date, year, month, day_of_week)
        VALUES (CURRENT_DATE, EXTRACT(YEAR FROM CURRENT_DATE),
                EXTRACT(MONTH FROM CURRENT_DATE),
                EXTRACT(DOW FROM CURRENT_DATE));

        DROP TABLE IF EXISTS {mart}.fact_land_features CASCADE;
        CREATE TABLE {mart}.fact_land_features (
            feature_id SERIAL PRIMARY KEY,
            feature_category VARCHAR(50),
            feature_name VARCHAR(255),
            feature_subtype VARCHAR(100),
            land_use_type_key INTEGER REFERENCES {mart}.dim_land_use_type(land_use_type_key),
            date_key INTEGER REFERENCES {mart}.dim_date(date_key),
            area_sqm DOUBLE PRECISION,
            perimeter_m DOUBLE PRECISION,
            length_m DOUBLE PRECISION,
            centroid_lat DOUBLE PRECISION,
            centroid_lon DOUBLE PRECISION,
            geometry geometry(Geometry, 4326)
        );
    """

    with engine.connect() as conn:
        conn.execute(text(ddl))
        conn.commit()
        logger.info("Dimensional model created")

What does this code do?

This function creates the three tables that form your star schema:

  • dim_land_use_type holds 12 rows describing categories (building, road, park, landuse) and their subcategories. This is the "what kind" dimension.
  • dim_date records the extraction date. In a production warehouse, this would grow over time as you re-run the pipeline on a schedule.
  • fact_land_features is the central fact table with foreign keys pointing to each dimension, plus numeric measures (area, perimeter, length, coordinates) and the geometry itself.

Why a star schema instead of flat tables?

Flat tables duplicate descriptive text on every row. A star schema normalizes that context into small dimension tables, making queries faster and analytics cleaner. BI tools like Tableau are optimized to work with this pattern.

  • Save pipeline/mart/mart.py.

Seeing an import error?

Make sure the pipeline/mart/ directory exists. You may need to create an empty __init__.py file inside it if Python cannot find the module.

Help me debug my mart module imports.

Populate the fact table from intermediate data

With the schema in place, you need to move data from the intermediate layer into the fact table. Each INSERT...SELECT joins intermediate features with their matching dimension row so the foreign key relationship is established.

  • In pipeline/mart/mart.py, add the populate_facts() function below create_dimensional_model():

def populate_facts():
    """Populate fact table from intermediate layer."""
    engine = create_engine(SQLALCHEMY_URL)
    mart = SCHEMAS["mart"]
    int_schema = SCHEMAS["intermediate"]

    insert_sql = f"""
        INSERT INTO {mart}.fact_land_features
            (feature_category, feature_name, feature_subtype, land_use_type_key,
             date_key, area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry)
        SELECT
            'building', building_name, building_type,
            d.land_use_type_key, 1,
            area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry
        FROM {int_schema}.int_buildings b
        LEFT JOIN {mart}.dim_land_use_type d
            ON d.category = 'building'
            AND d.subcategory = b.building_type
        ;

        INSERT INTO {mart}.fact_land_features
            (feature_category, feature_name, feature_subtype, land_use_type_key,
             date_key, length_m, centroid_lat, centroid_lon, geometry)
        SELECT
            'road', road_name, road_type,
            d.land_use_type_key, 1,
            length_m, centroid_lat, centroid_lon, geometry
        FROM {int_schema}.int_roads r
        LEFT JOIN {mart}.dim_land_use_type d
            ON d.category = 'road'
            AND d.subcategory = r.road_type
        ;

        INSERT INTO {mart}.fact_land_features
            (feature_category, feature_name, feature_subtype, land_use_type_key,
             date_key, area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry)
        SELECT
            'park', park_name, 'park',
            d.land_use_type_key, 1,
            area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry
        FROM {int_schema}.int_parks p
        LEFT JOIN {mart}.dim_land_use_type d
            ON d.category = 'park'
            AND d.subcategory = 'park'
        ;
    """

    with engine.connect() as conn:
        conn.execute(text(insert_sql))
        conn.commit()
        logger.info("Fact table populated")

What does this code do?

Three INSERT...SELECT statements move buildings, roads, and parks from the intermediate schema into the single fact_land_features table. Each joins to dim_land_use_type on category and subcategory to establish the foreign key.

The date_key is hardcoded to 1 because there is only one date row (today's extraction). In a production system, you would look up the correct date key dynamically.

  • Save pipeline/mart/mart.py.

Getting a foreign key violation?

If populate_facts() fails with a foreign key error, the dimension tables may not have been created yet. Make sure you call create_dimensional_model() before populate_facts().

Help me fix a foreign key constraint error in my mart module.

Export mart data and create the pipeline orchestrator

Tableau Public cannot connect directly to PostgreSQL. It only supports file-based sources like GeoJSON and CSV. Your export function will read the fact table joined with dimensions and write both file formats to the exports/ directory.

  • In pipeline/mart/mart.py, add the export_for_tableau() function and the if __name__ block below populate_facts():

def export_for_tableau():
    """Export mart data as GeoJSON and CSV for Tableau Public."""
    engine = create_engine(SQLALCHEMY_URL)
    mart = SCHEMAS["mart"]
    os.makedirs(EXPORT_DIR, exist_ok=True)

    gdf = gpd.read_postgis(
        f"""
        SELECT f.*, d.category, d.subcategory, d.description
        FROM {mart}.fact_land_features f
        LEFT JOIN {mart}.dim_land_use_type d
            ON f.land_use_type_key = d.land_use_type_key
        """,
        engine,
        geom_col="geometry",
    )

    geojson_path = os.path.join(EXPORT_DIR, "land_features.geojson")
    gdf.to_file(geojson_path, driver="GeoJSON")
    logger.info(f"Exported GeoJSON: {geojson_path}")

    csv_df = gdf.drop(columns=["geometry"]).copy()
    csv_path = os.path.join(EXPORT_DIR, "land_features.csv")
    csv_df.to_csv(csv_path, index=False)
    logger.info(f"Exported CSV: {csv_path}")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    create_dimensional_model()
    populate_facts()
    export_for_tableau()

What does this code do?

  • gpd.read_postgis() reads the fact table joined with dimension descriptions back into a GeoDataFrame.
  • The GeoJSON export preserves geometry for spatial visualizations in Tableau (map layers).
  • The CSV export drops geometry (which Tableau cannot parse as text) and keeps only tabular columns for charts and KPI sheets.
  • Save pipeline/mart/mart.py.

Now create the main pipeline orchestrator that ties all stages together in sequence.

  • Create a new file at pipeline/run_pipeline.py and paste this code:
"""Main pipeline orchestrator."""

import logging

from pipeline.extract.extract import extract_osm_data, load_raw_to_postgres
from pipeline.staging.staging import run_staging
from pipeline.intermediate.intermediate import run_intermediate
from pipeline.mart.mart import create_dimensional_model, populate_facts, export_for_tableau

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


def main():
    """Run the full ETL pipeline."""
    logger.info("Starting geospatial ETL pipeline")

    logger.info("Step 1: Extracting OSM data")
    datasets = extract_osm_data()
    load_raw_to_postgres(datasets)

    logger.info("Step 2: Running staging layer")
    run_staging()

    logger.info("Step 3: Running intermediate layer")
    run_intermediate()

    logger.info("Step 4: Building mart dimensional model")
    create_dimensional_model()
    populate_facts()

    logger.info("Step 5: Exporting for Tableau")
    export_for_tableau()

    logger.info("Pipeline complete")


if __name__ == "__main__":
    main()

What does this code do?

This orchestrator calls every pipeline stage in the correct dependency order: extract, staging, intermediate, mart creation, fact population, and export. The timestamped log format makes it easy to trace execution when debugging.

  • Save pipeline/run_pipeline.py.
  • Run the mart module directly to create and populate your dimensional model by running:
python -m pipeline.mart.mart
  • You should see log messages confirming Dimensional model created, Fact table populated, and export paths for both GeoJSON and CSV files.
  • Confirm the export files exist by running:
ls exports/
  • You should see land_features.geojson and land_features.csv in the output.

Seeing an empty exports directory?

If the export files are missing, check that the intermediate tables exist first. The mart module depends on int_buildings, int_roads, and int_parks being populated from the previous step.

Also confirm your database password in config/settings.py matches the one you set during PostgreSQL installation.

Help me debug why my mart export produced no files.

✔️ Awesome, I've got everything!

Great. Double check you have saved both pipeline/mart/mart.py and pipeline/run_pipeline.py.

ⓧ I'd like to double check the full code

Here is the complete pipeline/mart/mart.py:

"""Mart module: create dimensional model and export for BI."""

import logging
import os

import geopandas as gpd
import pandas as pd
from sqlalchemy import create_engine, text

from config.settings import EXPORT_DIR, SCHEMAS, SQLALCHEMY_URL

logger = logging.getLogger(__name__)


def create_dimensional_model():
    """Create star schema tables in the mart schema."""
    engine = create_engine(SQLALCHEMY_URL)
    mart = SCHEMAS["mart"]
    int_schema = SCHEMAS["intermediate"]

    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {mart}"))
        conn.commit()

    ddl = f"""
        DROP TABLE IF EXISTS {mart}.dim_land_use_type CASCADE;
        CREATE TABLE {mart}.dim_land_use_type (
            land_use_type_key SERIAL PRIMARY KEY,
            category VARCHAR(50) NOT NULL,
            subcategory VARCHAR(100),
            description VARCHAR(255)
        );

        INSERT INTO {mart}.dim_land_use_type (category, subcategory, description)
        VALUES
            ('building', 'residential', 'Residential buildings'),
            ('building', 'commercial', 'Commercial buildings'),
            ('building', 'yes', 'Unclassified buildings'),
            ('road', 'primary', 'Primary roads'),
            ('road', 'secondary', 'Secondary roads'),
            ('road', 'residential', 'Residential streets'),
            ('road', 'service', 'Service roads'),
            ('road', 'unclassified', 'Unclassified roads'),
            ('park', 'park', 'Public parks and recreation'),
            ('landuse', 'residential', 'Residential land use'),
            ('landuse', 'commercial', 'Commercial land use'),
            ('landuse', 'unclassified', 'Unclassified land use');

        DROP TABLE IF EXISTS {mart}.dim_date CASCADE;
        CREATE TABLE {mart}.dim_date (
            date_key SERIAL PRIMARY KEY,
            extraction_date DATE NOT NULL DEFAULT CURRENT_DATE,
            year INTEGER,
            month INTEGER,
            day_of_week INTEGER
        );

        INSERT INTO {mart}.dim_date (extraction_date, year, month, day_of_week)
        VALUES (CURRENT_DATE, EXTRACT(YEAR FROM CURRENT_DATE),
                EXTRACT(MONTH FROM CURRENT_DATE),
                EXTRACT(DOW FROM CURRENT_DATE));

        DROP TABLE IF EXISTS {mart}.fact_land_features CASCADE;
        CREATE TABLE {mart}.fact_land_features (
            feature_id SERIAL PRIMARY KEY,
            feature_category VARCHAR(50),
            feature_name VARCHAR(255),
            feature_subtype VARCHAR(100),
            land_use_type_key INTEGER REFERENCES {mart}.dim_land_use_type(land_use_type_key),
            date_key INTEGER REFERENCES {mart}.dim_date(date_key),
            area_sqm DOUBLE PRECISION,
            perimeter_m DOUBLE PRECISION,
            length_m DOUBLE PRECISION,
            centroid_lat DOUBLE PRECISION,
            centroid_lon DOUBLE PRECISION,
            geometry geometry(Geometry, 4326)
        );
    """

    with engine.connect() as conn:
        conn.execute(text(ddl))
        conn.commit()
        logger.info("Dimensional model created")


def populate_facts():
    """Populate fact table from intermediate layer."""
    engine = create_engine(SQLALCHEMY_URL)
    mart = SCHEMAS["mart"]
    int_schema = SCHEMAS["intermediate"]

    insert_sql = f"""
        INSERT INTO {mart}.fact_land_features
            (feature_category, feature_name, feature_subtype, land_use_type_key,
             date_key, area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry)
        SELECT
            'building', building_name, building_type,
            d.land_use_type_key, 1,
            area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry
        FROM {int_schema}.int_buildings b
        LEFT JOIN {mart}.dim_land_use_type d
            ON d.category = 'building'
            AND d.subcategory = b.building_type
        ;

        INSERT INTO {mart}.fact_land_features
            (feature_category, feature_name, feature_subtype, land_use_type_key,
             date_key, length_m, centroid_lat, centroid_lon, geometry)
        SELECT
            'road', road_name, road_type,
            d.land_use_type_key, 1,
            length_m, centroid_lat, centroid_lon, geometry
        FROM {int_schema}.int_roads r
        LEFT JOIN {mart}.dim_land_use_type d
            ON d.category = 'road'
            AND d.subcategory = r.road_type
        ;

        INSERT INTO {mart}.fact_land_features
            (feature_category, feature_name, feature_subtype, land_use_type_key,
             date_key, area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry)
        SELECT
            'park', park_name, 'park',
            d.land_use_type_key, 1,
            area_sqm, perimeter_m, centroid_lat, centroid_lon, geometry
        FROM {int_schema}.int_parks p
        LEFT JOIN {mart}.dim_land_use_type d
            ON d.category = 'park'
            AND d.subcategory = 'park'
        ;
    """

    with engine.connect() as conn:
        conn.execute(text(insert_sql))
        conn.commit()
        logger.info("Fact table populated")


def export_for_tableau():
    """Export mart data as GeoJSON and CSV for Tableau Public."""
    engine = create_engine(SQLALCHEMY_URL)
    mart = SCHEMAS["mart"]
    os.makedirs(EXPORT_DIR, exist_ok=True)

    gdf = gpd.read_postgis(
        f"""
        SELECT f.*, d.category, d.subcategory, d.description
        FROM {mart}.fact_land_features f
        LEFT JOIN {mart}.dim_land_use_type d
            ON f.land_use_type_key = d.land_use_type_key
        """,
        engine,
        geom_col="geometry",
    )

    geojson_path = os.path.join(EXPORT_DIR, "land_features.geojson")
    gdf.to_file(geojson_path, driver="GeoJSON")
    logger.info(f"Exported GeoJSON: {geojson_path}")

    csv_df = gdf.drop(columns=["geometry"]).copy()
    csv_path = os.path.join(EXPORT_DIR, "land_features.csv")
    csv_df.to_csv(csv_path, index=False)
    logger.info(f"Exported CSV: {csv_path}")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    create_dimensional_model()
    populate_facts()
    export_for_tableau()

And here is the complete pipeline/run_pipeline.py:

"""Main pipeline orchestrator."""

import logging

from pipeline.extract.extract import extract_osm_data, load_raw_to_postgres
from pipeline.staging.staging import run_staging
from pipeline.intermediate.intermediate import run_intermediate
from pipeline.mart.mart import create_dimensional_model, populate_facts, export_for_tableau

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


def main():
    """Run the full ETL pipeline."""
    logger.info("Starting geospatial ETL pipeline")

    logger.info("Step 1: Extracting OSM data")
    datasets = extract_osm_data()
    load_raw_to_postgres(datasets)

    logger.info("Step 2: Running staging layer")
    run_staging()

    logger.info("Step 3: Running intermediate layer")
    run_intermediate()

    logger.info("Step 4: Building mart dimensional model")
    create_dimensional_model()
    populate_facts()

    logger.info("Step 5: Exporting for Tableau")
    export_for_tableau()

    logger.info("Pipeline complete")


if __name__ == "__main__":
    main()

Your dimensional warehouse is live and your export files are ready. Next up, you will connect these files to Tableau Public and build spatial visualizations that bring your land-use analytics to life.

Build Your Tableau Public Dashboard

Your dimensional warehouse is loaded with clean, enriched geospatial data. The mart layer exports are sitting in your exports/ folder, ready for visualization.

Now you need a way to see your data spatially and share those insights publicly. Tableau Public is a free BI tool that can render GeoJSON files as interactive maps and charts. In this step, you will connect your exported data to Tableau and build a dashboard that visualizes land-use patterns in Chinatown.

In this step, get ready to:
  • Connect your GeoJSON spatial export to Tableau Public and create a map visualization.
  • Build analytical charts from the CSV export showing land-use distribution and building density.
  • Assemble and publish a multi-sheet dashboard with filters to Tableau Public.
Connect to the GeoJSON spatial file and create a map

Tableau Desktop Public Edition reads GeoJSON files natively. You will connect to the exported spatial file and build a map showing building footprints colored by their land-use category.

  • If you do not already have Tableau Desktop Public Edition installed, open your browser and navigate to:
  • https://public.tableau.com
  • Click DOWNLOAD THE APP.
  • Fill out the registration form and click DOWNLOAD THE APP on the form to start the download.
  • Run the installer and follow the prompts to complete the installation.

Why Tableau Public instead of a direct database connection?

Tableau Public does not support live connections to databases. It only works with file-based data sources like CSV, Excel, and GeoJSON. That is why your pipeline exports mart data as files in the exports/ directory.

  • Open Tableau Desktop Public Edition by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Tableau Public and press Enter to open it.
  • On the start screen, under Connect, look for the To a File section.
  • Click Spatial file.
  • Navigate to your project's exports/ folder on the Desktop.
  • Select land_features.geojson and click Open.

Tableau loads the spatial data and shows it on the Data Source page. You should see a Geometry column along with fields like feature_category, area_sqm, and category.

  • Click the Sheet 1 tab at the bottom to open a new worksheet.
  • Double-click Geometry in the Measures pane on the left. Tableau automatically creates a map with your spatial features.
  • Drag Category from the Dimensions pane to the Color shelf on the Marks card. Each land-use type (building, road, park) now displays in a different color.
  • Drag Area Sqm to the Tooltip shelf on the Marks card.
  • Drag Feature Category to the Tooltip shelf as well. Now hovering over any feature shows its area and category.
  • Right-click the Sheet 1 tab at the bottom and select Rename. Name it Land Use Map.

You should see colored polygons on the map representing Chinatown's buildings, roads, and parks. Hovering over a polygon shows its area and feature category in the tooltip.

Map is blank or shows no features?

  • Make sure you selected Spatial file as the connector type. If you used Text file, Tableau cannot read the geometry data.
  • Confirm that the land_features.geojson file was exported successfully by checking its size. It should be larger than a few kilobytes.
  • If no geometry column appears, re-run the export step by running python -m pipeline.mart.mart to regenerate the GeoJSON.

Help me troubleshoot why my Tableau map is not showing spatial features from my GeoJSON file.

Add the CSV data source and build analytical charts

The map shows spatial layout, but you also need charts that summarize the data numerically. You will add the CSV export as a second data source for building bar charts, treemaps, and KPI summaries.

  • Click Data in the top menu bar.
  • Click New Data Source.
  • Select Text file from the connector list.
  • Navigate to your exports/ folder, select land_features.csv, and click Open.

Tableau loads the CSV and shows a preview on the Data Source page. You should see columns like feature_category, area_sqm, category, and subcategory.

Now build three sheets: a bar chart, a treemap, and a KPI summary.

  • Create a new worksheet by clicking the New Worksheet icon (the tab with a plus sign at the bottom).
  • Make sure the land_features.csv data source is selected in the Data pane.
  • Drag Category to the Rows shelf.
  • Drag land_features.csv (Count) or the Number of Records measure to the Columns shelf. This creates a horizontal bar chart showing the count of features per land-use category.
  • Drag Category to the Color shelf to match colors with the map.
  • Rename this sheet to Land Use Distribution.
  • Create another new worksheet.
  • Drag Subcategory to the Rows shelf. Then drag it again to the Label shelf on the Marks card.
  • Drag Area Sqm (set to SUM) to the Size shelf on the Marks card.
  • In the Show Me panel on the top right, select the treemap chart type.
  • Drag Category to the Color shelf.
  • Rename this sheet to Area by Type.
  • Create one more new worksheet for the KPI summary.
  • Drag Number of Records to the Text shelf on the Marks card. This shows total features processed.
  • Double-click in the Rows shelf and type a calculated field: SUM([Area Sqm]). Press Enter to add total area to the view.
  • To show feature count by type, drag Category to the Rows shelf and Number of Records to the Columns shelf. Switch the mark type to Text to display as numbers.
  • Rename this sheet to KPI Summary.

What do these charts show?

The bar chart reveals the distribution of land-use features. Buildings typically dominate in urban areas like Chinatown, with roads second and parks a small fraction.

The treemap sizes each subcategory by its total area in square meters. This helps identify which building types or road classes take up the most physical space.

Assemble the dashboard and publish to Tableau Public

You have four sheets: a map, a bar chart, a treemap, and a KPI summary. Now you will combine them into a single interactive dashboard and publish it to Tableau Public for sharing.

  • Click the New Dashboard icon at the bottom (the icon with a grid and plus sign).
  • In the Size section on the left, set the dashboard size to Automatic so it adapts to different screen sizes.
  • Drag Land Use Map from the Sheets list on the left into the main dashboard area.
  • Drag Land Use Distribution to the right side of the map.
  • Drag Area by Type below the bar chart on the right.
  • Drag KPI Summary above the map or in a top banner position.
  • To add a filter for land-use type, click on the Land Use Map sheet inside the dashboard.
  • Click the small dropdown arrow on the Category color legend.
  • Select Apply to Worksheets, then All Using This Data Source. This makes the color legend act as an interactive filter across all sheets.
  • To add a geographic zone filter, click the Land Use Map sheet again.
  • Click the funnel icon in the sheet's toolbar (or go to Analysis then Filters).
  • Add Feature Category as a filter and select Show Filter.

What do these filters do?

Interactive filters let anyone exploring your dashboard isolate specific feature types. Clicking a single category on the legend highlights only those polygons on the map and updates the charts simultaneously.

  • To publish the dashboard, click File in the top menu bar.
  • Click Save to Tableau Public As....
  • Sign in with your Tableau Public account (or create one if prompted).
  • Enter a name for your workbook, such as Chinatown Land Use Analytics.
  • Click Save. Tableau uploads your workbook and opens it in your browser.
  • Copy the public URL from your browser's address bar. This is your shareable dashboard link.

Your browser opens with the published dashboard showing the interactive map, charts, and filters. Anyone with the URL can explore your geospatial analytics.

Publish button is grayed out or missing?

  • Make sure you are using Tableau Desktop Public Edition (not the Free Edition). Tableau Desktop Free Edition cannot save to Tableau Public.
  • Verify you are signed into your Tableau Public account under Server in the menu.
  • If the option says Save to Tableau Public without the "As..." part, that works too. It publishes directly.

Help me publish my Tableau workbook to Tableau Public. The save option seems unavailable.

Secret mission

Orchestrate with Apache Airflow

Your pipeline works, but you run it manually. In production, data engineers schedule and monitor pipelines with orchestration tools. Can you wrap your entire ETL in an Apache Airflow DAG running inside Docker Compose?

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 primarily on local infrastructure, so there are no ongoing cloud costs.

Resources you used:

  • PostgreSQL database (geo_warehouse) running locally with PostGIS extension.
  • Docker Compose containers (Airflow webserver, scheduler, and PostgreSQL backend) if you completed the Secret Mission.
  • Project folder on your Desktop containing pipeline code, raw GeoPackage files, and exported GeoJSON/CSV.
  • Python virtual environment with geospatial packages (osmnx, geopandas, psycopg2-binary, sqlalchemy, shapely).

Keep everything running

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

  • PostgreSQL uses minimal resources when idle. Your geo_warehouse database stays available for querying or extending with new data.
  • If you completed the Secret Mission, your Docker Compose containers remain running and accessible at localhost:8080.
  • Your Tableau Public dashboard remains live at the URL you published.

Pause - I'll come back to this later

Shut down running processes to free up memory and CPU while keeping all your files intact.

  • Stop the PostgreSQL service. Open the Windows Services app, find the service named postgresql-x64-17, right-click it, and select Stop.
  • If you completed the Secret Mission, stop the Docker Compose containers by running this command from your project folder:
docker compose stop
  • To resume later, restart the PostgreSQL service from Services and run docker compose start from your project folder.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow these steps in order.

  • If you completed the Secret Mission, remove the Docker Compose containers, volumes, and images by running this from your project folder:
docker compose down --rmi all -v
  • Drop the geo_warehouse database. Open pgAdmin, connect to your local server, open the Query Tool, and run:
DROP DATABASE geo_warehouse;
  • Uninstall PostgreSQL entirely. Open Windows Settings, select Apps, then Apps & features. Find PostgreSQL 17 in the list and click Uninstall.
  • Delete the project folder from your Desktop. On macOS or Linux:
cd ~/Desktop
rm -rf geo-etl-pipeline
  • On Windows, open File Explorer, navigate to your Desktop, right-click the project folder, and select Delete. This removes all pipeline code, raw data, exports, and the virtual environment.

Nice Work!

Nice Work!

Nice work! You just built a complete geospatial data platform from scratch. Real OpenStreetMap data flows through your pipeline, transforms through dimensional layers, lands in a PostGIS warehouse, and powers a live Tableau dashboard.

You've learned how to:

  • Build a modular Python ETL pipeline that extracts real geospatial data from OpenStreetMap and transforms it through dbt-style data layers (staging, intermediate, mart).
  • Design and populate a star schema dimensional warehouse in PostgreSQL/PostGIS with fact and dimension tables, spatial queries, and automated exports.
  • Create spatial analytics dashboards in Tableau Public consuming exported GeoJSON and CSV files for land-use visualizations.
  • Secret Mission: Orchestrated the entire pipeline with an Apache Airflow DAG running in Docker Compose.

Ready to quiz yourself?