Building a Data Orchestrator with Dagster and dbt

December 30, 202412 min read

Data EngineeringDagsterdbtDuckDB

Moving data from A to B is the easy part. The pipeline is everything around it: what runs when, what depends on what, whether the numbers are right, and which run produced them. This post walks through a weather pipeline built on Dagster, dbt, and DuckDB.

What We Are Building

An ELT pipeline. Dagster pulls hourly and daily weather from the Open-Meteo API, loads it into DuckDB, then hands off to dbt for the staging, intermediate, and marts layers.

Key Features:

Asset-based orchestration with Dagster, not task-based DAGs
SQL transformations in dbt, with 19 data quality tests
Zero infrastructure: DuckDB is embedded, one file on disk
No API keys: Open-Meteo is free and unauthenticated
Runs under Docker Compose

Technology Stack

Six pieces, nothing exotic:

Dagster 1.9+: asset orchestration with lineage tracking
dbt 1.9+: SQL transformations, tests, docs
DuckDB 1.1+: embedded analytics database, single file
Python 3.11+: Pydantic for typed config
UV: dependency resolution and a committed lockfile
Docker: multi-container deployment

Project Structure

text

dataOrchestrator/
|-- config/
|   +-- cities.yml              # Configurable city coordinates
|
|-- data/
|   |-- raw/                    # JSON extraction files
|   +-- warehouse/              # DuckDB database file
|
|-- dagster_project/
|   |-- definitions.py          # Main entry point
|   |-- assets/
|   |   |-- extract.py          # Weather API extraction
|   |   |-- load.py             # DuckDB loading
|   |   +-- transform.py        # dbt integration
|   |-- resources/
|   |   |-- weather_api.py      # HTTP client resource
|   |   +-- duckdb.py           # Database resource
|   +-- schedules/
|       +-- daily.py            # 6 AM UTC daily schedule
|
+-- dbt_project/
    |-- models/
    |   |-- staging/            # Clean and rename
    |   |-- intermediate/       # Add business logic
    |   +-- marts/              # Analytics-ready tables
    +-- profiles.yml            # DuckDB adapter config

Asset-Based Orchestration

Airflow asks what to run. Dagster asks what should exist. The unit is an asset: a table, a file, a dbt model. Something with a current state that can be rebuilt on demand.

Why assets over tasks?

Lineage comes free. The dependency graph is the code
Rebuild one asset without rerunning everything above it
Every materialization records row counts, file sizes, and timestamps
The UI shows the graph, so onboarding is a screenshot

Extract: Fetching Weather Data

One asset, one API call per city, cities defined in config/cities.yml. Each call returns 7 days of history and 7 days of forecast, so daily runs overlap heavily. That overlap is deliberate: forecasts get revised, and the rerun picks up the correction.

python

@asset(
    description="Extract weather data from Open-Meteo API",
    group_name="extract",
)
def raw_weather_data(
    context: AssetExecutionContext,
    weather_api: WeatherAPIClient,
) -> MaterializeResult:
    settings = get_settings()
    all_data = {"cities": {}, "extracted_at": timestamp}

    for city_name, city_config in settings.cities.items():
        response = weather_api.fetch_weather(
            latitude=city_config.lat,
            longitude=city_config.lon,
            past_days=7,
            forecast_days=7,
        )
        all_data["cities"][city_name] = response

    # Save raw JSON
    output_file = RAW_DATA_PATH / f"weather_{timestamp}.json"
    output_file.write_text(json.dumps(all_data, indent=2))

    return MaterializeResult(
        metadata={
            "cities_extracted": len(settings.cities),
            "total_hourly_records": total_records,
            "output_file": MetadataValue.path(str(output_file)),
        }
    )

WeatherAPIClient is a Dagster resource, so the asset receives it as an argument instead of constructing it. Tests pass a fake and never touch the network:

python

class WeatherAPIClient(ConfigurableResource):
    base_url: str = "https://api.open-meteo.com/v1/forecast"
    timeout: int = 30

    def fetch_weather(
        self,
        latitude: float,
        longitude: float,
        past_days: int = 7,
        forecast_days: int = 7,
    ) -> dict:
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "hourly": "temperature_2m,relative_humidity_2m,precipitation",
            "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
            "past_days": past_days,
            "forecast_days": forecast_days,
        }

        with httpx.Client(timeout=self.timeout) as client:
            response = client.get(self.base_url, params=params)
            response.raise_for_status()
            return response.json()

