Orca

Orca

Data pipelines, declared in Mojo

Orca is the layer that decides when work runs, in what order, and what happens when a step fails. A pipeline is a DAG of steps written in Mojo, and the compiler checks that each step takes what the step before it produces. It's partitioned by month or by day and scheduled in cron syntax, and the orca scheduler starts its runs, backfills ranges of partitions and retries what failed. The heavy work inside a step runs on Dista, the data it reads and writes lives in Iceberg tables, and Orca's own record of runs lives in Postgres. A web UI comes next. 

from dista import Seq
from orca import Cron, Ctx, Gather, Mapped, Month, Monthly, Pipeline, Step, orca_main

def size(mut ctx: Ctx, m: Month) raises -> Int:
    return m.month * 100

def total(mut ctx: Ctx, sizes: Seq[Int]) raises -> Int:
    var t = 0
    for i in range(len(sizes)):
        t += sizes[i]
    return t

comptime Size = Step["size", Month, Int, size, Mapped]
comptime Total = Step["total", Seq[Int], Int, total, Gather, after=Size]
comptime Sizes = Pipeline[
    "sizes", Size, Total,
    partitions=Monthly["2024-01"],
    schedule=Cron["0 3 2 * *", "America/New_York"],
    max_runs=2,
]

def main() raises:
    orca_main[Sizes]()

Size runs once per month of the run; Total runs once and gets all of their outputs as a Seq. Declaring Total with an input that isn't a Seq[Int] doesn't compile. The binary gets a command line: ./sizes run --partitions 2024-01..2024-03 runs it, and with ORCA_DATABASE_URL set, every step and attempt is recorded so a run that was killed can be resumed where it stopped. Registered with the scheduler, it runs at 03:00 New York time on the 2nd of each month for the month before, two runs at most at once.

Why

An orchestrator that knows the data lake

Dagster and Airflow are Python. A Mojo step runs behind a Python definition, and the compiler checks neither against the other. They track runs of code, not which Iceberg snapshot a step read or wrote. Orca is built from the same tins as the rest of magmalake, and keeps track of the data as well as the code.

Edges are type checks

A step's input type must match what its upstream produces, checked with comptime asserts when the pipeline is built. There is no YAML and no templating.

Reruns read the same data

A step's reads are pinned to the snapshot its first attempt saw, so a retry or a resume sees the same rows whatever was committed in between.

Commits are recognised

Every snapshot a step commits carries the run, step, partition and attempt in its summary. A retry after a crash finds its own commit instead of loading the month again.

Small

One binary per pipeline, which is also its Dista driver and workers, one Postgres database, and one scheduler process that holds no state of its own: it can be stopped and restarted at any time.

Status

Four of 5 phases done; Kubernetes is next

Phase 1, DAGs and runs, is done: the 2023–2024 NYC taxi load runs as a pipeline, to the row count and column sums of taxibench's table, and a run killed with kill -9 halfway resumes at the month it was on. Phase 2, partitions, cron schedules in time zones and the scheduler, is done too: every month of taxi trips since 2011, 1.44 billion rows, loaded as a backfill four runs at a time in 11 minutes, and the schedule filled the next month as it fell due. Phase 3, the web UI, is done: a failed step is found, its log read and its run resumed from the browser, and runs are started, rerun from a step, backfilled and cancelled there too. Phase 4, lineage and sensors, is done: a snapshot committed to a source table, by Orca or by anything else, runs exactly the pipelines downstream of it, in order. Phase 5 runs Orca on Kubernetes. The roadmap has every phase and its gate.

Works today

  • Steps and pipelines declared in Mojo, with every edge type-checked by the compiler
  • Partitions by month or by day, from a start: Monthly["2011-01"], Daily["2024-01-01"]
  • Runs recorded in Postgres; retries per step, and resume at the steps that didn't succeed
  • One owner per run, held by a heartbeat; a killed run resumes once its heartbeat is stale
  • Iceberg reads pinned to the snapshot the first attempt saw, and commits recognised on retry
  • Dista tasks inside steps, one runtime per run; per-attempt logs and the run's timeline
  • Cron schedules in a time zone, read from the system's TZif files, daylight saving included
  • The Orca scheduler: fires once each, backfills, max_runs, retries with backoff, missed fires caught up
  • The NYC taxi load as a pipeline: 2023–2024 to 79,478,796 rows, resumed after kill -9
  • Every month of taxi trips since 2011 backfilled four runs at a time, and the next filled by its schedule
  • orca ui: pipelines, runs, attempts' logs and the DAG in the browser, live while a run goes on, and every page as JSON
  • Start, resume, rerun from a step, backfill and cancel from the browser, as rows the scheduler acts on
  • Datasets declared on steps, reads= and writes=, checked whenever a step touches a table
  • Lineage: from a snapshot back to the runs, code and snapshots it was made from, and forward to what was made from it, on the command line and as a graph in the UI
  • Sensors: a pipeline that runs when a dataset it reads gets a snapshot, from Orca or from anything else, once per snapshot
  • Rebuild every pipeline downstream of a dataset, in dependency order, from the command line or the browser

Not yet

  • Runs on Kubernetes, and logs in S3 (phase 5)
  • A published package: Orca builds from source today

No benchmark numbers yet: Orca's measure so far is correctness, the taxi load against taxibench's table. Next fires take about 2 µs each.