ELT Pipeline with S3 and dbt

Build an ELT pipeline that extracts weather data, lands it in S3, and transforms it with dbt.

Introduction

30 Second Summary

Every weather app, financial dashboard, and recommendation engine depends on data that flows from messy sources into clean, queryable tables. The system that makes this happen is called a data pipeline.

In this project, you will build a complete ELT pipeline that extracts live weather data from an API, stores the raw JSON in cloud storage, then transforms it into analytics-ready models using industry-standard tools.

What You'll Build

When you run the pipeline end-to-end, you will watch raw weather forecasts for five cities land in your S3 bucket, then transform into clean daily summaries you can query to answer questions like "which city is warmest this week?"

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

  • A live extraction script that pulls multi-city weather forecasts from the Open-Meteo API and lands the raw JSON in S3 with date-partitioned keys.
  • A dbt project with a staging layer that flattens nested JSON and a marts layer that aggregates hourly data into daily summaries inside DuckDB.
  • Automated data quality tests that block bad data before it reaches your analytics layer, catching nulls, duplicates, and impossible values.
  • Secret Mission: Add incremental materialization so your pipeline only processes new data on each run.

Are there any prerequisites?

This project assumes you are comfortable writing Python scripts and basic SQL queries. You also need an AWS account with the AWS CLI already configured.

If you have not set up the AWS CLI yet, complete that first. You will need working credentials that can create S3 buckets and upload files.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters for your data engineering career.

Set Up Your Data Engineering Environment

Before you can extract and transform data, you need the tools that make it possible. dbt (data build tool) handles transformations, and DuckDB gives you a local analytics database that runs without a cloud warehouse.

In this step, you will install dbt with the DuckDB adapter, scaffold a project, and configure it to read from S3. These are the same patterns you would use with Snowflake or BigQuery, just without the cost or setup time.

In this step, get ready to:
  • Create a Python virtual environment and install dbt-duckdb.
  • Scaffold a dbt project and configure it for DuckDB with S3 access.
  • Verify your AWS CLI credentials are working.
Create a virtual environment and install dbt-duckdb

A virtual environment keeps your project dependencies isolated from other Python projects on your machine. Installing dbt-duckdb pulls in both dbt-core and DuckDB automatically.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir weather-elt-pipeline
cd weather-elt-pipeline
  • Create a Python virtual environment by running:
python3 -m venv dbt-env
  • Activate the virtual environment by running:

macOS

source dbt-env/bin/activate

Windows

dbt-env\Scripts\activate

You should see (dbt-env) appear at the beginning of your terminal prompt.

  • Install dbt-duckdb by running:
pip3 install dbt-duckdb

What does this install?

The dbt-duckdb package is the DuckDB adapter for dbt. When you install it, pip automatically pulls in dbt-core (the transformation engine) and duckdb (the analytics database) as dependencies.

One install command gives you all three tools.

  • Confirm dbt installed correctly by running:
dbt --version

You should see dbt-core version information printed in your terminal.

Command not found?

Make sure your virtual environment is active. You should see (dbt-env) in your terminal prompt. If not, re-run the activate command above.

If pip installed successfully but dbt is not found, try running python3 -m dbt --version instead. Help me troubleshoot my dbt installation.

Scaffold the dbt project and configure profiles.yml

The dbt init command creates the standard folder structure that every dbt project uses. After scaffolding, you will configure the connection profile so dbt knows to use DuckDB with S3 access.

  • Scaffold the dbt project by running:
dbt init weather_pipeline --skip-profile-setup

What does this command do?

The dbt init weather_pipeline command creates a new folder called weather_pipeline with dbt's standard directory structure (models, tests, macros).

The --skip-profile-setup flag tells dbt to skip the interactive database connection prompts. You will configure the connection manually in the next step to add S3 access settings that the interactive flow does not cover.

  • Move into the new project folder by running:
cd weather_pipeline
  • Confirm the project structure exists by running:
ls

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

Now you need to create a profiles.yml file in your project directory. This file tells dbt how to connect to DuckDB and where to find data in S3.

  • Create the profiles.yml file in your project root by pasting the following configuration:
weather_pipeline:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: warehouse.duckdb
      extensions:
        - httpfs
      settings:
        s3_region: us-east-1

