---
question: How do I schedule module pipelines with Dagster?
kind: guide
---

> **You are reading the v2.x docs (engine modules). New projects use v3.** [Go to the v3 docs](/capabilities/platform/overview) · [Migrate from 2.x to 3.0](/capabilities/upgrade/migrate-to-3)

# Dagster (legacy)

Dagster schedules jobs, data pipelines and asset materialization for a module's orchestrations. Where it runs depends on the ABI line:

| ABI line | Dagster |
|---|---|
| ABI 2.x | The scheduler of every project. `naas-abi-core` installs it, and `abi dev up` and `abi deploy local` start it. |
| ABI 3.x, project without NATS | A legacy, optional extra, to be removed in the next major. |
| ABI 3.x, NATS project | Not available. [Module jobs](/v2/capabilities/build/modules/module-jobs) replace it. |

:::warning Legacy on ABI 3.x
On ABI 3.x, Dagster is a **legacy, optional extra** and will be removed in the next major. In a NATS project, [module jobs](/v2/capabilities/build/modules/module-jobs) replace it: each module declares its jobs, and JetStream schedules them. Move orchestrations to module jobs before you turn NATS on: [Migrate from 2.x to 3.0](/user-guide/upgrade/migrate-to-3). The rest of the setup without NATS is on [Legacy mode: without NATS](/user-guide/get-started/legacy-mode).
:::

---

## Installing Dagster

On ABI 2.x, `naas-abi-core` installs Dagster: there is nothing to add.

On ABI 3.x, `naas-abi-core` no longer installs Dagster, and `naas-abi-core[all]` does not include it. A project without NATS (no `nats:` block in the config) needs the `dagster` extra (and `rabbitmq` for a RabbitMQ bus):

```bash
uv add 'naas-abi-core[all,dagster,rabbitmq]'    # ABI 3.x, project without NATS
```

`abi new project --without-nats` and `abi deploy local` (for a project without a `nats:` block) add it for you; `--without-nats` is an ABI 3.x option. Without the extra, `DagsterOrchestration.New()`, the RabbitMQ bus adapter and the Dagster app raise `MissingLegacyExtra` (a `ModuleNotFoundError` that prints the `uv add` command), the engine leaves out modules' Dagster orchestrations (one INFO line per module), and `abi dev up` leaves Dagster out with a notice. The extra keeps SQLAlchemy below 2.1, which breaks Dagster 1.13's PostgreSQL storage.

In a NATS project (ABI 3.x), one engine serves the kernel services, so Dagster, a second engine, cannot run next to the API (`EngineAlreadyServing`): the NATS stack has no Dagster service, and `abi dev up` leaves it out.

---

## Starting Dagster

On ABI 2.x, and on ABI 3.x in a project without NATS, `abi dev up` starts Dagster natively, and the Docker Compose stack (`abi deploy local`) runs it as the `dagster` service. In the ABI repository, the Makefile targets run it with Docker Compose, behind the `local` Compose profile on ABI 2.x and the `legacy` profile on ABI 3.x:

```bash
make dagster-dev           # Foreground
make dagster-up            # Background service
make dagster-ui            # Print the URL (http://localhost:3001) and run it in the foreground
make dagster-logs          # Follow its logs
```

---

## What Dagster does in ABI

Dagster orchestrates ABI pipelines as **assets** - materialized data artifacts with lineage tracking, scheduling, and failure recovery. Use it when you want to:

- Run data ingestion pipelines on a schedule (e.g. sync LinkedIn data every hour).
- Track which data assets are stale and need re-computation.
- Retry failed pipeline runs automatically.
- Monitor pipeline health in a visual UI.

---

## Defining assets and schedules

`abi new orchestration <name> [path]` writes a `DagsterOrchestration` subclass, in the current directory unless you give a path. It belongs in the module's `orchestrations/` folder, which `abi new module` creates on ABI 2.x, and on ABI 3.x in a project without NATS. Its `New()` returns the Dagster `Definitions`: assets, schedules, jobs and sensors. In a NATS project (ABI 3.x), the equivalent is a module job: see [From Dagster to module jobs](/v2/capabilities/build/modules/module-jobs#from-dagster-to-module-jobs).

```python title="orchestrations/MyOrchestration.py"
import dagster as dg
from naas_abi_core.orchestrations.DagsterOrchestration import DagsterOrchestration


@dg.asset(name="my_service_items", group_name="my_module")
def my_service_items(context: dg.AssetExecutionContext) -> None:
    from my_module.pipelines.MyPipeline import (
        MyPipeline,
        MyPipelineConfiguration,
        MyPipelineParameters,
    )

    graph = MyPipeline(MyPipelineConfiguration()).run(MyPipelineParameters())
    context.log.info(f"Ingested {len(graph)} triples")


hourly_sync = dg.ScheduleDefinition(
    job=dg.define_asset_job("hourly_myservice_sync", selection=["my_service_items"]),
    cron_schedule="0 * * * *",  # every hour
)


class MyOrchestration(DagsterOrchestration):
    """Dagster definitions of the module."""

    @classmethod
    def New(cls) -> "MyOrchestration":
        return cls(
            definitions=dg.Definitions(
                assets=[my_service_items],
                schedules=[hourly_sync],
            )
        )
```

Build the pipeline the way its own page does: a pipeline whose configuration needs an integration or a service takes them there. See [Creating a Pipeline](/v2/capabilities/build/knowledge-graph/creating-a-pipeline).

---

## Service management

```bash
make dagster-status         # Check asset materialization status
make dagster-materialize    # Trigger all assets immediately
make dagster-down           # Stop Dagster service
```