Urban Analytics Medallion Pipeline

Introduction

30 Second Summary

Every day, New York City receives thousands of 311 complaints about everything from noise violations to broken streetlights. When you combine this data with weather patterns, you can uncover which neighborhoods are under the most stress and why.

In this project, you will build a batch data pipeline that ingests NYC 311 complaint data and NOAA weather observations from public APIs. The pipeline processes records through a medallion architecture on AWS S3 and produces a daily Neighborhood Stress Index per ZIP code.

What You'll Build

A production-grade Python pipeline that ingests heterogeneous public APIs, enforces schemas with Pydantic, and writes Hive-partitioned Parquet files to a medallion lakehouse on S3.

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

• A working medallion lakehouse (Bronze/Silver/Gold) on AWS S3 storing NYC operational data as Hive-partitioned Parquet files.
• A Python pipeline with Pydantic schema enforcement that validates and normalizes data from two heterogeneous public APIs.
• A computed Neighborhood Stress Index in the Gold layer that correlates 311 complaints with weather conditions.
• Secret Mission: Data quality gates that halt the pipeline when validation failures exceed a configurable threshold.

Are there any prerequisites?

You should have Python installed, an AWS account with the CLI configured, and familiarity with Python classes and type hints. Basic SQL knowledge is also helpful.

Before We Start

Every city generates a constant stream of operational data. In New York City, over 30,000 311 complaints and hundreds of weather observations are recorded each day. Each source lives in its own API with its own format.

In this step, you will set up a Python project that can ingest these sources, validate them through a medallion architecture (Bronze/Silver/Gold), and store them as columnar Parquet on AWS S3. First, you need a working environment with the right dependencies, an S3 bucket, and a clean project structure.

In this step, get ready to:

• Create a Python virtual environment and install dependencies.
• Verify your AWS CLI and create an S3 bucket for the lakehouse.
• Set up the project directory structure with all required modules.

Create a virtual environment and install dependencies

A virtual environment keeps your project's packages isolated from your system Python. This project requires DuckDB, Pydantic, requests, and boto3.

DuckDB requires Python 3.10 or higher. Let's confirm your version first.

• Check your Python version by running this command:

python3 --version

✔️ I see version 3.10 or higher

Your Python version is compatible. Continue to the next instruction.

ⓧ I see an older version

Your system Python is below 3.10. macOS often ships with an older version that shadows newer installs.

• Download and install Python 3.10+ from python.org.
• After installation, run python3.10 --version (or python3.11 --version) to confirm the new version is accessible.
• Use the versioned binary (python3.10 or python3.11) in place of python3 for the rest of this project.

ⓧ Command not found

Python is not installed or not on your PATH.

• Download and install Python 3.10+ from python.org.
• After installation, close and reopen your terminal, then run python3 --version again.

• Move to your Desktop by running this command:

cd ~/Desktop

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

mkdir nyc-stress-pipeline
cd nyc-stress-pipeline

• Create a virtual environment by running:

python3 -m venv .venv

• Activate the virtual environment:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

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

• Create a requirements.txt file with the project dependencies by running:

cat > requirements.txt << 'EOF'
duckdb==1.5.3
pydantic==2.13.4
requests==2.32.3
boto3==1.38.0
EOF

What are these packages?

• duckdb is an in-process SQL engine that reads and writes Parquet files directly from S3.
• pydantic enforces data contracts via Python type hints. Invalid records fail loudly instead of silently propagating bad data.
• requests handles HTTP calls to the NYC 311 and NOAA APIs.
• boto3 is the AWS SDK for Python, used for writing quality reports to S3.

• Install all dependencies by running:

pip install -r requirements.txt

• Verify the installations work by running a quick import check:

python3 -c "import duckdb; import pydantic; import requests; import boto3; print('All packages imported successfully')"

You should see All packages imported successfully printed to the terminal.

Seeing an import error?

• Make sure your virtual environment is active. You should see (.venv) in your terminal prompt.
• If pip install failed earlier, check that you have internet access and re-run the install command.
• help me debug a package import error after pip install

Verify AWS CLI and create an S3 bucket

Your pipeline stores all data layers (Bronze, Silver, Gold) in an S3 bucket. Before creating one, let's confirm your AWS CLI can authenticate.

• Verify your AWS CLI is configured by running:

aws sts get-caller-identity

✔️ I see a JSON response with my Account and Arn

Your AWS CLI is authenticated. You should see output containing UserId, Account, and Arn fields. Continue to the next instruction.

ⓧ I see an error

Your AWS CLI is not configured or your credentials have expired.

• Run aws configure and enter your Access Key ID, Secret Access Key, default region (us-east-1), and output format (json).
• Re-run aws sts get-caller-identity to confirm it works.
• help me fix AWS CLI authentication

• Create your S3 bucket for the lakehouse by running:

aws s3 mb s3://[[LAKEHOUSE_BUCKET="nyc-stress-lakehouse-yourname"]] --region us-east-1

Why does the bucket name need to be unique?

S3 bucket names are globally unique across all AWS accounts. Add your name or a few random characters to the end to avoid conflicts with other learners.

You should see make_bucket: followed by your bucket name in the output.

Seeing a BucketAlreadyExists error?

• The bucket name you chose is taken. Try a different name with more unique characters appended.
• Re-run the aws s3 mb command with the new name.
• help me fix an S3 bucket creation error

Create the project directory structure

Your pipeline is organized as a Python package with separate modules for configuration, schemas, ingestion, transformations, storage, and the pipeline runner.

• Create the src package directory and all module files by running:

mkdir src
touch src/__init__.py src/config.py src/schemas.py src/ingest.py src/transformations.py src/storage.py src/pipeline.py

• Confirm the files exist by running:

ls src/

You should see __init__.py, config.py, schemas.py, ingest.py, transformations.py, storage.py, and pipeline.py listed in the output.

What does each module do?

• config.py holds pipeline settings like the S3 bucket name, API tokens, and target ZIP codes.
• schemas.py defines the Pydantic models that enforce data types and validate every record.
• ingest.py fetches raw data from the NYC 311 and NOAA APIs.
• transformations.py validates Bronze data and computes the Gold stress index.
• storage.py handles writing and reading Parquet files on S3 via DuckDB.
• pipeline.py orchestrates all steps in sequence with idempotency and logging.

Files not showing up?

• Make sure you ran mkdir src before the touch command. The directory must exist first.
• On Windows, use New-Item instead of touch to create files.
• help me create the project directory structure

Your environment is ready. Next up, you will build the Bronze layer by ingesting raw data from two public APIs and writing it to S3 as Parquet files.

Configure the Pipeline

Your project structure is in place and all dependencies are installed. Now you need a single source of truth that tells the pipeline where to store data, which APIs to call, and how to connect everything together.

In this step, you will build a configuration module using Pydantic. This gives you a typed, validated configuration object that reads from environment variables and provides sensible defaults.

In this step, get ready to:

• Obtain a NOAA API token for weather data access.
• Build the PipelineConfig model in src/config.py.
• Set environment variables and verify the config loads correctly.

Get your NOAA API token

The pipeline pulls weather data from NOAA's Climate Data Online API. This API is free but requires a token for authentication.

• Open the NOAA CDO token request page: https://www.ncdc.noaa.gov/cdo-web/token
• Enter your email address in the field.
• Click Submit.
• Check your email for the token. NOAA sends it within a few minutes.
• Copy the token and save it somewhere safe. You will use it in the next substep.

Why do I need a NOAA token?

NOAA rate-limits anonymous requests. Your token identifies your application and allows up to 5 requests per second and 10,000 requests per day.

Build the pipeline configuration module

A centralized configuration object prevents scattered hardcoded values across your codebase. Every module will import this one config instead of defining its own constants.

• Open src/config.py in your editor and add the following code:

import os
from pydantic import BaseModel, ConfigDict


class PipelineConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    # S3 bucket and region for the lakehouse
    s3_bucket: str = os.environ.get("LAKEHOUSE_S3_BUCKET", "nyc-stress-lakehouse")
    aws_region: str = os.environ.get("AWS_REGION", "us-east-1")

    # API tokens for data sources
    noaa_api_token: str = os.environ.get("NOAA_API_TOKEN", "your-noaa-token-here")
    socrata_app_token: str = os.environ.get("SOCRATA_APP_TOKEN", "")

    # NYC ZIP codes to track
    nyc_zip_codes: list[str] = [
        "10001", "10002", "10003", "10004", "10005",
        "10006", "10007", "10008", "10009", "10010",
        "10011", "10012", "10013", "10014", "10016",
    ]

    # NOAA Central Park weather station
    noaa_station_id: str = "GHCND:USW00094728"

    # Quality gate threshold (5% failure rate)
    quality_failure_threshold: float = 0.05

    # Medallion layer prefixes on S3
    bronze_prefix: str = "bronze"
    silver_prefix: str = "silver"
    gold_prefix: str = "gold"
    quality_reports_prefix: str = "quality_reports"

What does this code do?

ConfigDict(frozen=True) makes the config object immutable after creation. No code can accidentally change a setting at runtime.