What does each field do?

  • weather_pipeline matches the profile name in dbt_project.yml so dbt knows which connection settings to use.
  • path: warehouse.duckdb tells DuckDB to persist data in a local file. DuckDB creates it automatically on first run.
  • extensions: httpfs loads the HTTP filesystem extension so DuckDB can read files directly from S3 URLs.
  • settings: s3_region tells the httpfs extension which AWS region to use when accessing S3.

Next, update the dbt_project.yml to configure how models are materialized.

  • Open dbt_project.yml in your code editor and replace its contents with:
name: weather_pipeline
version: "1.0.0"

profile: weather_pipeline

model-paths: ["models"]
test-paths: ["tests"]
source-paths: ["models"]

models:
  weather_pipeline:
    staging:
      +materialized: view
    marts:
      +materialized: table

What does the models configuration do?

The models section defines default materialization strategies for each layer of your project.

Staging models use view (lightweight, always reflects current source data). Marts models use table (persisted for fast queries by analytics consumers).

  • Save both profiles.yml and dbt_project.yml.
  • Verify dbt can find your configuration by running:
dbt debug

You should see green checkmarks for configuration, dependencies, and connection. If the connection test passes, DuckDB is working and profiles.yml is configured correctly.

Seeing a profile not found error?

Make sure the profile name in profiles.yml (weather_pipeline) matches the profile field in dbt_project.yml exactly.

Also confirm that profiles.yml is in the same directory as dbt_project.yml (the project root). dbt searches the current working directory first. Help me fix my dbt profile configuration.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Here is the complete profiles.yml:

weather_pipeline:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: warehouse.duckdb
      extensions:
        - httpfs
      settings:
        s3_region: us-east-1

And here is the complete dbt_project.yml:

name: weather_pipeline
version: "1.0.0"

profile: weather_pipeline

model-paths: ["models"]
test-paths: ["tests"]
source-paths: ["models"]

models:
  weather_pipeline:
    staging:
      +materialized: view
    marts:
      +materialized: table
Verify your AWS CLI credentials

Your extraction script will upload data to S3 in the next step. Before you get there, confirm that your AWS CLI credentials are properly configured.

  • Verify your AWS credentials by running:
aws sts get-caller-identity

Before you check the output, what do you expect to see if your credentials are configured correctly?

You should see a JSON response containing your Account, UserId, and Arn. This confirms your AWS CLI can authenticate and interact with AWS services like S3.

Seeing an authentication error?

If you see a credentials error, your AWS CLI is not configured. Run aws configure and enter your Access Key ID, Secret Access Key, default region (us-east-1), and output format (json).

If you do not have an access key yet, create one in the AWS IAM console under your user's **Security credentials** tab. Help me configure my AWS CLI credentials.

Your data engineering environment is ready. Next up, you will write a Python script that pulls live weather data from an API and lands it in S3.

Extract Weather Data and Land It in S3

Your data engineering environment is ready to go. Now it's time to build the first real piece of the pipeline: extraction.

Real ELT pipelines always start by pulling raw data from a source and landing it in cloud storage. This creates an immutable copy that the transformation layer can reference later without ever hitting the source API again.

In this step, get ready to:
  • Create an S3 bucket for raw data storage.
  • Write a Python script that extracts weather data from the Open-Meteo API for 5 cities.
  • Run the script and confirm the raw JSON lands in S3.
Create your S3 bucket

Before your script can upload data, it needs somewhere to put it. S3 buckets are globally unique, so you need to pick a name that no one else has used.

  • Create your S3 bucket by running this command:
aws s3 mb s3://[[BUCKET_NAME="your-weather-pipeline-bucket"]]

What does this command do?

The aws s3 mb command creates ("makes") a new S3 bucket. Your bucket name must be globally unique across all AWS accounts, so consider adding your name or a random number to the end (e.g. weather-pipeline-jsmith-2026).

  • Confirm the bucket exists by running:
aws s3 ls

You should see your new bucket name in the list of S3 buckets.

Bucket name already exists?

S3 bucket names are globally unique. If you see a "BucketAlreadyExists" error, choose a different name by adding numbers or your initials to the end.

help me fix my S3 bucket creation error

Write the extraction script