Load: DuckDB Staging Tables

The load asset reads the newest JSON file and upserts it into DuckDB. The raw tables use (city, timestamp) as their primary key, so INSERT OR REPLACE overwrites the overlapping rows instead of stacking them. Rerun it ten times, same table.

python

@asset(
    deps=[raw_weather_data],
    description="Load weather data into DuckDB staging tables",
    group_name="load",
)
def staged_weather_data(
    context: AssetExecutionContext,
    duckdb_resource: DuckDBResource,
) -> MaterializeResult:
    raw_file = get_latest_raw_file()
    data = json.loads(raw_file.read_text())

    with duckdb_resource.get_connection() as conn:
        for city_name, city_data in data["cities"].items():
            # Upsert hourly data
            for i, timestamp in enumerate(city_data["hourly"]["time"]):
                conn.execute("""
                    INSERT OR REPLACE INTO raw_hourly_weather
                    (city, timestamp, temperature_c, humidity_pct, ...)
                    VALUES (?, ?, ?, ?, ...)
                """, [city_name, timestamp, temp, humidity, ...])

    return MaterializeResult(
        metadata={
            "hourly_records_loaded": hourly_count,
            "daily_records_loaded": daily_count,
        }
    )

DuckDB Resource Pattern

The connection lives behind a context manager, so it closes even when the asset raises:

python