Each field uses os.environ.get() with a fallback default. This means the pipeline works out of the box for development, but you can override any value through environment variables in production.

The nyc_zip_codes list defines the 14 ZIP codes your pipeline will track. The noaa_station_id points to the Central Park weather station, which is NYC's primary observation point.

• Save src/config.py.

Seeing a syntax error?

Make sure you have the list[str] type hint syntax. This requires Python 3.10 or higher. If you are on an older version, you would need from typing import List and use List[str] instead.

help me fix a syntax error in my Pydantic config model

Set environment variables and verify the config

The config reads from environment variables so you can keep secrets out of your source code. You need to set the S3 bucket name and your NOAA token.

• In your terminal (with the virtual environment active), export the S3 bucket name by running:

export LAKEHOUSE_S3_BUCKET=nyc-stress-lakehouse

• Export your NOAA API token by running:

export NOAA_API_TOKEN=your-noaa-token-here

What do I replace here?

Replace your-noaa-token-here with the actual token NOAA emailed to you. If you named your S3 bucket something different in Step 1, replace nyc-stress-lakehouse with your bucket name.

• Verify the config loads by opening a Python shell:

python3 -c "from src.config import PipelineConfig; c = PipelineConfig(); print(c.s3_bucket, c.aws_region, c.noaa_station_id)"

• You should see your bucket name, region, and station ID printed:

nyc-stress-lakehouse us-east-1 GHCND:USW00094728

Seeing a ModuleNotFoundError?

Make sure you are running the command from the project root directory (the folder that contains the src/ folder). Also confirm src/__init__.py exists. Without it, Python cannot find the src package.

help me fix a ModuleNotFoundError when importing from src.config

Why frozen=True instead of strict=True?

The config model uses frozen=True (immutability) rather than strict=True (type coercion prevention). Environment variables are always strings, so strict mode would reject them. Frozen mode allows the initial string-to-type coercion but prevents any changes after construction.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/config.py and that your environment variables are set.

ⓧ I'd like to double check the full code

import os
from pydantic import BaseModel, ConfigDict


class PipelineConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    s3_bucket: str = os.environ.get("LAKEHOUSE_S3_BUCKET", "nyc-stress-lakehouse")
    aws_region: str = os.environ.get("AWS_REGION", "us-east-1")
    noaa_api_token: str = os.environ.get("NOAA_API_TOKEN", "your-noaa-token-here")
    socrata_app_token: str = os.environ.get("SOCRATA_APP_TOKEN", "")
    nyc_zip_codes: list[str] = [
        "10001", "10002", "10003", "10004", "10005",
        "10006", "10007", "10008", "10009", "10010",
        "10011", "10012", "10013", "10014", "10016",
    ]
    noaa_station_id: str = "GHCND:USW00094728"
    quality_failure_threshold: float = 0.05
    bronze_prefix: str = "bronze"
    silver_prefix: str = "silver"
    gold_prefix: str = "gold"
    quality_reports_prefix: str = "quality_reports"

Your pipeline now has a typed, immutable configuration that every module can import. Next up, you will use this config to ingest raw data from the NYC 311 and NOAA APIs into your Bronze layer on S3.

Build the Bronze Layer

Your environment is configured and your S3 bucket is ready to receive data. Now it's time to build the first layer of your medallion architecture: the Bronze layer.

The Bronze layer captures raw data exactly as it arrives from external sources. No transformations, no filtering. You need to pull from two completely different APIs (NYC 311 complaints and NOAA weather) and store their responses as Parquet files on S3 with consistent Hive partitioning.

In this step, get ready to:

• Build the NYC 311 complaint ingestion function.
• Build the NOAA weather observation ingestion function.
• Test both functions end-to-end and verify Bronze data in S3.

Build the NYC 311 ingestion function

NYC receives millions of 311 service requests every year. The city publishes this data through the Socrata Open Data API (SODA), which lets you query complaint records by date range.

• Create a new file called src/ingest.py and add the following code for the 311 ingestion function:

import logging
from datetime import date

import requests

logger = logging.getLogger(__name__)


def fetch_311_complaints(
    start_date: date,
    end_date: date,
    app_token: str = "",
    limit: int = 10000,
) -> list[dict]:
    # NYC 311 SODA API endpoint
    base_url = "https://data.cityofnewyork.us/resource/erm2-nwe9.json"
    params = {
        "$where": f"created_date >= '{start_date.isoformat()}T00:00:00' AND created_date < '{end_date.isoformat()}T00:00:00'",
        "$limit": str(limit),
        "$select": "unique_key,created_date,closed_date,agency,complaint_type,descriptor,incident_zip,borough,status",
    }
    headers = {}
    if app_token:
        headers["X-App-Token"] = app_token

    logger.info(f"Fetching 311 data: {start_date} to {end_date}")
    response = requests.get(base_url, params=params, headers=headers, timeout=60)
    response.raise_for_status()
    records = response.json()
    logger.info(f"Fetched {len(records)} 311 records")
    return records

What does this code do?

The fetch_311_complaints() function queries the NYC 311 dataset via the SODA API. It builds a SoQL query using $where to filter records by date range, and $select to pull only the columns you need.

The optional X-App-Token header gives you a higher rate limit. Without it, you share a pool with other anonymous users and may get throttled.

Now you need a way to write fetched data to S3 as Parquet. DuckDB can write directly to S3 using its httpfs extension, so you don't need to write files locally first.

• Create a new file called src/storage.py and add the following code for writing Parquet to S3:

import logging
from datetime import date

import duckdb

from src.config import PipelineConfig

logger = logging.getLogger(__name__)


def write_parquet_to_s3(
    records: list[dict],
    config: PipelineConfig,
    layer: str,
    source: str,
    partition_date: date,
) -> str:
    if not records:
        logger.warning(f"No records to write for {layer}/{source}/{partition_date}")
        return ""

    # Connect to DuckDB and configure S3 access
    conn = duckdb.connect()
    conn.execute("INSTALL httpfs; LOAD httpfs;")
    conn.execute(f"""
        CREATE SECRET (
            TYPE s3,
            PROVIDER credential_chain,
            REGION '{config.aws_region}'
        );
    """)

    # Load records into a staging table
    conn.execute("CREATE TABLE staging AS SELECT * FROM ?", [records])

    # Write to S3 with Hive-style date partitioning
    s3_path = f"s3://{config.s3_bucket}/{layer}/{source}/date={partition_date.isoformat()}/data.parquet"
    conn.execute(f"COPY staging TO '{s3_path}' (FORMAT parquet, OVERWRITE_OR_IGNORE true)")

    row_count = conn.execute("SELECT COUNT(*) FROM staging").fetchone()[0]
    conn.close()

    logger.info(f"Wrote {row_count} records to {s3_path}")
    return s3_path

What does this code do?

The write_parquet_to_s3() function takes a list of dictionaries (raw API responses) and writes them directly to S3 as a Parquet file. It uses DuckDB's credential_chain provider, which picks up the AWS credentials already configured on your machine.

The S3 path follows a Hive partitioning convention: date=YYYY-MM-DD in the folder name. This lets query engines like DuckDB skip irrelevant partitions when filtering by date.

The OVERWRITE_OR_IGNORE true option makes re-runs safe by overwriting existing files at the same path instead of throwing an error.

• Save both src/ingest.py and src/storage.py.

You won't see a change yet. The next substep adds the weather function, then you'll test both together.

Build the NOAA weather ingestion function

The second data source is NOAA's Climate Data Online (CDO) API v2. It provides daily temperature and precipitation observations from weather stations across the country. You'll pull data from the Central Park station (GHCND:USW00094728), which is NYC's primary observation point.

• In src/ingest.py, add the following function below fetch_311_complaints():

def fetch_noaa_weather(
    station_id: str,
    start_date: date,
    end_date: date,
    api_token: str,
) -> list[dict]:
    # NOAA Climate Data Online API v2 endpoint
    base_url = "https://www.ncei.noaa.gov/cdo-web/api/v2/data"
    params = {
        "datasetid": "GHCND",
        "stationid": station_id,
        "startdate": start_date.isoformat(),
        "enddate": end_date.isoformat(),
        "datatypeid": "TMAX,TMIN,TAVG,PRCP",
        "units": "metric",
        "limit": "1000",
    }
    headers = {"token": api_token}

    logger.info(f"Fetching NOAA weather: {start_date} to {end_date}, station={station_id}")
    response = requests.get(base_url, params=params, headers=headers, timeout=60)
    response.raise_for_status()
    data = response.json()
    results = data.get("results", [])
    logger.info(f"Fetched {len(results)} weather observations")
    return results

What does this code do?

The fetch_noaa_weather() function queries the NOAA CDO v2 API for daily weather observations. It requests four data types: maximum temperature, minimum temperature, average temperature, and precipitation.

The API requires a token header for authentication. This is the NOAA token you saved in your NOAA_API_TOKEN environment variable in Step 1.

The response wraps results in a results key, so the function extracts that list before returning.