The Open-Meteo API provides free weather forecast data with no API key required. Your script will call it for 5 cities, collect the JSON responses, and upload the combined result to your S3 bucket.

  • Install boto3 (the AWS SDK for Python) by running:
pip3 install boto3

Seeing a permissions error?

Make sure your virtual environment is active. You should see (dbt-env) at the start of your terminal prompt. If not, activate it again with source dbt-env/bin/activate (macOS/Linux) or dbt-env\Scripts\activate (Windows).

help me fix my pip install error

  • Create a new file called extract_weather.py in your project root (the same folder as dbt_project.yml).
  • Add the imports and city configuration by pasting this code into extract_weather.py:
import json
import datetime
import urllib.request
import boto3

# Define the 5 cities to extract weather data for
CITIES = {
    "Berlin": {"lat": 52.52, "lon": 13.41},
    "London": {"lat": 51.51, "lon": -0.13},
    "New York": {"lat": 40.71, "lon": -74.01},
    "Tokyo": {"lat": 35.68, "lon": 139.69},
    "Sydney": {"lat": -33.87, "lon": 151.21},
}

# Your S3 bucket name (must match the bucket you just created)
S3_BUCKET = "[[BUCKET_NAME="your-weather-pipeline-bucket"]]"

What does this code do?

The CITIES dictionary maps each city name to its latitude and longitude coordinates. The Open-Meteo API uses these coordinates to return weather forecasts for each location.

The S3_BUCKET variable stores your bucket name so the upload function knows where to put the data.

  • Add the extraction function and entry point below the S3_BUCKET line:

def extract_weather():
    # Use today's date for the S3 key path
    today = datetime.date.today().isoformat()
    all_data = {}

    # Call the Open-Meteo API for each city
    for city, coords in CITIES.items():
        url = (
            f"https://api.open-meteo.com/v1/forecast"
            f"?latitude={coords['lat']}&longitude={coords['lon']}"
            f"&hourly=temperature_2m,relative_humidity_2m,wind_speed_10m"
            f"&forecast_days=7&timezone=auto"
        )
        with urllib.request.urlopen(url) as response:
            data = json.loads(response.read().decode())
            data["city_name"] = city
            all_data[city] = data

    # Upload combined JSON to S3 with date-partitioned key
    s3_key = f"raw/weather/{today}/cities.json"
    s3_client = boto3.client("s3")
    s3_client.put_object(
        Bucket=S3_BUCKET,
        Key=s3_key,
        Body=json.dumps(all_data, indent=2),
        ContentType="application/json",
    )
    print(f"Uploaded weather data to s3://{S3_BUCKET}/{s3_key}")


if __name__ == "__main__":
    extract_weather()

What does this code do?

The extract_weather() function does two things. First, it loops through each city and calls the Open-Meteo API requesting hourly temperature_2m, relative_humidity_2m, and wind_speed_10m for the next 7 days.

Second, it uploads the combined JSON to S3 using a date-partitioned key like raw/weather/2026-08-11/cities.json. This pattern means each day's extraction lands in its own folder, creating an immutable history of raw data.

  • Save extract_weather.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

Your extract_weather.py file has two sections you built above. Scroll up and confirm each one is present, in this order:

  • Imports and city configuration (the import lines, CITIES dictionary, and S3_BUCKET variable).
  • The extract_weather() function and the if __name__ == "__main__" entry point at the bottom.
Run the script and confirm data lands in S3

Your script is ready to pull live weather data and upload it. Before you run it, think about what you expect to see.

  • What do you think will happen when you run the script? Will it print anything? Will it take a few seconds?
  • Run your extraction script:
python3 extract_weather.py

You should see a message confirming the upload, something like:

Uploaded weather data to s3://your-weather-pipeline-bucket/raw/weather/2026-08-11/cities.json

Why does it take a few seconds?

The script makes 5 separate API calls (one per city) and then uploads the combined result to S3. Each API call adds a small amount of network latency.

  • Verify the file landed in S3 by running:
aws s3 ls s3://[[BUCKET_NAME="your-weather-pipeline-bucket"]]/raw/weather/

You should see a folder with today's date listed. Drill into it to confirm the JSON file is there:

aws s3 ls s3://[[BUCKET_NAME="your-weather-pipeline-bucket"]]/raw/weather/$(date +%Y-%m-%d)/