class DuckDBResource(ConfigurableResource):
    database_path: str = "data/warehouse/weather.duckdb"

    @contextmanager
    def get_connection(self):
        conn = duckdb.connect(self.database_path)
        try:
            yield conn
        finally:
            conn.close()

    def init_schema(self):
        with self.get_connection() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS raw_hourly_weather (
                    city VARCHAR,
                    timestamp TIMESTAMP,
                    temperature_c DOUBLE,
                    humidity_pct DOUBLE,
                    precipitation_mm DOUBLE,
                    PRIMARY KEY (city, timestamp)
                )
            """)

Transform: dbt Medallion Architecture

Three dbt layers, the usual medallion split. Views until the last step, then a table:

Staging Layer

Staging renames columns and pulls out the date parts. No logic, so there's nothing here to argue about later:

sql

-- models/staging/stg_hourly_weather.sql
select
    city,
    timestamp as observation_timestamp,
    date_trunc('day', timestamp) as observation_date,
    extract(hour from timestamp) as hour_of_day,
    temperature_c,
    humidity_pct as relative_humidity,
    precipitation_mm,
    wind_speed_kmh,
    weather_code
from {{ source('raw', 'raw_hourly_weather') }}

Intermediate Layer

Intermediate is where the definitions live. What counts as cold, what counts as heavy rain, where morning ends. Define it once here and every mart agrees:

sql

-- models/intermediate/int_hourly_enriched.sql
select
    *,
    case
        when temperature_c < 0 then 'freezing'
        when temperature_c < 10 then 'cold'
        when temperature_c < 20 then 'mild'
        when temperature_c < 30 then 'warm'
        else 'hot'
    end as temp_category,

    case
        when precipitation_mm = 0 then 'dry'
        when precipitation_mm < 2.5 then 'light'
        when precipitation_mm < 7.5 then 'moderate'
        else 'heavy'
    end as precip_category,

    case
        when hour_of_day between 6 and 11 then 'morning'
        when hour_of_day between 12 and 17 then 'afternoon'
        when hour_of_day between 18 and 21 then 'evening'
        else 'night'
    end as time_of_day

from {{ ref('stg_hourly_weather') }}

Marts Layer

Marts materialize as tables, so a dashboard reads a result set instead of re-running the whole view chain:

sql

-- models/marts/fct_city_comparison.sql
{{ config(materialized='table') }}

with daily_stats as (
    select
        city,
        avg(temp_avg_c) as avg_temperature,
        sum(precipitation_sum_mm) as total_precipitation,
        avg(avg_humidity) as avg_humidity,
        count(*) as total_days
    from {{ ref('fct_daily_weather') }}
    group by city
)

select
    *,
    rank() over (order by avg_temperature desc) as warmest_rank,
    rank() over (order by total_precipitation desc) as wettest_rank,
    rank() over (order by avg_humidity desc) as humidity_rank
from daily_stats

Data Quality Testing

19 dbt tests run on every build:

yaml

# models/staging/_schema.yml
version: 2

models:
  - name: stg_hourly_weather
    description: "Cleaned hourly weather observations"
    columns:
      - name: city
        tests:
          - not_null
      - name: observation_timestamp
        tests:
          - not_null
      - name: temperature_c
        tests:
          - not_null

  - name: fct_city_comparison
    columns:
      - name: city
        tests:
          - not_null
          - unique

dbt build runs each model and then its tests, in dependency order. A failing staging test skips everything downstream, so the marts never get built on top of bad rows.

Dagster + dbt Integration

dagster-dbt reads the dbt manifest and turns every model into a Dagster asset. The ref() graph becomes the asset graph, no wiring by hand:

python

# dagster_project/assets/transform.py
from dagster_dbt import DbtCliResource, dbt_assets, DbtProject

dbt_project = DbtProject(
    project_dir=Path(__file__).parent.parent.parent / "dbt_project",
)

@dbt_assets(
    manifest=dbt_project.manifest_path,
    dagster_dbt_translator=CustomDagsterDbtTranslator(),
)
def dbt_weather_models(
    context: AssetExecutionContext,
    dbt: DbtCliResource,
):
    yield from dbt.cli(["build"], context=context).stream()

So stg_hourly_weather, int_hourly_enriched, and fct_daily_weather sit in the same graph as the extract and load assets. One lineage view from API call to mart, not two tools pointing at each other.

Scheduling

The pipeline runs daily at 6 AM UTC:

python

# dagster_project/schedules/daily.py
from dagster import ScheduleDefinition, AssetSelection

daily_weather_schedule = ScheduleDefinition(
    name="daily_weather_pipeline",
    cron_schedule="0 6 * * *",  # 6 AM UTC daily
    target=AssetSelection.all(),
    default_status=DefaultScheduleStatus.STOPPED,
)

It ships stopped. A schedule that starts itself on first deploy is how you find out your API quota is smaller than you thought, so turn it on from the UI when the pipeline has earned it.

Running the Pipeline

Option 1: Docker Compose

bash

# Start the containers
docker-compose up -d

# Access Dagster UI
open http://localhost:3333

# View logs
docker-compose logs -f

Option 2: Local Development

bash

# Install dependencies
uv sync

# Start Dagster development server
uv run dagster dev

# Access UI at http://localhost:3000

Materialize Assets

In the Dagster UI, click Materialize All. The run view gives you:

The asset graph with live status per node
Streaming logs, including dbt's own output
Row counts, file paths, and timestamps per materialization
Run history, so you can compare today against last Tuesday

Sample Results

Then query the marts:

sql

-- Query city comparison rankings
SELECT
    city,
    round(avg_temperature, 1) as avg_temp_c,
    round(total_precipitation, 1) as total_precip_mm,
    warmest_rank,
    wettest_rank
FROM fct_city_comparison
ORDER BY warmest_rank;

-- Results:
-- | city     | avg_temp_c | total_precip_mm | warmest_rank | wettest_rank |
-- |----------|------------|-----------------|--------------|--------------|
-- | Dubai    | 21.3       | 0.1             | 1            | 4            |
-- | Riyadh   | 15.1       | 0.0             | 2            | 3            |
-- | London   | 3.1        | 2.4             | 3            | 2            |
-- | New York | -0.8       | 19.6            | 4            | 1            |

Key Takeaways

Five things I'd carry into the next pipeline:

Assets over tasks. Naming what should exist gives you lineage and partial reruns for free. Naming what should run gives you neither
One job per tool. Dagster orchestrates, dbt transforms. Neither reaches into the other's work
Idempotent loads. A primary key plus INSERT OR REPLACE means a retry is boring, which is what you want at 6 AM
Resources, not globals. The asset takes its connection as an argument, so tests hand it a fake
Tests in the build. Bad rows stop at staging instead of ranking cities wrong in a mart

Full source on GitHub.