You also need a utility function to check whether a partition already exists in S3. This is essential for idempotency: the pipeline can skip dates it has already ingested.

• In src/storage.py, add the following function below write_parquet_to_s3():

def check_partition_exists(
    config: PipelineConfig,
    layer: str,
    source: str,
    partition_date: date,
) -> bool:
    conn = duckdb.connect()
    conn.execute("INSTALL httpfs; LOAD httpfs;")
    conn.execute(f"""
        CREATE SECRET (
            TYPE s3,
            PROVIDER credential_chain,
            REGION '{config.aws_region}'
        );
    """)

    # Check if any Parquet files exist at this partition path
    s3_path = f"s3://{config.s3_bucket}/{layer}/{source}/date={partition_date.isoformat()}/"
    try:
        result = conn.execute(f"SELECT COUNT(*) FROM glob('{s3_path}*.parquet')").fetchone()
        conn.close()
        return result[0] > 0
    except Exception:
        conn.close()
        return False

What does this code do?

The check_partition_exists() function uses DuckDB's glob() to look for Parquet files at a specific partition path in S3. If files exist, it returns True.

This powers idempotent ingestion. Before fetching data from an API, the pipeline can check whether that day's data already exists and skip the request entirely.

• Save both src/ingest.py and src/storage.py.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on to testing.

ⓧ I'd like to double check the full code

Here's the complete src/ingest.py:

import logging
from datetime import date

import requests

logger = logging.getLogger(__name__)


def fetch_311_complaints(
    start_date: date,
    end_date: date,
    app_token: str = "",
    limit: int = 10000,
) -> list[dict]:
    base_url = "https://data.cityofnewyork.us/resource/erm2-nwe9.json"
    params = {
        "$where": f"created_date >= '{start_date.isoformat()}T00:00:00' AND created_date < '{end_date.isoformat()}T00:00:00'",
        "$limit": str(limit),
        "$select": "unique_key,created_date,closed_date,agency,complaint_type,descriptor,incident_zip,borough,status",
    }
    headers = {}
    if app_token:
        headers["X-App-Token"] = app_token

    logger.info(f"Fetching 311 data: {start_date} to {end_date}")
    response = requests.get(base_url, params=params, headers=headers, timeout=60)
    response.raise_for_status()
    records = response.json()
    logger.info(f"Fetched {len(records)} 311 records")
    return records


def fetch_noaa_weather(
    station_id: str,
    start_date: date,
    end_date: date,
    api_token: str,
) -> list[dict]:
    base_url = "https://www.ncei.noaa.gov/cdo-web/api/v2/data"
    params = {
        "datasetid": "GHCND",
        "stationid": station_id,
        "startdate": start_date.isoformat(),
        "enddate": end_date.isoformat(),
        "datatypeid": "TMAX,TMIN,TAVG,PRCP",
        "units": "metric",
        "limit": "1000",
    }
    headers = {"token": api_token}

    logger.info(f"Fetching NOAA weather: {start_date} to {end_date}, station={station_id}")
    response = requests.get(base_url, params=params, headers=headers, timeout=60)
    response.raise_for_status()
    data = response.json()
    results = data.get("results", [])
    logger.info(f"Fetched {len(results)} weather observations")
    return results

Here's the complete src/storage.py:

import logging
from datetime import date

import duckdb

from src.config import PipelineConfig

logger = logging.getLogger(__name__)


def write_parquet_to_s3(
    records: list[dict],
    config: PipelineConfig,
    layer: str,
    source: str,
    partition_date: date,
) -> str:
    if not records:
        logger.warning(f"No records to write for {layer}/{source}/{partition_date}")
        return ""

    conn = duckdb.connect()
    conn.execute("INSTALL httpfs; LOAD httpfs;")
    conn.execute(f"""
        CREATE SECRET (
            TYPE s3,
            PROVIDER credential_chain,
            REGION '{config.aws_region}'
        );
    """)

    conn.execute("CREATE TABLE staging AS SELECT * FROM ?", [records])

    s3_path = f"s3://{config.s3_bucket}/{layer}/{source}/date={partition_date.isoformat()}/data.parquet"
    conn.execute(f"COPY staging TO '{s3_path}' (FORMAT parquet, OVERWRITE_OR_IGNORE true)")

    row_count = conn.execute("SELECT COUNT(*) FROM staging").fetchone()[0]
    conn.close()

    logger.info(f"Wrote {row_count} records to {s3_path}")
    return s3_path


def check_partition_exists(
    config: PipelineConfig,
    layer: str,
    source: str,
    partition_date: date,
) -> bool:
    conn = duckdb.connect()
    conn.execute("INSTALL httpfs; LOAD httpfs;")
    conn.execute(f"""
        CREATE SECRET (
            TYPE s3,
            PROVIDER credential_chain,
            REGION '{config.aws_region}'
        );
    """)

    s3_path = f"s3://{config.s3_bucket}/{layer}/{source}/date={partition_date.isoformat()}/"
    try:
        result = conn.execute(f"SELECT COUNT(*) FROM glob('{s3_path}*.parquet')").fetchone()
        conn.close()
        return result[0] > 0
    except Exception:
        conn.close()
        return False

Test both ingestion functions end-to-end

Time to run both ingestion functions against the live APIs and verify that Bronze Parquet files land in your S3 bucket with the correct partitioning.

• Run the following test script from your project root to ingest data for a date 7 days ago:

# test_bronze.py - tests Bronze layer ingestion end-to-end
from datetime import date, timedelta
from src.config import PipelineConfig
from src.ingest import fetch_311_complaints, fetch_noaa_weather
from src.storage import write_parquet_to_s3

import logging
logging.basicConfig(level=logging.INFO)

config = PipelineConfig()
target_date = date.today() - timedelta(days=7)

# Fetch and write 311 complaints to Bronze
raw_311 = fetch_311_complaints(
    start_date=target_date,
    end_date=target_date + timedelta(days=1),
    app_token=config.socrata_app_token,
)
print(f"Fetched {len(raw_311)} 311 records")
path_311 = write_parquet_to_s3(raw_311, config, config.bronze_prefix, "complaints", target_date)
print(f"Written to: {path_311}")

# Fetch and write weather observations to Bronze
raw_weather = fetch_noaa_weather(
    station_id=config.noaa_station_id,
    start_date=target_date,
    end_date=target_date,
    api_token=config.noaa_api_token,
)
print(f"Fetched {len(raw_weather)} weather observations")
path_weather = write_parquet_to_s3(raw_weather, config, config.bronze_prefix, "weather", target_date)
print(f"Written to: {path_weather}")

• Save test_bronze.py in your project root, then run it:

python test_bronze.py

You should see log output showing records fetched from both APIs, followed by confirmation that Parquet files were written to S3.

Seeing a connection or authentication error?

• If you see a 401 error from the NOAA endpoint, check that your NOAA_API_TOKEN environment variable is set correctly in your terminal session.
• If you see a timeout error, the API may be temporarily slow. Try running the script again.
• If DuckDB reports an S3 access error, verify your AWS CLI is configured by running aws sts get-caller-identity.

Help me debug my Bronze layer ingestion errors.

Now verify the data is queryable by reading it back from S3 with DuckDB.

• Create and run the following verification script:

# verify_bronze.py - query Bronze data back from S3
import duckdb
from src.config import PipelineConfig

config = PipelineConfig()

conn = duckdb.connect()
conn.execute("INSTALL httpfs; LOAD httpfs;")
conn.execute(f"""
    CREATE SECRET (
        TYPE s3,
        PROVIDER credential_chain,
        REGION '{config.aws_region}'
    );
""")

# Count 311 records in Bronze
result_311 = conn.execute(f"""
    SELECT COUNT(*) FROM read_parquet(
        's3://{config.s3_bucket}/bronze/complaints/**/*.parquet',
        hive_partitioning=true
    )
""").fetchone()
print(f"311 records in Bronze: {result_311[0]}")

# Count weather records in Bronze
result_weather = conn.execute(f"""
    SELECT COUNT(*) FROM read_parquet(
        's3://{config.s3_bucket}/bronze/weather/**/*.parquet',
        hive_partitioning=true
    )
""").fetchone()
print(f"Weather records in Bronze: {result_weather[0]}")

conn.close()

• Save verify_bronze.py and run it:

python verify_bronze.py

You should see non-zero record counts for both 311 and weather data. This confirms that DuckDB can read your Hive-partitioned Parquet files directly from S3.

Why query the data back?

Writing data to S3 can succeed without the files being readable. Querying with read_parquet() and hive_partitioning=true proves three things: the Parquet format is valid, the Hive partition structure is correct, and DuckDB can resolve your S3 credentials for reads.

Seeing zero records?

• Check that test_bronze.py completed without errors first. If the API returned zero records, try a different target date (some days have very few complaints).
• Make sure the bucket name in your LAKEHOUSE_S3_BUCKET environment variable matches the bucket you created in Step 1.

Help me figure out why my Bronze data query returns zero records.

Your Bronze layer is working. Raw data from two heterogeneous APIs is now stored as queryable Parquet files on S3. Next up, you'll build the Silver layer to validate and normalize this raw data using Pydantic schema contracts.