You should see cities.json listed with a file size of roughly 100-200 KB.

Seeing a NoSuchBucket or AccessDenied error?

Make sure the bucket name in extract_weather.py matches the bucket you created earlier exactly. Also confirm your AWS credentials are still valid by running aws sts get-caller-identity.

help me fix my S3 upload error

Your raw data is now safely stored in S3. Next up, you will teach dbt how to read this JSON and transform it into clean, queryable tables.

Load Raw Data into DuckDB with a dbt Source

Your raw weather data is sitting in S3, but your analytics layer has no way to query it yet. Right now, the JSON is just a file in a bucket with no connection to your transformation pipeline.

In this step, you will declare that S3 data as a dbt source and write a staging model that flattens the nested JSON into clean, queryable rows inside DuckDB. Sources mark the boundary between raw and transformed data in your transformation DAG.

In this step, get ready to:
  • Declare the S3 JSON as a dbt source.
  • Write a staging model that flattens nested JSON into tabular columns.
  • Run dbt and query the results in DuckDB.
Declare the S3 data as a dbt source

dbt sources tell your project where raw data lives. By declaring the S3 JSON as a source, every model downstream can reference it by name instead of hardcoding paths.

  • Create the models/staging/ directory inside your dbt project by running this command:
mkdir -p models/staging
  • Confirm the directory was created by running:
ls models/staging/

The directory should exist and the command runs without error, though it will be empty for now.

  • Create a file called sources.yml inside models/staging/ and paste the following configuration:
version: 2

sources:
  - name: raw_weather
    description: Raw weather data from Open-Meteo API stored in S3
    meta:
      external_location: "s3://your-weather-pipeline-bucket/raw/weather/**/cities.json"

What does this file do?

The name: raw_weather declares this source's identity in your dbt project. Other models can reference it by this name.

The external_location in meta points DuckDB at your S3 path. The ** glob matches any date folder, so the source always picks up the latest extraction.

  • Save models/staging/sources.yml.
  • Confirm the file exists by running:
ls models/staging/

You should see sources.yml listed in the output.

Don't see the file?

Make sure you ran mkdir -p models/staging first. Double check you are in your dbt project directory (the one containing dbt_project.yml).

Help me debug my directory structure

Write a staging model to flatten the JSON

The raw JSON from the API is deeply nested. Each city contains arrays of hourly timestamps, temperatures, humidity, and wind speed.

Your staging model will unpack all of that into one flat table with a row per measurement.

  • Create a file called stg_weather_hourly.sql inside models/staging/ and paste the following SQL:
with raw_data as (
    select
        unnest(from_json(
            content,
            '{"Berlin": {"city_name": "VARCHAR", "hourly": {"time": "VARCHAR[]", "temperature_2m": "DOUBLE[]", "relative_humidity_2m": "DOUBLE[]", "wind_speed_10m": "DOUBLE[]"}}}'
        )) as city_data
    from read_json_auto('s3://your-weather-pipeline-bucket/raw/weather/**/cities.json')
),

flattened as (
    select
        city_data.city_name as city,
        unnest(city_data.hourly.time) as timestamp_raw,
        unnest(city_data.hourly.temperature_2m) as temperature_c,
        unnest(city_data.hourly.relative_humidity_2m) as humidity_pct,
        unnest(city_data.hourly.wind_speed_10m) as wind_speed_kmh
    from raw_data
)

select
    city,
    cast(timestamp_raw as timestamp) as measurement_at,
    cast(timestamp_raw as date) as measurement_date,
    temperature_c,
    humidity_pct,
    wind_speed_kmh
from flattened
where city is not null

What does this model do?

  • The raw_data CTE uses DuckDB's read_json_auto to pull the JSON directly from S3 via the httpfs extension you configured in profiles.yml.
  • The from_json and unnest functions crack open the nested JSON structure and produce one row per city.
  • The flattened CTE further unnests the hourly arrays so each row represents a single hourly measurement for one city.
  • The final select casts timestamps and filters out any null city values.
  • Save models/staging/stg_weather_hourly.sql.
Run dbt and query the staging model

With the source declared and the staging model written, you can now execute the transformation and see real data flow through.

  • Before you run this, what do you think will happen? Will dbt read from S3 directly and produce a local DuckDB view?
  • Run the dbt models by executing:
