Dagster Definitions / Job Structure -> Orchestra Pipeline Root

Overview

A Dagster project is described by a Definitions object that groups assets, jobs, schedules, sensors, and resources. Scheduling, retries, concurrency, and timeouts are set on jobs (via tags and RetryPolicy) and on ScheduleDefinition objects. None of these map to individual Orchestra tasks — they map to root-level fields on the Orchestra pipeline YAML. Apply this skill first, before any asset/op-level conversion, to establish the correct pipeline skeleton.

A useful rule of thumb: one Dagster job (or one cohesive asset selection driven by a schedule) -> one Orchestra pipeline.

Field Mapping Reference

ScheduleDefinition / @schedule -> schedule:

Orchestra requires a 6-field AWS EventBridge cron (minute hour day-of-month month day-of-week year), not Dagster's standard 5-field cron_schedule. Exactly one of day-of-month/day-of-week must be ?. Passing a 5-field string fails validation with "Incorrect number of values ... 6 required, 5 provided."

Orchestra ScheduleModel:

schedule:
  - cron: '0 2 * * ? *'
    timezone: UTC
    name: nightly
    run_inputs: {}
    exclude: []

Job tags + RetryPolicy -> configuration:

Concurrency -> configuration.concurrency.max_active

# Dagster (job tag / instance config)
@job(tags={"dagster/concurrency_key": "elt", "dagster/max_concurrent": 1})
def my_job(): ...

# Orchestra
configuration:
  concurrency:
    max_active: 1     # 0 or null = no limit

Partitions / backfills -> (drop)

Dagster PartitionsDefinition, partitioned schedules, and backfills have no Orchestra equivalent — Orchestra runs from activation forward and does not backfill. Convert only the cron; flag partition/backfill semantics as a manual review item.

Config classes / RunConfig / EnvVar -> inputs:

# Dagster
from dagster import Config

class ELTConfig(Config):
    target_env: str = "prod"
    table: str = "orders"

# Orchestra
inputs:
  target_env:
    type: string
    default: prod
  table:
    type: string
    default: orders
    optional: true

Valid type values: string, number, boolean, dict, list. Reference in task parameters as ${{ inputs.target_env }}.

EnvVar("SECRET") for credentials does not become an input — it becomes part of the Orchestra connection (see dagster-connections-to-orchestra).

Asset/op group + run tags -> task-level tags: only (not on the group)

pipeline:
  stage-001:
    tasks:
      task-001:
        tags:
          - elt
          - nightly

Confirmed live against Orchestra's /pipelines/schema validator: tags: directly on a stage (pipeline.<stage-id>.tags) is rejected with "extra_forbidden" — tags only exists on TaskModel, not TaskGroupModel. Copy Dagster's job/op-group tags onto every task, not once onto the enclosing stage.

Conversion Checklist

• ScheduleDefinition.cron_schedule -> schedule: block with cron: and timezone:
• concurrency limits -> configuration.concurrency.max_active
• RetryPolicy(max_retries, delay) -> configuration.retries + configuration.retry_delay (minutes — convert Dagster's seconds delay by dividing by 60)
• dagster/max_runtime tag -> configuration.timeout (seconds)
• failure/success notifications (run status sensors, hooks) -> alerts: block (see dagster-alerts-to-orchestra)
• Config / RunConfig non-secret values -> inputs: block
• EnvVar secrets -> Orchestra connection (never inline)
• tags -> preserve on every task individually — NOT on the enclosing stage (TaskGroupModel has no tags field)
• Drop: partitions, backfills, Backoff/Jitter strategies
• no schedule -> omit schedule:, set webhook: {enabled: true} if event-triggered

Full Before / After Example

Dagster (before)

from dagster import Definitions, job, op, ScheduleDefinition, RetryPolicy, Backoff

default_retry = RetryPolicy(max_retries=2, delay=300, backoff=Backoff.LINEAR)

@op(retry_policy=default_retry)
def extract(): ...

@op(retry_policy=default_retry)
def load(extracted): ...

@job(tags={"dagster/max_runtime": 3600, "team": "data-team"})
def nightly_elt():
    load(extract())

nightly_schedule = ScheduleDefinition(
    job=nightly_elt, cron_schedule="0 2 * * *", execution_timezone="UTC",
)

defs = Definitions(jobs=[nightly_elt], schedules=[nightly_schedule])

Orchestra YAML root (after)

version: v1
name: nightly-elt

# ScheduleDefinition(cron_schedule="0 2 * * *", execution_timezone="UTC")
schedule:
  - cron: '0 2 * * ? *'
    timezone: UTC

# RetryPolicy(max_retries=2, delay=300 seconds -> 5 minutes) + dagster/max_runtime=3600
configuration:
  retries: 2
  retry_delay: 5
  timeout: 3600

# Notifications converted from run status sensors / hooks
alerts:
  - name: slack-on-failure
    statuses:
      - FAILED
    destinations:
      - integration: SLACK
        destination: '#data-alerts'

# Dropped: Backoff.LINEAR strategy, team tag (use meta: if needed)
pipeline:
  # ... extract -> load tasks below, each with tags: [elt, nightly] (task-level only)

Gotchas

• RetryPolicy.delay is seconds, but retry_delay is minutes — Orchestra's field is integer MINUTES despite the name, so divide Dagster's delay by 60 (round up), don't just cast to int. No timedelta conversion (that is the Airflow trap, not Dagster). Cap at 120 minutes regardless.
• cron_schedule is 5-field, Orchestra's cron is 6-field — never copy verbatim; append ?/year per the AWS EventBridge rule above.
• Backoff / Jitter — Orchestra retry_delay is a flat integer. Backoff/jitter strategies have no equivalent; use the base delay and note the simplification.
• Partitions / backfills — PartitionsDefinition, partitioned schedules and backfills have no Orchestra equivalent. Convert the cron only and flag the partition logic.
• Definitions with multiple jobs — each Dagster job typically becomes its own Orchestra pipeline (separate YAML), not multiple stages in one pipeline.
• RunConfig / Config — op/asset run config becomes Orchestra inputs: referenced via ${{ inputs.x }}.
• EnvVar — non-secret values map to inputs:; secrets map to the Orchestra connection, never inline.
• Concurrency — op/run concurrency limits map to configuration.concurrency.max_active.
• tags on a stage fails real validation — confirmed by calling Orchestra's actual /pipelines/schema endpoint — TaskGroupModel rejects any tags key with "extra_forbidden". Always put tags: on each TaskModel, never on the stage wrapping it.

References

• Orchestra schedule docs: https://docs.getorchestra.io/docs/core-concepts/pipelines/schema
• Dagster schedules: https://docs.dagster.io/concepts/automation/schedules
• Dagster RetryPolicy: https://docs.dagster.io/concepts/ops-jobs-graphs/op-retries
• Dagster run config: https://docs.dagster.io/concepts/configuration/config-schema