Build the Silver Layer

Your Bronze layer holds raw API responses exactly as they arrived. The problem is that raw data is messy. ZIP codes have trailing spaces, dates arrive as strings, and some records are missing required fields entirely.

The Silver layer fixes this by running every record through a strict schema contract. Records that pass become clean, typed, and normalized. Records that fail get logged and dropped. This guarantees that anything downstream can trust the data without defensive checks.

In this step, get ready to:

• Define Pydantic schema models for 311 and weather data with strict validation.
• Build transformation functions that validate and normalize Bronze records.
• Run the Silver transformation and verify the output in S3.

Define the Silver schema models

A schema contract is a strict definition of what a valid record looks like. If a record does not match the contract exactly, it gets rejected. This catches upstream data drift immediately rather than letting bad data silently propagate.

You will use Pydantic models with strict=True to enforce types without silent coercion. A string "123" will NOT be accepted where an integer is expected.

Why Pydantic over dbt tests?

Pydantic validates at the Python level before data is written, providing immediate feedback and halting on failure. dbt tests run post-write in a warehouse context. Since this pipeline has no warehouse (DuckDB reads Parquet directly from S3), Pydantic is the correct validation layer.

• Create a new file called src/schemas.py in your project. Start with the imports and the Silver311Record model by adding the following code:

from datetime import datetime, date
from typing import Optional

from pydantic import BaseModel, ConfigDict, field_validator


class Silver311Record(BaseModel):
    model_config = ConfigDict(strict=True)

    unique_key: str
    created_date: datetime
    closed_date: Optional[datetime] = None
    agency: str
    complaint_type: str
    descriptor: Optional[str] = None
    incident_zip: str
    borough: Optional[str] = None
    status: str

    @field_validator("incident_zip")
    @classmethod
    def normalize_zip(cls, v: str) -> str:
        cleaned = v.strip()[:5]
        if not cleaned.isdigit() or len(cleaned) != 5:
            raise ValueError(f"Invalid ZIP code: {v}")
        return cleaned

What does this code do?

ConfigDict(strict=True) prevents Pydantic from silently converting types. If the API sends a number where you expect a string, validation fails immediately.

The normalize_zip field validator strips whitespace, takes only the first 5 characters, and rejects anything that is not a valid 5-digit ZIP code. This standardizes messy ZIP data like "10001-3456" or " 10001 " into a clean "10001".

• Below the Silver311Record class, add the SilverWeatherRecord model:

class SilverWeatherRecord(BaseModel):
    model_config = ConfigDict(strict=True)

    station: str
    observation_date: date
    tmax: Optional[float] = None
    tmin: Optional[float] = None
    tavg: Optional[float] = None
    prcp: Optional[float] = None

    @field_validator("observation_date", mode="before")
    @classmethod
    def parse_date(cls, v):
        if isinstance(v, str):
            return date.fromisoformat(v)
        return v

What does this code do?

The weather model uses Optional[float] for measurement fields because not every station reports every metric every day. A missing temperature is valid. A wrong type is not.

The parse_date validator uses mode="before" so it runs before strict type checking. This converts a date string from the API into a Python date object, which then passes strict validation.

• Below the SilverWeatherRecord class, add the GoldStressRecord model. You will use this in the next step for the Gold layer output:

class GoldStressRecord(BaseModel):
    model_config = ConfigDict(frozen=True)

    record_date: date
    zip_code: str
    complaint_count: int
    complaint_intensity: float
    weather_deviation: float
    stress_index: float
    computation_timestamp: datetime
    data_freshness: date
    source_311_count: int
    source_weather_count: int

What does this code do?

GoldStressRecord uses ConfigDict(frozen=True) instead of strict=True. This makes instances immutable after creation. Gold records are computed outputs that should never be modified after the fact.

The metadata fields (computation_timestamp, data_freshness, source counts) let you audit exactly when and from what data each stress score was derived.

• Save src/schemas.py.

Seeing an import error?

Make sure pydantic is installed in your virtual environment. Run pip list | grep pydantic to confirm. If missing, run pip install pydantic==2.13.4.

help me fix my Pydantic import error

✔️ Awesome, I've got everything!

Great. Double check you have saved src/schemas.py with all three models.

ⓧ I'd like to double check the full code

from datetime import datetime, date
from typing import Optional

from pydantic import BaseModel, ConfigDict, field_validator


class Silver311Record(BaseModel):
    model_config = ConfigDict(strict=True)

    unique_key: str
    created_date: datetime
    closed_date: Optional[datetime] = None
    agency: str
    complaint_type: str
    descriptor: Optional[str] = None
    incident_zip: str
    borough: Optional[str] = None
    status: str

    @field_validator("incident_zip")
    @classmethod
    def normalize_zip(cls, v: str) -> str:
        cleaned = v.strip()[:5]
        if not cleaned.isdigit() or len(cleaned) != 5:
            raise ValueError(f"Invalid ZIP code: {v}")
        return cleaned


class SilverWeatherRecord(BaseModel):
    model_config = ConfigDict(strict=True)

    station: str
    observation_date: date
    tmax: Optional[float] = None
    tmin: Optional[float] = None
    tavg: Optional[float] = None
    prcp: Optional[float] = None

    @field_validator("observation_date", mode="before")
    @classmethod
    def parse_date(cls, v):
        if isinstance(v, str):
            return date.fromisoformat(v)
        return v


class GoldStressRecord(BaseModel):
    model_config = ConfigDict(frozen=True)

    record_date: date
    zip_code: str
    complaint_count: int
    complaint_intensity: float
    weather_deviation: float
    stress_index: float
    computation_timestamp: datetime
    data_freshness: date
    source_311_count: int
    source_weather_count: int

Build the Silver transformation functions

With the schemas defined, you need functions that take raw Bronze records, attempt to validate each one, and separate them into two lists: valid records that pass the schema contract and invalid records that get logged and dropped.

• Create a new file called src/transformations.py. Add the imports and the validate_311_records function:

import logging
from datetime import date, datetime
from typing import Any

import duckdb
from pydantic import ValidationError

from src.config import PipelineConfig
from src.schemas import Silver311Record, SilverWeatherRecord, GoldStressRecord

logger = logging.getLogger(__name__)


def validate_311_records(
    raw_records: list[dict],
) -> tuple[list[dict], list[dict]]:
    valid = []
    invalid = []
    for record in raw_records:
        try:
            validated = Silver311Record.model_validate(record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"record": record, "errors": e.errors()})
    logger.info(f"311 validation: {len(valid)} valid, {len(invalid)} invalid")
    return valid, invalid

What does this code do?

model_validate() attempts to parse each raw dictionary through the Silver311Record schema. If validation succeeds, model_dump() converts the validated model back into a dictionary with clean, normalized values.

If validation fails, the function catches the ValidationError and stores both the bad record and its error details. This lets you diagnose problems later without losing the data entirely.

• Below validate_311_records, add the weather validation functions. The NOAA API returns one observation per row per metric (so TMAX, TMIN, and PRCP for the same day arrive as three separate records). The _pivot_weather_observations helper groups them into one record per station per day before validation:

def validate_weather_records(
    raw_observations: list[dict],
) -> tuple[list[dict], list[dict]]:
    pivot = _pivot_weather_observations(raw_observations)
    valid = []
    invalid = []
    for record in pivot:
        try:
            validated = SilverWeatherRecord.model_validate(record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"record": record, "errors": e.errors()})
    logger.info(f"Weather validation: {len(valid)} valid, {len(invalid)} invalid")
    return valid, invalid


def _pivot_weather_observations(observations: list[dict]) -> list[dict]:
    by_date: dict[str, dict[str, Any]] = {}
    for obs in observations:
        obs_date = obs.get("date", "")[:10]
        station = obs.get("station", "")
        key = f"{station}_{obs_date}"
        if key not in by_date:
            by_date[key] = {"station": station, "observation_date": obs_date}
        datatype = obs.get("datatype", "")
        value = obs.get("value")
        if datatype == "TMAX":
            by_date[key]["tmax"] = value
        elif datatype == "TMIN":
            by_date[key]["tmin"] = value
        elif datatype == "TAVG":
            by_date[key]["tavg"] = value
        elif datatype == "PRCP":
            by_date[key]["prcp"] = value
    return list(by_date.values())

What does this code do?

_pivot_weather_observations reshapes the flat NOAA response into one dictionary per station per day, with separate keys for each metric (tmax, tmin, tavg, prcp). Without this pivot, each raw observation would fail SilverWeatherRecord validation because it only has one metric field populated.

validate_weather_records then runs each pivoted record through the Pydantic model, applying the same valid/invalid split pattern as the 311 validator.

• Save src/transformations.py.

Seeing a circular import error?

Check that src/schemas.py does not import anything from src/transformations.py. The dependency goes one way: transformations imports schemas, never the reverse.

help me debug my circular import error in my pipeline project

✔️ Awesome, I've got everything!

Great. Make sure you have saved src/transformations.py with both validation functions and the pivot helper.