dbt run

You should see output confirming that stg_weather_hourly was created as a view. dbt reads the JSON from S3 and registers it as a queryable view in warehouse.duckdb.

Seeing an S3 access error?

Make sure your AWS credentials are configured. DuckDB uses the same credentials as your AWS CLI. Run aws sts get-caller-identity to confirm they are still valid.

Also verify that the s3_region in profiles.yml matches the region where you created your bucket.

Help me debug my S3 access error

  • Query the staging model to confirm real data is flowing through by running:
duckdb warehouse.duckdb -c "SELECT city, measurement_date, temperature_c FROM stg_weather_hourly LIMIT 10;"

You should see rows of weather data with city names, dates, and temperature values for cities like Berlin, London, and Tokyo.

What should I see?

The output shows a table with columns for city, measurement_date, and temperature_c. You will see rows for cities like Berlin, London, and Tokyo with realistic temperature values for the coming week.

Query returns no rows?

Check that the S3 path in stg_weather_hourly.sql matches your actual bucket name. Also confirm that your extraction script from Step 2 successfully uploaded the JSON file by running aws s3 ls s3://your-weather-pipeline-bucket/raw/weather/.

Help me debug empty query results

✔️ Awesome, I've got everything!

Great. Double check you have saved both models/staging/sources.yml and models/staging/stg_weather_hourly.sql.

ⓧ I'd like to double check the full code

Here is models/staging/sources.yml:

version: 2

sources:
  - name: raw_weather
    description: Raw weather data from Open-Meteo API stored in S3
    meta:
      external_location: "s3://your-weather-pipeline-bucket/raw/weather/**/cities.json"

Here is models/staging/stg_weather_hourly.sql:

with raw_data as (
    select
        unnest(from_json(
            content,
            '{"Berlin": {"city_name": "VARCHAR", "hourly": {"time": "VARCHAR[]", "temperature_2m": "DOUBLE[]", "relative_humidity_2m": "DOUBLE[]", "wind_speed_10m": "DOUBLE[]"}}}'
        )) as city_data
    from read_json_auto('s3://your-weather-pipeline-bucket/raw/weather/**/cities.json')
),

flattened as (
    select
        city_data.city_name as city,
        unnest(city_data.hourly.time) as timestamp_raw,
        unnest(city_data.hourly.temperature_2m) as temperature_c,
        unnest(city_data.hourly.relative_humidity_2m) as humidity_pct,
        unnest(city_data.hourly.wind_speed_10m) as wind_speed_kmh
    from raw_data
)

select
    city,
    cast(timestamp_raw as timestamp) as measurement_at,
    cast(timestamp_raw as date) as measurement_date,
    temperature_c,
    humidity_pct,
    wind_speed_kmh
from flattened
where city is not null

Your raw S3 data is now flowing through dbt into DuckDB as clean, tabular rows. Next up, you will build a marts layer that aggregates this hourly data into daily summaries and add tests to catch bad data automatically.

Build a Marts Layer and Add Data Quality Tests

Your staging model is producing clean, tabular weather data from the raw JSON in S3. Analytics teams need more than clean rows, though. They need aggregated, business-ready tables that answer questions directly.

Without automated tests, bad data silently corrupts your models and nobody notices until a dashboard shows nonsense numbers. In this step, you'll build a mart that rolls hourly measurements into daily summaries, then add tests that fail loudly when data quality drops.

In this step, get ready to:
  • Build a daily weather summary mart that aggregates hourly data by city and date.
  • Add data quality tests for null checks, uniqueness, and temperature range validation.
  • Run dbt build to confirm all models and tests pass together.
Create the daily weather summary mart

A mart is where staging data gets shaped into business-ready tables. Each row in your mart should answer a question an analyst would ask, like "what was the average temperature in Berlin yesterday?"

  • Create the models/marts/ directory inside your dbt project by running:
mkdir -p models/marts
  • Create a new file called mart_daily_weather_summary.sql inside models/marts/ and paste this SQL:
with hourly as (
    select * from {{ ref('stg_weather_hourly') }}
)

