Data pipeline guide

Build a Data Pipeline with Dagster

Dagster describes a pipeline by the data it produces: tables, files, and models are assets, and Dagster keeps track of which ones are up to date. This guide builds a daily pipeline that loads data from an API into PostgreSQL and derives a report table from it, on Wodby.

An API feeding a Dagster asset pipeline that loads a PostgreSQL warehouse

What you will build

Two assets fed by the euro exchange rates the European Central Bank publishes every business day, from the free Frankfurter API, which needs no key:

  • exchange_rates: a table with one row per day and currency. It is partitioned by day, so every day is loaded, retried, and backfilled on its own.
  • monthly_rates: the average rate of every currency per month, built from exchange_rates.

Schedules load the previous day every morning and refresh the report after it. Swap the API for your own sources and the tables for the ones your team reports on, and the structure stays the same.

What you need

  • A Wodby account and a connected Kubernetes cluster.
  • A GitHub, GitLab, or Bitbucket integration, so Wodby can create and build your repository.

1. Create the Dagster app

  1. Open the Dagster stack in the Wodby catalog and create an app from it.
  2. For the code service's source, choose Clone boilerplate and the Vanilla Dagster project. Wodby creates a repository with app/definitions.py and requirements.txt, builds it as a code location that Dagster loads, and rebuilds on every push.
  3. Wait for the first deployment, then open the Dagster domain under Endpoints. Its password is under Endpoints → Auths.

2. Create the warehouse database

The stack's PostgreSQL already stores Dagster's runs and events. Give your data a separate database and user on the same server.

  1. Open Databases and select the app's PostgreSQL database.
  2. Under DBs, create a DB named warehouse with the UTF8 charset.
  3. Under Users, create a user named etl with a strong password, and give it access to warehouse.

For production data you can use a managed database instead; only the connection URL in the next step changes.

3. Give the code location the connection

Your code runs in the code service, so the connection goes there. Open its Configuration → Variables and add:

WAREHOUSE_URL=postgresql://etl:<password>@postgres:5432/warehouse

Turn on Secret, since the value contains the password, and URL-encode any special characters in it. postgres is the database's private host, shown on its page in Databases. The code reads it with Dagster's EnvVar, so the password never appears in the repository or in Dagster's interface. See environment variables.

4. Define the assets

Clone the repository Wodby created and replace app/definitions.py with the code below. Set the partitions' start_date to a date a week or two back.

"""Loads the European Central Bank's daily euro exchange rates into the warehouse.

exchange_rates is a daily partitioned asset: each partition is one day's rates,
upserted into PostgreSQL. monthly_rates is built from it. A schedule
materializes yesterday's partition every day, and older days can be backfilled
from Dagster's interface.
"""

import dagster as dg
import psycopg2
import requests

CURRENCIES = ["USD", "GBP", "JPY", "CHF", "CNY"]

daily = dg.DailyPartitionsDefinition(start_date="2026-11-01")


class Warehouse(dg.ConfigurableResource):
    """The PostgreSQL database the pipeline loads into."""

    url: str

    def connect(self):
        return psycopg2.connect(self.url)


