Week 3 · Orchestrate with Dagster › Day 16

Day 16. Dagster assets and your first real project

1. Why this matters

Today the hotel pipeline becomes a graph you can see, run and reason about. After this, every lab in the course runs through Dagster.

2. Project layout

hotel-dagster/
├── pyproject.toml
├── src/hotel_pipeline/
│   ├── core/            # pure functions: parse, clean, dedup (no dagster import)
│   ├── defs/
│   │   ├── assets.py
│   │   ├── resources.py
│   │   └── checks.py
│   └── definitions.py   # the Definitions object Dagster loads
└── tests/

Dagster's scaffolding tool and recommended layout change between releases (recent versions use create-dagster and the dg CLI). Follow the current quickstart at docs.dagster.io for the scaffold; the code below works the same either way.

3. Assets

# src/hotel_pipeline/defs/assets.py
import dagster as dg
import polars as pl
from hotel_pipeline.core import pms

@dg.asset(group_name="hotel", kinds={"python", "parquet"}, owners=["team:bi"],
          description="Hotel reference data: rooms per hotel.")
def hotels() -> pl.DataFrame:
    return pl.read_csv("data/reference/hotels.csv")

@dg.asset(group_name="hotel", kinds={"polars"})
def stg_bookings(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    df = pms.read_all("data/landing/sftp/hotel")        # pure function from core/
    latest = pms.latest_version(df)                       # Day 12 logic
    latest.write_parquet("data/staging/hotel/stg_bookings.parquet")
    return dg.MaterializeResult(metadata={
        "rows": len(latest),
        "status_changes": int(df.height - latest.height),
        "preview": dg.MetadataValue.md(latest.head(5).to_pandas().to_markdown()),
    })

@dg.asset(group_name="hotel", deps=[stg_bookings, hotels])
def otb_by_stay_date() -> dg.MaterializeResult:
    """On-the-books room nights per hotel per stay date (confirmed only)."""
    out = pms.on_the_books("data/staging/hotel/stg_bookings.parquet", "data/reference/hotels.csv")
    out.write_parquet("data/marts/hotel/otb_by_stay_date.parquet")
    return dg.MaterializeResult(metadata={"rows": len(out)})

# src/hotel_pipeline/definitions.py
import dagster as dg
from hotel_pipeline.defs import assets
defs = dg.Definitions(assets=dg.load_assets_from_modules([assets]))

Run dagster dev, open http://localhost:3000, and click Materialize all. Every run is recorded with logs, metadata and duration.

4. Two ways to pass data

1. You write the file (above) and declare deps=. Explicit, easy to understand, good for big files and SQL tables.
2. Return a value and let an IO manager store it, and receive it as a function argument downstream. Less code; Day 17 covers it.