select
    city,
    measurement_date,
    round(avg(temperature_c), 1) as avg_temperature_c,
    round(min(temperature_c), 1) as min_temperature_c,
    round(max(temperature_c), 1) as max_temperature_c,
    round(avg(humidity_pct), 1) as avg_humidity_pct,
    round(max(humidity_pct), 1) as max_humidity_pct,
    round(max(wind_speed_kmh), 1) as max_wind_speed_kmh,
    count(*) as hours_measured
from hourly
group by city, measurement_date

What does this model do?

  • ref('stg_weather_hourly') tells dbt to pull from your staging model, creating a dependency in the DAG.
  • The GROUP BY city, measurement_date collapses 24 hourly rows per city into one daily summary row.
  • Each aggregate column answers a specific question: average temperature, peak wind speed, humidity ceiling, and how many hours of data contributed.
  • Save mart_daily_weather_summary.sql.
  • Run the model to materialize it in DuckDB:
dbt run

You should see dbt run both the staging view and the new mart table, finishing with a summary showing 2 models completed successfully and 0 errors.

  • Query the mart to confirm data is flowing through by running:
python3 -c "import duckdb; duckdb.connect('warehouse.duckdb').sql('SELECT city, measurement_date, avg_temperature_c, max_wind_speed_kmh FROM mart_daily_weather_summary ORDER BY city, measurement_date LIMIT 5').show()"

You should see a formatted table with one row per city per day, showing numeric averages and maximums.

Seeing a "Catalog Error" or empty results?

  • Make sure the file is saved inside models/marts/ (not models/staging/).
  • Confirm the ref('stg_weather_hourly') name matches your staging model file name exactly (without the .sql extension).
  • Check that your S3 bucket still has the raw JSON file from the previous step.

help me debug my dbt mart model query results

✔️ Awesome, I've got everything!

Great. Make sure you've saved mart_daily_weather_summary.sql and confirmed query results before moving on.

ⓧ I'd like to double check the full code

with hourly as (
    select * from {{ ref('stg_weather_hourly') }}
)

select
    city,
    measurement_date,
    round(avg(temperature_c), 1) as avg_temperature_c,
    round(min(temperature_c), 1) as min_temperature_c,
    round(max(temperature_c), 1) as max_temperature_c,
    round(avg(humidity_pct), 1) as avg_humidity_pct,
    round(max(humidity_pct), 1) as max_humidity_pct,
    round(max(wind_speed_kmh), 1) as max_wind_speed_kmh,
    count(*) as hours_measured
from hourly
group by city, measurement_date
Add data quality tests

dbt tests run automatically alongside your models when you use dbt build. If a test fails, the pipeline stops and reports exactly what went wrong.

  • Create a new file called schema.yml inside models/marts/ and paste this configuration:
version: 2

models:
  - name: mart_daily_weather_summary
    description: Daily weather aggregates per city
    columns:
      - name: city
        tests:
          - not_null
      - name: measurement_date
        tests:
          - not_null
      - name: avg_temperature_c
        tests:
          - not_null
    tests:
      - unique:
          column_name: "city || '-' || cast(measurement_date as varchar)"

What do these tests check?

  • not_null on city, measurement_date, and avg_temperature_c ensures no row has missing values in these critical columns.
  • The unique test on the concatenated city and date guarantees you never have duplicate summaries for the same city on the same day.

These tests act as a gate. If bad data sneaks in, dbt refuses to proceed.

  • Save schema.yml.
  • Now create a custom singular test. Create the tests/ directory and a file called assert_temperature_in_range.sql by running:
mkdir -p tests
  • Paste this SQL into tests/assert_temperature_in_range.sql:
select *
from {{ ref('stg_weather_hourly') }}
where temperature_c < -60 or temperature_c > 60

How do singular tests work in dbt?

A singular test is a SQL query that returns rows that should NOT exist. If the query returns zero rows, the test passes. If any rows come back, those are the bad records and the test fails.

This test catches any temperature reading below -60C or above 60C. Those values would indicate corrupted data from the API.

  • Save assert_temperature_in_range.sql.

✔️ Awesome, I've got everything!

Great. You should now have models/marts/schema.yml and tests/assert_temperature_in_range.sql saved.

ⓧ I'd like to double check the full code

version: 2