@dg.asset(
    partitions_def=daily,
    retry_policy=dg.RetryPolicy(max_retries=3, delay=300),
    kinds={"postgres"},
)
def exchange_rates(context: dg.AssetExecutionContext, warehouse: Warehouse) -> dg.MaterializeResult:
    """One row per currency and day, from the Frankfurter API."""
    day = context.partition_key
    response = requests.get(
        f"https://api.frankfurter.dev/v1/{day}",
        params={"symbols": ",".join(CURRENCIES)},
        timeout=30,
    )
    response.raise_for_status()
    payload = response.json()
    # The ECB publishes no rates on weekends and holidays; the API then returns
    # the last business day, which its own partition already holds.
    rows = [] if payload["date"] != day else [(day, c, r) for c, r in sorted(payload["rates"].items())]

    with warehouse.connect() as conn, conn.cursor() as cur:
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS exchange_rates (
                day date NOT NULL,
                currency char(3) NOT NULL,
                rate numeric(18, 6) NOT NULL,
                PRIMARY KEY (day, currency)
            )
            """
        )
        # Upserting makes reruns and backfills safe.
        cur.executemany(
            """
            INSERT INTO exchange_rates (day, currency, rate) VALUES (%s, %s, %s)
            ON CONFLICT (day, currency) DO UPDATE SET rate = EXCLUDED.rate
            """,
            rows,
        )
    return dg.MaterializeResult(metadata={"rows": len(rows)})


@dg.asset(deps=[exchange_rates], kinds={"postgres"})
def monthly_rates(warehouse: Warehouse) -> dg.MaterializeResult:
    """The average rate of every currency per month."""
    with warehouse.connect() as conn, conn.cursor() as cur:
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS monthly_rates (
                month date NOT NULL,
                currency char(3) NOT NULL,
                average_rate numeric(18, 6) NOT NULL,
                days integer NOT NULL,
                PRIMARY KEY (month, currency)
            );
            INSERT INTO monthly_rates
            SELECT date_trunc('month', day)::date, currency, avg(rate), count(*)
            FROM exchange_rates
            GROUP BY 1, 2
            ON CONFLICT (month, currency) DO UPDATE
            SET average_rate = EXCLUDED.average_rate, days = EXCLUDED.days
            """
        )
        cur.execute("SELECT count(*) FROM monthly_rates")
        count = cur.fetchone()[0]
    return dg.MaterializeResult(metadata={"rows": count})


daily_rates_job = dg.define_asset_job("daily_rates", selection=[exchange_rates], partitions_def=daily)
monthly_rates_job = dg.define_asset_job("monthly_rates_job", selection=[monthly_rates])

defs = dg.Definitions(
    assets=[exchange_rates, monthly_rates],
    jobs=[daily_rates_job, monthly_rates_job],
    schedules=[
        # Materializes the previous day's partition every day at 06:00 UTC.
        dg.build_schedule_from_partitioned_job(
            daily_rates_job, hour_of_day=6, default_status=dg.DefaultScheduleStatus.RUNNING
        ),
        dg.ScheduleDefinition(
            job=monthly_rates_job, cron_schedule="30 6 * * *", default_status=dg.DefaultScheduleStatus.RUNNING
        ),
    ],
    resources={"warehouse": Warehouse(url=dg.EnvVar("WAREHOUSE_URL"))},
)

What the pieces do:

  • Daily partitions. Each partition of exchange_rates is one day. The schedule built from the partitioned job materializes the previous day at 06:00 UTC, after the ECB has published it.
  • A resource for the warehouse. Warehouse holds the connection URL, so assets ask for a connection instead of reading variables themselves.
  • Upserts and retries. A rerun of a day replaces its rows, and a failed day retries three times, five minutes apart.
  • Weekends are skipped. The API answers a weekend or holiday with the last business day; that partition then loads no rows.
  • Schedules start on. They are marked as running in the code, so they work right after deployment.

requests and the PostgreSQL driver come with Dagster's image, so requirements.txt stays empty here. Add other packages there; don't list dagster itself, and pick Dagster libraries that match the Dagster service's version.

5. Deploy and backfill

  1. Commit and push. Wodby builds the code location, restarts it, and Dagster reloads the app location with the two assets, two jobs, and two schedules.
  2. Load the days before today: in Dagster, open exchange_rates, click Materialize, select the partitions since the start date, and launch the backfill. Each day runs as its own run.
  3. Materialize monthly_rates once the backfill finishes. From then on the schedules keep both up to date.

Runs execute inside the code location, which has your code and packages, and Dagster stores their history in PostgreSQL. Give the code service enough memory and CPU for your heaviest run.

6. Check the data

On the PostgreSQL app service's Overview, click Connect via web terminal (see web terminal), run psql -U postgres -d warehouse, and query the tables:

SELECT day, currency, rate FROM exchange_rates ORDER BY day DESC, currency LIMIT 10;
SELECT * FROM monthly_rates ORDER BY month, currency;

You should see one row per business day and currency, and one monthly average per currency. In Dagster, each exchange_rates partition shows how many rows it loaded.

Where to go next

  • Point a dashboard at the warehouse: the Metabase stack can use it as a data source.
  • Add assets for more sources and reports; Dagster shows how they depend on each other.
  • Back up the PostgreSQL database; the warehouse and Dagster's history live there. See backups.
  • Read the Dagster stack guide for code locations, runs, and limits.