ⓧ I'd like to double check the full code

import logging
from datetime import date, datetime
from typing import Any

import duckdb
from pydantic import ValidationError

from src.config import PipelineConfig
from src.schemas import Silver311Record, SilverWeatherRecord, GoldStressRecord

logger = logging.getLogger(__name__)


def validate_311_records(
    raw_records: list[dict],
) -> tuple[list[dict], list[dict]]:
    valid = []
    invalid = []
    for record in raw_records:
        try:
            validated = Silver311Record.model_validate(record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"record": record, "errors": e.errors()})
    logger.info(f"311 validation: {len(valid)} valid, {len(invalid)} invalid")
    return valid, invalid


def validate_weather_records(
    raw_observations: list[dict],
) -> tuple[list[dict], list[dict]]:
    pivot = _pivot_weather_observations(raw_observations)
    valid = []
    invalid = []
    for record in pivot:
        try:
            validated = SilverWeatherRecord.model_validate(record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"record": record, "errors": e.errors()})
    logger.info(f"Weather validation: {len(valid)} valid, {len(invalid)} invalid")
    return valid, invalid


def _pivot_weather_observations(observations: list[dict]) -> list[dict]:
    by_date: dict[str, dict[str, Any]] = {}
    for obs in observations:
        obs_date = obs.get("date", "")[:10]
        station = obs.get("station", "")
        key = f"{station}_{obs_date}"
        if key not in by_date:
            by_date[key] = {"station": station, "observation_date": obs_date}
        datatype = obs.get("datatype", "")
        value = obs.get("value")
        if datatype == "TMAX":
            by_date[key]["tmax"] = value
        elif datatype == "TMIN":
            by_date[key]["tmin"] = value
        elif datatype == "TAVG":
            by_date[key]["tavg"] = value
        elif datatype == "PRCP":
            by_date[key]["prcp"] = value
    return list(by_date.values())

Run the Silver transformation and verify output

Time to prove the Silver layer works end-to-end. You will fetch fresh data from both APIs, run it through the validation functions, write the results to S3, and then query the Silver Parquet back with DuckDB to confirm schema enforcement.

• Run the Silver transformation by executing this script in your terminal:

python -c "
from datetime import date, timedelta
from src.config import PipelineConfig
from src.ingest import fetch_311_complaints, fetch_noaa_weather
from src.transformations import validate_311_records, validate_weather_records
from src.storage import write_parquet_to_s3

config = PipelineConfig()
target_date = date.today() - timedelta(days=7)

# Fetch and validate 311 data
raw_311 = fetch_311_complaints(target_date, target_date + timedelta(days=1), config.socrata_app_token)
valid_311, invalid_311 = validate_311_records(raw_311)
print(f'311: {len(valid_311)} valid, {len(invalid_311)} invalid')
write_parquet_to_s3(valid_311, config, config.silver_prefix, 'complaints', target_date)

# Fetch and validate weather data
raw_weather = fetch_noaa_weather(config.noaa_station_id, target_date, target_date, config.noaa_api_token)
valid_weather, invalid_weather = validate_weather_records(raw_weather)
print(f'Weather: {len(valid_weather)} valid, {len(invalid_weather)} invalid')
write_parquet_to_s3(valid_weather, config, config.silver_prefix, 'weather', target_date)

print('Silver layer written successfully.')
"

What does this script do?

This script fetches data from both APIs for a date 7 days ago, validates each dataset through the Pydantic models, and writes the valid records to S3 under the silver/ prefix. You should see counts printed for valid and invalid records from each source.

• You should see output similar to:

311: 4523 valid, 12 invalid
Weather: 1 valid, 0 invalid
Silver layer written successfully.

The exact counts depend on the day's data volume. The key confirmation is that some records passed (valid > 0) and the script completed without errors.

• Now query the Silver layer with DuckDB to confirm schema types are correct and partitioning works. Run this in your terminal:

python -c "
import duckdb
from src.config import PipelineConfig