models:
  - name: mart_daily_weather_summary
    description: Daily weather aggregates per city
    columns:
      - name: city
        tests:
          - not_null
      - name: measurement_date
        tests:
          - not_null
      - name: avg_temperature_c
        tests:
          - not_null
    tests:
      - unique:
          column_name: "city || '-' || cast(measurement_date as varchar)"
select *
from {{ ref('stg_weather_hourly') }}
where temperature_c < -60 or temperature_c > 60
Run dbt build to execute models and tests together

The dbt build command runs models AND tests in DAG order. If a test fails, dbt stops and reports the failure before downstream models run.

Before you run this, do you think all tests will pass? Consider whether real weather data could ever have null cities or temperatures outside the -60 to 60 range.

  • Run the full build to execute models and tests together:
dbt build

You should see dbt process your models first, then run each test. The final summary line shows all resources passing with 0 errors.

What does a passing build look like?

dbt prints each resource with its status. Models show as OK and tests show as PASS. The final line reads something like Done. PASS=6 WARN=0 ERROR=0 SKIP=0 TOTAL=6 confirming everything succeeded.

  • Query the final mart to answer "which city has the highest average temperature?" by running:
python3 -c "import duckdb; duckdb.connect('warehouse.duckdb').sql('SELECT city, measurement_date, avg_temperature_c FROM mart_daily_weather_summary ORDER BY avg_temperature_c DESC LIMIT 5').show()"

You should see a table showing the top 5 hottest city-days, ordered by average temperature.

Tests failing on dbt build?

  • If the not_null test on city fails, check your staging model's WHERE city IS NOT NULL filter is present.
  • If the unique test fails, you may have duplicate raw data from running the extract script multiple times on the same day. Re-run with a single JSON file in your S3 bucket.
  • If the temperature range test fails, that means your source API returned extreme values. This is the test working as designed.

help me debug failing dbt tests in my weather pipeline

Your pipeline now has a safety net. Bad data gets caught before it reaches your analytics layer. Next up, you'll run the full pipeline end-to-end to prove extraction through transformation works as a complete system.

Run the Full Pipeline End-to-End

Your dbt project has staging models, a marts layer, and data quality tests that all pass individually. But individual pieces working in isolation does not prove the pipeline works as a system.

In this step, you will run extraction through transformation in sequence and verify the complete data flow. This is exactly what you would demo in an interview to prove you can build and operate an end-to-end ELT pipeline.

In this step, get ready to:
  • Run the extraction script to pull fresh data from the API and land it in S3.
  • Run dbt build --full-refresh to rebuild all models and execute all tests.
  • Query the final mart to answer a real analytics question.
Pull fresh weather data from the API

The extraction script you built in Step 2 pulls live weather forecasts from the Open-Meteo API and uploads the combined JSON to S3. Running it again now gives you fresh data with today's date in the S3 key.

  • Run the extraction script by entering this command:
python3 extract_weather.py

You should see output confirming the upload:

Uploaded weather data to s3://your-weather-pipeline-bucket/raw/weather/2026-08-11/cities.json
  • Confirm the file landed in S3 by running:
aws s3 ls s3://your-weather-pipeline-bucket/raw/weather/

You should see a folder for today's date listed alongside any previous date partitions from earlier runs.

Not seeing today's date folder?

Make sure your virtual environment is active. The boto3 library lives inside it.

Check that aws sts get-caller-identity still returns your account. If your session expired, run aws configure again.

Help me debug my extraction script

Rebuild all models with dbt build

The --full-refresh flag tells dbt to drop and rebuild every model from scratch, rather than relying on cached state. This proves the pipeline works cleanly from raw data all the way through to the final mart.

  • Before you run this, predict: do you expect all tests to pass? Why or why not?
  • Run the full dbt build with a fresh rebuild:
dbt build --full-refresh

You should see dbt run each model in DAG order (staging first, then marts), followed by every test. All tests pass because the fresh weather data from Open-Meteo contains valid, non-null values within realistic temperature ranges.

What does --full-refresh do?

Without this flag, dbt reuses existing tables if nothing changed in the SQL. The --full-refresh flag forces dbt to drop the mart_daily_weather_summary table and rebuild it entirely from the current S3 data.

This is the pattern you use in production when source schemas change or when you need to guarantee a clean state.

Seeing test failures?

If the unique test fails, you may have duplicate date partitions in S3 from running the extraction script multiple times on the same day. Delete the duplicate with aws s3 rm and re-run the extract once.

If the not_null test fails, the API response may have returned incomplete data. Re-run python3 extract_weather.py to pull a fresh response, then run dbt build --full-refresh again.

Help me debug my dbt test failures

Query the mart to answer a real question

The whole point of a pipeline is to answer questions that raw data cannot answer quickly. Your mart has already done the heavy lifting of aggregating hourly readings into daily summaries per city.

  • Before you run the query below, predict: which of the 5 cities (Berlin, London, New York, Tokyo, Sydney) do you think has the highest average temperature today?
  • Query the mart to find out by running:
python3 -c "import duckdb; conn = duckdb.connect('warehouse.duckdb'); conn.sql('SELECT city, avg_temperature_c, max_temperature_c, hours_measured FROM mart_daily_weather_summary WHERE measurement_date = CURRENT_DATE ORDER BY avg_temperature_c DESC').show()"

You should see a table with all 5 cities ranked by average temperature, with the warmest city at the top. The result makes sense against the raw API data because the mart simply averages the hourly readings that came from Open-Meteo.

What does this prove?

This single query hits a pre-aggregated table that dbt built, tested, and validated. An analyst querying mart_daily_weather_summary never needs to know about the nested JSON in S3 or the flattening logic in staging.

That separation of concerns is exactly why companies use the staging-to-marts pattern. You just demonstrated it end-to-end.

Seeing zero rows?

If the query returns no rows, the extraction script may have run on a different date than today. Remove the WHERE measurement_date = CURRENT_DATE filter to see all available dates, then adjust your filter to match.

Help me query my DuckDB mart

Secret mission

Add Incremental Materialization

Production pipelines process terabytes daily. Rebuilding every table from scratch on each run is slow and expensive. Can you convert your mart to an incremental model that only processes new data?

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 uses S3 storage well within the free tier (under 5 GB) and a local DuckDB file, so there are no ongoing costs unless you manually re-run the extraction script.

Resources you used:

  • S3 bucket with raw weather JSON files (negligible storage, covered by free tier).
  • Local project directory with DuckDB database, Python scripts, and dbt models.
  • Python virtual environment with dbt-duckdb installed.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your pipeline or adding new cities and data sources.

  • The S3 bucket costs nothing under 5 GB of storage (free for 12 months on new AWS accounts).
  • The DuckDB file is local and takes no ongoing resources.
  • The pipeline does not run on a schedule, so no charges accumulate unless you manually re-run the extraction script.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and S3 data stay intact for when you return.

  • Deactivate your Python virtual environment by running:
deactivate
  • Close your terminal and code editor.
  • Your S3 bucket and local files will be waiting when you come back. Reactivate the virtual environment to pick up where you left off.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your S3 data, local files, and virtual environment permanently.

  • Delete your S3 bucket and all objects inside it by running:
aws s3 rb s3://[[BUCKET_NAME="your-weather-pipeline-bucket"]] --force
  • Deactivate your virtual environment if it is still active:
deactivate
  • Delete the local project directory (macOS/Linux):
rm -rf ~/Desktop/weather_pipeline
  • On Windows, run this instead:
rmdir /s /q %USERPROFILE%\Desktop\weather_pipeline
  • This removes the DuckDB database, extraction script, dbt models, tests, and the virtual environment all at once.

Nice Work!

Nice Work!

Nice work! You just built a production-style ELT pipeline that extracts live weather data, lands it in cloud storage, and transforms it into tested, analytics-ready models using dbt and DuckDB.

You've learned how to:

  • Built a Python extraction script that pulls live weather data from the Open-Meteo API for multiple cities and lands the raw JSON in S3 with date-partitioned keys, following the immutable raw data pattern.
  • Created a dbt project with a staging layer that flattens nested JSON and a marts layer that aggregates hourly readings into analytics-ready daily weather summaries per city.
  • Added automated data quality tests that block bad data from reaching the analytics layer, then ran the full pipeline end-to-end and queried meaningful insights from the final mart.
  • Secret Mission: Implemented incremental materialization so the pipeline only processes new data on each run, matching the pattern senior data engineers use in production.

Ready to quiz yourself?