config = PipelineConfig()
conn = duckdb.connect()
conn.execute('INSTALL httpfs; LOAD httpfs;')
conn.execute(f\"\"\"CREATE SECRET (TYPE s3, PROVIDER credential_chain, REGION '{config.aws_region}');\"\"\")

# Check Silver 311 schema and count
print('=== Silver 311 Complaints ===')
result = conn.execute(f\"SELECT COUNT(*) FROM read_parquet('s3://{config.s3_bucket}/silver/complaints/**/*.parquet', hive_partitioning=true)\").fetchone()
print(f'Total records: {result[0]}')

schema = conn.execute(f\"DESCRIBE SELECT * FROM read_parquet('s3://{config.s3_bucket}/silver/complaints/**/*.parquet', hive_partitioning=true)\").fetchall()
print('Schema:')
for col in schema:
    print(f'  {col[0]}: {col[1]}')

# Check a sample ZIP code
sample = conn.execute(f\"SELECT incident_zip, COUNT(*) as cnt FROM read_parquet('s3://{config.s3_bucket}/silver/complaints/**/*.parquet', hive_partitioning=true) GROUP BY incident_zip ORDER BY cnt DESC LIMIT 5\").fetchall()
print(f'Top 5 ZIP codes: {sample}')

conn.close()
print('\nSilver layer verified successfully.')
"

What should I see?

You should see a record count matching approximately the valid count from the previous step. The schema output confirms each column has the correct type (VARCHAR for strings, TIMESTAMP for dates). The ZIP codes should all be clean 5-digit strings with no trailing characters.

• Confirm the Silver Parquet files exist in S3 by running:

aws s3 ls s3://nyc-stress-lakehouse/silver/ --recursive

• You should see files at paths like silver/complaints/date=YYYY-MM-DD/data.parquet and silver/weather/date=YYYY-MM-DD/data.parquet.

Getting zero records or an API error?

If you see 0 valid 311 records, the target date may have no data yet. Try changing timedelta(days=7) to timedelta(days=14) to target an older date with confirmed data availability.

If the NOAA request fails with a 401 error, verify your NOAA_API_TOKEN environment variable is set correctly. NOAA requires a valid token for API access.

help me debug why my Silver layer transformation is failing

Your Silver layer is live. Every record in it has passed a strict schema contract and geographic normalization. Next up, you will join these two validated datasets to compute the Neighborhood Stress Index in the Gold layer.

Build the Gold Layer

Your Silver layer now holds validated, schema-enforced records for both 311 complaints and weather observations. The final step in the medallion architecture is the Gold layer, where you join multiple sources together and compute business-level metrics.

In this step, you will build a Neighborhood Stress Index that correlates complaint volume with weather conditions per ZIP code per day. This is the kind of derived analytical output that data teams build on top of clean, validated data.

In this step, get ready to:

• Build a Gold transformation function that joins Silver data and computes a stress index.
• Write Gold output to S3 as Hive-partitioned Parquet.
• Validate Gold layer results with DuckDB queries.

Add the S3 connection helper

Both the existing write_parquet_to_s3 function and the new Gold transformation need to configure DuckDB to read from S3. Instead of repeating that setup code, you will extract it into a reusable helper.

• In src/transformations.py, scroll to the bottom of the file (below the _pivot_weather_observations function) and add this helper function:

def _setup_s3(conn: duckdb.DuckDBPyConnection, config: PipelineConfig) -> None:
    conn.execute("INSTALL httpfs; LOAD httpfs;")
    conn.execute(f"""
        CREATE SECRET (
            TYPE s3,
            PROVIDER credential_chain,
            REGION '{config.aws_region}'
        );
    """)

What does this code do?

This helper installs and loads the httpfs extension, which lets DuckDB read and write files on S3.

The CREATE SECRET statement tells DuckDB to use your existing AWS credentials (from the CLI configuration) rather than hardcoding keys. The credential_chain provider automatically picks up credentials from environment variables, config files, or IAM roles.

• Save src/transformations.py.

You won't see a change yet. The next chunk uses this helper to query Silver data.

Build the Gold transformation function

The stress index formula combines two signals: how many complaints a ZIP code received relative to its recent average (complaint intensity), and how far the temperature deviated from a seasonal baseline (weather deviation). Intensity is weighted at 60% and deviation at 40%, producing a score capped at 100.

• In src/transformations.py, add the compute_stress_index function above _setup_s3 (so that _setup_s3 remains the last function in the file):

def compute_stress_index(
    config: PipelineConfig,
    target_date: date,
) -> list[dict]:
    conn = duckdb.connect()
    _setup_s3(conn, config)

    silver_311_path = f"s3://{config.s3_bucket}/{config.silver_prefix}/complaints/"
    silver_weather_path = f"s3://{config.s3_bucket}/{config.silver_prefix}/weather/"

    # SQL query joins complaints with weather and computes raw metrics
    query = f"""
    WITH daily_complaints AS (
        SELECT
            incident_zip AS zip_code,
            CAST(created_date AS DATE) AS record_date,
            COUNT(*) AS complaint_count
        FROM read_parquet('{silver_311_path}**/*.parquet', hive_partitioning=true)
        WHERE CAST(created_date AS DATE) = '{target_date.isoformat()}'
        GROUP BY incident_zip, CAST(created_date AS DATE)
    ),
    rolling_avg AS (
        SELECT
            incident_zip AS zip_code,
            AVG(daily_count) AS avg_30d
        FROM (
            SELECT
                incident_zip,
                CAST(created_date AS DATE) AS d,
                COUNT(*) AS daily_count
            FROM read_parquet('{silver_311_path}**/*.parquet', hive_partitioning=true)
            WHERE CAST(created_date AS DATE) BETWEEN '{target_date.isoformat()}'::DATE - INTERVAL 30 DAY AND '{target_date.isoformat()}'::DATE - INTERVAL 1 DAY
            GROUP BY incident_zip, CAST(created_date AS DATE)
        )
        GROUP BY incident_zip
    ),
    weather AS (
        SELECT
            observation_date,
            COALESCE(tavg, (tmax + tmin) / 2) AS temp_actual
        FROM read_parquet('{silver_weather_path}**/*.parquet', hive_partitioning=true)
        WHERE observation_date = '{target_date.isoformat()}'
    )
    SELECT
        dc.record_date,
        dc.zip_code,
        dc.complaint_count,
        CASE WHEN ra.avg_30d > 0 THEN dc.complaint_count / ra.avg_30d ELSE 1.0 END AS complaint_intensity,
        COALESCE(ABS(w.temp_actual - 15.0), 0) AS weather_deviation,
        dc.complaint_count AS source_311_count
    FROM daily_complaints dc
    LEFT JOIN rolling_avg ra ON dc.zip_code = ra.zip_code
    LEFT JOIN weather w ON dc.record_date = w.observation_date
    """

    results = conn.execute(query).fetchall()
    columns = ["record_date", "zip_code", "complaint_count", "complaint_intensity", "weather_deviation", "source_311_count"]

    # Convert raw SQL results into validated GoldStressRecord dicts
    gold_records = []
    for row in results:
        record_dict = dict(zip(columns, row))
        intensity = min(record_dict["complaint_intensity"] * 30, 60.0)
        deviation = min(record_dict["weather_deviation"] * 2, 40.0)
        stress = round(intensity + deviation, 2)

        gold_record = GoldStressRecord(
            record_date=record_dict["record_date"],
            zip_code=record_dict["zip_code"],
            complaint_count=record_dict["complaint_count"],
            complaint_intensity=round(record_dict["complaint_intensity"], 4),
            weather_deviation=round(record_dict["weather_deviation"], 2),
            stress_index=min(stress, 100.0),
            computation_timestamp=datetime.utcnow(),
            data_freshness=target_date,
            source_311_count=record_dict["source_311_count"],
            source_weather_count=1,
        )
        gold_records.append(gold_record.model_dump())

    conn.close()
    logger.info(f"Computed {len(gold_records)} Gold stress records for {target_date}")
    return gold_records

What does this code do?

This function runs a single SQL query in DuckDB that reads directly from your Silver Parquet files on S3. The query has three CTEs (Common Table Expressions):

• daily_complaints counts complaints per ZIP code for the target date.
• rolling_avg computes each ZIP's average daily complaint count over the previous 30 days. This is the baseline for measuring whether today is unusually busy.
• weather pulls the actual temperature for the target date, using tavg or falling back to the average of tmax and tmin.

The Python loop then applies the weighting formula: complaint intensity (ratio to 30-day average) contributes up to 60 points, and weather deviation (absolute departure from 15.0 degrees Celsius seasonal normal) contributes up to 40 points. Each record is validated through the GoldStressRecord Pydantic model before being returned.

• Save src/transformations.py.

You won't see output yet. The next section runs this function and writes the results to S3.

Seeing a syntax error?

Make sure the triple-quoted SQL string (f"""...""") is properly closed. Check that you have matching triple quotes at both the start and end of the query variable.

Also verify that compute_stress_index is placed above _setup_s3 in the file, since it calls that helper.

Help me debug a syntax error in my compute_stress_index function.

Write Gold output to S3 and validate results

Now you will run the Gold transformation, write its output to S3, and verify the results make sense. This confirms the join logic works and the stress scores fall within expected bounds.

• Run the Gold computation and write the output to S3 by running this script in your terminal:

python -c "
from datetime import date, timedelta
from src.config import PipelineConfig
from src.transformations import compute_stress_index
from src.storage import write_parquet_to_s3

config = PipelineConfig()
target_date = date.today() - timedelta(days=7)

print(f'Computing Gold stress index for {target_date}...')
gold_records = compute_stress_index(config, target_date)
print(f'Got {len(gold_records)} Gold records')

if gold_records:
    path = write_parquet_to_s3(gold_records, config, config.gold_prefix, 'stress_index', target_date)
    print(f'Gold data written to: {path}')
else:
    print('No Gold records produced - check that Silver data exists for this date')
"

What does this script do?

This script calls your new compute_stress_index function with a date 7 days ago (the same date you ingested data for in earlier steps). It then writes the resulting records to the Gold layer on S3 using the write_parquet_to_s3 function you built in Step 2.

The output path will be gold/stress_index/date=YYYY-MM-DD/data.parquet in your S3 bucket.

• You should see output showing the number of Gold records computed and the S3 path they were written to.
• Validate the Gold layer by running this query script:

python -c "
import duckdb
from src.config import PipelineConfig

config = PipelineConfig()
conn = duckdb.connect()
conn.execute('INSTALL httpfs; LOAD httpfs;')
conn.execute(f\"\"\"CREATE SECRET (TYPE s3, PROVIDER credential_chain, REGION '{config.aws_region}');\"\"\")

gold_path = f's3://{config.s3_bucket}/{config.gold_prefix}/stress_index/'

# Check stress scores are within 0-100
print('=== Stress Score Bounds ===')
result = conn.execute(f\"\"\"SELECT MIN(stress_index), MAX(stress_index), COUNT(*) FROM read_parquet('{gold_path}**/*.parquet', hive_partitioning=true)\"\"\").fetchone()
print(f'Min: {result[0]}, Max: {result[1]}, Total records: {result[2]}')

# Check ZIP code coverage
print('\\n=== ZIP Code Coverage ===')
zips = conn.execute(f\"\"\"SELECT zip_code, complaint_count, stress_index FROM read_parquet('{gold_path}**/*.parquet', hive_partitioning=true) ORDER BY stress_index DESC LIMIT 10\"\"\").fetchall()
for z in zips:
    print(f'  ZIP {z[0]}: {z[1]} complaints, stress={z[2]}')

# Check for nulls
print('\\n=== Null Check ===')
nulls = conn.execute(f\"\"\"SELECT COUNT(*) FILTER (WHERE stress_index IS NULL), COUNT(*) FILTER (WHERE zip_code IS NULL), COUNT(*) FILTER (WHERE complaint_count IS NULL) FROM read_parquet('{gold_path}**/*.parquet', hive_partitioning=true)\"\"\").fetchone()
print(f'Null stress_index: {nulls[0]}, Null zip_code: {nulls[1]}, Null complaint_count: {nulls[2]}')

conn.close()
print('\\nGold layer validation complete.')
"

What should I see?

The validation script checks three things:

• Stress Score Bounds: Min should be >= 0 and Max should be <= 100. If Max exceeds 100, the capping logic is broken.
• ZIP Code Coverage: You should see multiple ZIP codes listed with their complaint counts and stress scores. Higher scores indicate more complaints relative to the average or more extreme weather.
• Null Check: All three null counts should be 0. Nulls here would indicate a broken join or missing data in the Silver layer.

No Gold records produced?

If you see 0 records, verify that your Silver layer has data for the target date. Run aws s3 ls s3://nyc-stress-lakehouse/silver/complaints/ to check which date partitions exist, and make sure the date in the script matches one of those partitions.

If you ingested data for a different date in Step 3, adjust timedelta(days=7) in the script to match.

Help me debug why my Gold layer has no records.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/transformations.py with both the compute_stress_index function and the _setup_s3 helper.

ⓧ I'd like to double check the full code

Here is the complete src/transformations.py file at this point:

import logging
from datetime import date, datetime
from typing import Any

import duckdb
from pydantic import ValidationError

from src.config import PipelineConfig
from src.schemas import Silver311Record, SilverWeatherRecord, GoldStressRecord

logger = logging.getLogger(__name__)


def validate_311_records(
    raw_records: list[dict],
) -> tuple[list[dict], list[dict]]:
    valid = []
    invalid = []
    for record in raw_records:
        try:
            validated = Silver311Record.model_validate(record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"record": record, "errors": e.errors()})
    logger.info(f"311 validation: {len(valid)} valid, {len(invalid)} invalid")
    return valid, invalid


def validate_weather_records(
    raw_observations: list[dict],
) -> tuple[list[dict], list[dict]]:
    pivot = _pivot_weather_observations(raw_observations)
    valid = []
    invalid = []
    for record in pivot:
        try:
            validated = SilverWeatherRecord.model_validate(record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"record": record, "errors": e.errors()})
    logger.info(f"Weather validation: {len(valid)} valid, {len(invalid)} invalid")
    return valid, invalid


def _pivot_weather_observations(observations: list[dict]) -> list[dict]:
    by_date: dict[str, dict[str, Any]] = {}
    for obs in observations:
        obs_date = obs.get("date", "")[:10]
        station = obs.get("station", "")
        key = f"{station}_{obs_date}"
        if key not in by_date:
            by_date[key] = {"station": station, "observation_date": obs_date}
        datatype = obs.get("datatype", "")
        value = obs.get("value")
        if datatype == "TMAX":
            by_date[key]["tmax"] = value
        elif datatype == "TMIN":
            by_date[key]["tmin"] = value
        elif datatype == "TAVG":
            by_date[key]["tavg"] = value
        elif datatype == "PRCP":
            by_date[key]["prcp"] = value
    return list(by_date.values())


def compute_stress_index(
    config: PipelineConfig,
    target_date: date,
) -> list[dict]:
    conn = duckdb.connect()
    _setup_s3(conn, config)

    silver_311_path = f"s3://{config.s3_bucket}/{config.silver_prefix}/complaints/"
    silver_weather_path = f"s3://{config.s3_bucket}/{config.silver_prefix}/weather/"

    query = f"""
    WITH daily_complaints AS (
        SELECT
            incident_zip AS zip_code,
            CAST(created_date AS DATE) AS record_date,
            COUNT(*) AS complaint_count
        FROM read_parquet('{silver_311_path}**/*.parquet', hive_partitioning=true)
        WHERE CAST(created_date AS DATE) = '{target_date.isoformat()}'
        GROUP BY incident_zip, CAST(created_date AS DATE)
    ),
    rolling_avg AS (
        SELECT
            incident_zip AS zip_code,
            AVG(daily_count) AS avg_30d
        FROM (
            SELECT
                incident_zip,
                CAST(created_date AS DATE) AS d,
                COUNT(*) AS daily_count
            FROM read_parquet('{silver_311_path}**/*.parquet', hive_partitioning=true)
            WHERE CAST(created_date AS DATE) BETWEEN '{target_date.isoformat()}'::DATE - INTERVAL 30 DAY AND '{target_date.isoformat()}'::DATE - INTERVAL 1 DAY
            GROUP BY incident_zip, CAST(created_date AS DATE)
        )
        GROUP BY incident_zip
    ),
    weather AS (
        SELECT
            observation_date,
            COALESCE(tavg, (tmax + tmin) / 2) AS temp_actual
        FROM read_parquet('{silver_weather_path}**/*.parquet', hive_partitioning=true)
        WHERE observation_date = '{target_date.isoformat()}'
    )
    SELECT
        dc.record_date,
        dc.zip_code,
        dc.complaint_count,
        CASE WHEN ra.avg_30d > 0 THEN dc.complaint_count / ra.avg_30d ELSE 1.0 END AS complaint_intensity,
        COALESCE(ABS(w.temp_actual - 15.0), 0) AS weather_deviation,
        dc.complaint_count AS source_311_count
    FROM daily_complaints dc
    LEFT JOIN rolling_avg ra ON dc.zip_code = ra.zip_code
    LEFT JOIN weather w ON dc.record_date = w.observation_date
    """

    results = conn.execute(query).fetchall()
    columns = ["record_date", "zip_code", "complaint_count", "complaint_intensity", "weather_deviation", "source_311_count"]

    gold_records = []
    for row in results:
        record_dict = dict(zip(columns, row))
        intensity = min(record_dict["complaint_intensity"] * 30, 60.0)
        deviation = min(record_dict["weather_deviation"] * 2, 40.0)
        stress = round(intensity + deviation, 2)

        gold_record = GoldStressRecord(
            record_date=record_dict["record_date"],
            zip_code=record_dict["zip_code"],
            complaint_count=record_dict["complaint_count"],
            complaint_intensity=round(record_dict["complaint_intensity"], 4),
            weather_deviation=round(record_dict["weather_deviation"], 2),
            stress_index=min(stress, 100.0),
            computation_timestamp=datetime.utcnow(),
            data_freshness=target_date,
            source_311_count=record_dict["source_311_count"],
            source_weather_count=1,
        )
        gold_records.append(gold_record.model_dump())

    conn.close()
    logger.info(f"Computed {len(gold_records)} Gold stress records for {target_date}")
    return gold_records


def _setup_s3(conn: duckdb.DuckDBPyConnection, config: PipelineConfig) -> None:
    conn.execute("INSTALL httpfs; LOAD httpfs;")
    conn.execute(f"""
        CREATE SECRET (
            TYPE s3,
            PROVIDER credential_chain,
            REGION '{config.aws_region}'
        );
    """)

Your Gold layer is live. You now have a complete medallion architecture on S3: raw data in Bronze, validated data in Silver, and computed analytics in Gold. Next up, you will wire everything together into a single pipeline runner with idempotency and structured logging.

Build the Pipeline Runner

You have all the building blocks: ingestion, validation, storage, and transformation modules. Each one works independently, but a production pipeline needs a single orchestrator that calls them in the right order and handles failures gracefully.

In this step, you will build a pipeline runner that sequences Bronze, Silver, and Gold processing with two critical properties. First, idempotency: the runner checks which dates already exist on S3 before fetching, so re-runs never produce duplicate data. Second, structured logging: every stage emits JSON log records to both the console and a file for full observability.

In this step, get ready to:

• Set up structured JSON logging for the pipeline.
• Build the orchestration logic with date-based idempotency.
• Run the full pipeline end-to-end and verify idempotency by re-running.

Set up structured logging for the pipeline

Production data pipelines need observable logging. Without it, you cannot tell which stage failed, how many records were processed, or how long the run took.

• Create a new file called src/pipeline.py in your src/ directory.
• Add the imports and structured logging setup by pasting this code:

import json
import logging
import sys
from datetime import date, timedelta, datetime

from src.config import PipelineConfig
from src.ingest import fetch_311_complaints, fetch_noaa_weather
from src.transformations import validate_311_records, validate_weather_records, compute_stress_index
from src.storage import write_parquet_to_s3, check_partition_exists


def setup_logging() -> logging.Logger:
    logger = logging.getLogger("pipeline")
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        json.dumps({
            "timestamp": "%(asctime)s",
            "level": "%(levelname)s",
            "module": "%(module)s",
            "message": "%(message)s",
        })
    )

    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    file_handler = logging.FileHandler("pipeline.log")
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    return logger

What does this code do?

The imports pull in every module you built in prior steps. The runner will call them in sequence.

The setup_logging() function creates a logger that formats every message as a JSON string with timestamp, level, module, and message fields. It attaches two handlers: one writes to the console (stdout) and one writes to a local pipeline.log file. This gives you both real-time visibility and a persistent record.

• Save src/pipeline.py.

Build the orchestration logic with idempotency

The runner needs to be safe to re-run. If a pipeline crashes mid-way, you should be able to restart it without duplicating data that was already written. The pattern here is watermark-based idempotency: before fetching, check whether that date's partition already exists on S3.

• In src/pipeline.py, add the first half of the run_pipeline function below the setup_logging() function:

def run_pipeline(target_date: date, config: PipelineConfig | None = None) -> None:
    logger = setup_logging()
    if config is None:
        config = PipelineConfig()

    logger.info(f"Pipeline started for date: {target_date}")
    start_time = datetime.utcnow()

    # Step 1: Bronze ingestion (idempotent)
    if not check_partition_exists(config, config.bronze_prefix, "complaints", target_date):
        raw_311 = fetch_311_complaints(
            start_date=target_date,
            end_date=target_date + timedelta(days=1),
            app_token=config.socrata_app_token,
        )
        write_parquet_to_s3(raw_311, config, config.bronze_prefix, "complaints", target_date)
    else:
        logger.info(f"Bronze/complaints/{target_date} already exists, skipping")
        raw_311 = []

    if not check_partition_exists(config, config.bronze_prefix, "weather", target_date):
        raw_weather = fetch_noaa_weather(
            station_id=config.noaa_station_id,
            start_date=target_date,
            end_date=target_date,
            api_token=config.noaa_api_token,
        )
        write_parquet_to_s3(raw_weather, config, config.bronze_prefix, "weather", target_date)
    else:
        logger.info(f"Bronze/weather/{target_date} already exists, skipping")
        raw_weather = []

What does this code do?

The function accepts a target_date and an optional config. It calls check_partition_exists() before each Bronze fetch. If the partition is already on S3, it skips the API call and sets the raw list to empty. This is the idempotency guarantee.

When data does not yet exist, it fetches from the API and writes immediately to the Bronze layer. The else branch logs a clear skip message so you can confirm idempotency is working.

• Now add the Silver validation, Gold computation, and entry point directly below the Bronze section (still inside run_pipeline):

    # Step 2: Silver validation
    if raw_311:
        valid_311, invalid_311 = validate_311_records(raw_311)
        failure_rate = len(invalid_311) / max(len(raw_311), 1)
        if failure_rate > config.quality_failure_threshold:
            logger.error(f"Quality gate FAILED: 311 failure rate {failure_rate:.2%} exceeds threshold {config.quality_failure_threshold:.0%}")
            return
        write_parquet_to_s3(valid_311, config, config.silver_prefix, "complaints", target_date)

    if raw_weather:
        valid_weather, invalid_weather = validate_weather_records(raw_weather)
        write_parquet_to_s3(valid_weather, config, config.silver_prefix, "weather", target_date)

    # Step 3: Gold computation
    gold_records = compute_stress_index(config, target_date)
    if gold_records:
        write_parquet_to_s3(gold_records, config, config.gold_prefix, "stress_index", target_date)

    elapsed = (datetime.utcnow() - start_time).total_seconds()
    logger.info(f"Pipeline completed in {elapsed:.1f}s for date: {target_date}")


if __name__ == "__main__":
    target = date.today() - timedelta(days=7)
    run_pipeline(target)

What does this code do?

The Silver section validates records through Pydantic models. If the 311 failure rate exceeds quality_failure_threshold (5% by default), the pipeline halts immediately and logs an error. This prevents bad data from reaching the Gold layer.

The Gold section calls compute_stress_index() and writes the output to S3. Finally, the elapsed time is logged. The if __name__ block sets a default target of 7 days ago so you can run the script directly.

• Save src/pipeline.py.

Why idempotency matters

In production, pipelines crash for many reasons: network timeouts, API rate limits, disk full. Without idempotency, restarting the pipeline would ingest duplicate records into Bronze, which would cascade through Silver and Gold as inflated counts.

By checking S3 partitions before fetching, the runner skips any date that already has data. This makes it safe to re-run at any time without worrying about duplication.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/pipeline.py before continuing.

ⓧ I'd like to double check the full code

import json
import logging
import sys
from datetime import date, timedelta, datetime

from src.config import PipelineConfig
from src.ingest import fetch_311_complaints, fetch_noaa_weather
from src.transformations import validate_311_records, validate_weather_records, compute_stress_index
from src.storage import write_parquet_to_s3, check_partition_exists


def setup_logging() -> logging.Logger:
    logger = logging.getLogger("pipeline")
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        json.dumps({
            "timestamp": "%(asctime)s",
            "level": "%(levelname)s",
            "module": "%(module)s",
            "message": "%(message)s",
        })
    )

    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    file_handler = logging.FileHandler("pipeline.log")
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    return logger


def run_pipeline(target_date: date, config: PipelineConfig | None = None) -> None:
    logger = setup_logging()
    if config is None:
        config = PipelineConfig()

    logger.info(f"Pipeline started for date: {target_date}")
    start_time = datetime.utcnow()

    # Step 1: Bronze ingestion (idempotent)
    if not check_partition_exists(config, config.bronze_prefix, "complaints", target_date):
        raw_311 = fetch_311_complaints(
            start_date=target_date,
            end_date=target_date + timedelta(days=1),
            app_token=config.socrata_app_token,
        )
        write_parquet_to_s3(raw_311, config, config.bronze_prefix, "complaints", target_date)
    else:
        logger.info(f"Bronze/complaints/{target_date} already exists, skipping")
        raw_311 = []

    if not check_partition_exists(config, config.bronze_prefix, "weather", target_date):
        raw_weather = fetch_noaa_weather(
            station_id=config.noaa_station_id,
            start_date=target_date,
            end_date=target_date,
            api_token=config.noaa_api_token,
        )
        write_parquet_to_s3(raw_weather, config, config.bronze_prefix, "weather", target_date)
    else:
        logger.info(f"Bronze/weather/{target_date} already exists, skipping")
        raw_weather = []

    # Step 2: Silver validation
    if raw_311:
        valid_311, invalid_311 = validate_311_records(raw_311)
        failure_rate = len(invalid_311) / max(len(raw_311), 1)
        if failure_rate > config.quality_failure_threshold:
            logger.error(f"Quality gate FAILED: 311 failure rate {failure_rate:.2%} exceeds threshold {config.quality_failure_threshold:.0%}")
            return
        write_parquet_to_s3(valid_311, config, config.silver_prefix, "complaints", target_date)

    if raw_weather:
        valid_weather, invalid_weather = validate_weather_records(raw_weather)
        write_parquet_to_s3(valid_weather, config, config.silver_prefix, "weather", target_date)

    # Step 3: Gold computation
    gold_records = compute_stress_index(config, target_date)
    if gold_records:
        write_parquet_to_s3(gold_records, config, config.gold_prefix, "stress_index", target_date)

    elapsed = (datetime.utcnow() - start_time).total_seconds()
    logger.info(f"Pipeline completed in {elapsed:.1f}s for date: {target_date}")


if __name__ == "__main__":
    target = date.today() - timedelta(days=7)
    run_pipeline(target)

Run the pipeline end-to-end and verify idempotency

The real test of your pipeline is running it against live data and then running it again to confirm idempotency works.

• Run the pipeline from your project root by executing this command:

python -m src.pipeline

What should I see?

You should see JSON-formatted log lines in your terminal showing the pipeline start, Bronze ingestion counts, Silver validation results, Gold output counts, and a completion message with elapsed time.

If the target date's data already exists from earlier steps, you will see "already exists, skipping" messages for the Bronze layer. The Silver and Gold stages will still run because they read from whatever is already in S3.

• Verify that pipeline.log was created in your project root by running:

cat pipeline.log

You should see the same JSON log entries that appeared in your terminal, now persisted to a file.

• Now verify idempotency by running the pipeline again with the same command:

python -m src.pipeline

This time, both Bronze sources should log "already exists, skipping" messages. No new data is fetched or written. The Silver and Gold stages process the existing S3 data without creating duplicates.

Pipeline failing with an import error?

• Make sure you are running the command from your project root directory (the folder containing the src/ directory).
• Confirm that src/__init__.py exists. Without it, Python does not recognize src as a package.
• Check that your virtual environment is active. Your terminal prompt should show the environment name.

Help me debug a pipeline import error in my Python project

Seeing an API timeout or HTTP error?

• The NYC 311 and NOAA APIs can be slow. If you hit a timeout, try running the pipeline again. The idempotency logic means any data already written will be skipped.
• For NOAA, confirm your NOAA_API_TOKEN environment variable is set correctly. An invalid token returns a 401 error.

Help me fix an API error in my data pipeline

Secret mission

Add Data Quality Gates

Your pipeline currently trusts its data sources. But what happens when an upstream API sends garbage records? Build defensive quality gates that halt the pipeline before bad data reaches the Silver layer, and produce diagnostic JSON reports so you can pinpoint exactly what went wrong.

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. The S3 data stored in this project is minimal (a few MB of Parquet files), so ongoing storage costs are fractions of a penny per month.

Resources you used:

• 1 AWS S3 bucket storing Bronze, Silver, Gold, and quality report data as Parquet and JSON files.
• Local project directory on your Desktop containing Python source code, virtual environment, and pipeline.log.

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the pipeline with additional data sources.

• Your S3 bucket stays active. At this data volume, storage costs are covered by the AWS Free Tier (up to $200 in credits for new customers, valid 12 months).
• Your local project files remain on your Desktop, ready for future runs.
• Your Python virtual environment stays in place with all dependencies installed.

Pause - I'll come back to this later

There is no running compute to pause. S3 is pure storage with no active processes.

• Your S3 data stays as-is. You are only charged for storage (fractions of a cent at this scale).
• Deactivate your Python virtual environment if it is still active by running:

deactivate

• Your project files and virtual environment remain on disk, ready to reactivate when you return.

Delete - I don't want to use this again

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

• Delete your S3 bucket and all objects inside it by running:

aws s3 rb s3://[[S3_BUCKET="your-lakehouse-bucket-name"]] --force

What does this command do?

The aws s3 rb command removes an S3 bucket. The --force flag deletes all objects inside the bucket first, then removes the bucket itself.

• Delete your local project directory by running:

cd ~/Desktop
rm -rf nyc-stress-lakehouse

• If you are on Windows, run this instead:

cd ~\Desktop
rmdir /s /q nyc-stress-lakehouse

• Confirm the directory is removed by running:

ls

You should no longer see the nyc-stress-lakehouse folder in the output.

Nice Work!

Nice Work!

You just built a production-grade data lakehouse pipeline from scratch. Your pipeline ingests real NYC data from two public APIs, validates it through strict schema contracts, and produces an analytical output that correlates urban complaints with weather conditions.

You've learned how to:

• Build a medallion lakehouse architecture (Bronze/Silver/Gold) on AWS S3 with Hive-partitioned Parquet files and idempotent batch processing.
• Ingest data from heterogeneous public APIs (NYC 311 via Socrata SODA and NOAA CDO v2) and enforce data contracts using Pydantic models with strict validation.
• Compute a daily Neighborhood Stress Index in the Gold layer that correlates complaint intensity with weather deviation per ZIP code using DuckDB SQL transformations.
• Secret Mission: Built data quality gates that halt the pipeline when validation failure rates exceed a configurable threshold and produce diagnostic JSON reports to S3.

Ready to quiz yourself?