Docs

Using Orca

Orca turns a Mojo file of steps into a pipeline binary with a command line. This page covers orca-mojo 0.1.0: installing it, declaring steps and pipelines, their partitions and schedules, running them by hand or from the scheduler, on this machine, on Kubernetes or on EC2, the web UI, recording and resuming runs in Postgres, reading and writing Iceberg tables from a step, lineage across pipelines, and sensors and rebuilds. The plan is in the repository's DESIGN.md.

Build it

Orca is the mojoshelf tin orca-mojo. With pixi and mojoshelf's pixi shelf installed, a new project adds it, and the tins it needs come with it:

pixi init -c https://conda.modular.com/max-nightly -c conda-forge lake && cd lake
sed -i.bak 's/^\[workspace\]/[workspace]\npreview = ["pixi-build"]/' pixi.toml   # tins build with pixi-build
pixi add "mojo-compiler==1.2.0"
pixi shelf add orca-mojo

A pipeline is one Mojo file whose main hands it to orca_main (Steps and Pipelines say what goes in it). Build it, then run it in memory to try it:

pixi run mojo build lake.mojo -o build/lake
./build/lake run --partitions 2024-01..2024-03

The orca command, which registers pipelines, schedules runs and serves the UI, is in the package too, as orca.command. A project's own orca is two lines, in any file not named orca.mojo (that name would be the module orca itself):

# bin/scheduler.mojo
from orca.command import orca_command

def main() raises:
    orca_command()

pixi run mojo build bin/scheduler.mojo -o build/orca
./build/orca locate build/lake --root .     # where each step's code is, for the UI's links
ORCA_DATABASE_URL=postgresql:///orca ./build/orca register ./build/lake
ORCA_DATABASE_URL=postgresql:///orca ./build/orca ui     # http://127.0.0.1:8270

To see the UI before running anything real, the repository's playground starts it on this machine with a throwaway Postgres, the example pipelines and their runs, two of them on a stand-in EC2:

git clone https://github.com/magmalake/orca.mojo && cd orca.mojo
git clone https://github.com/magmalake/dista.mojo ../dista.mojo
git clone https://github.com/magmalake/k8s.mojo ../k8s.mojo
pixi run playground          # then http://127.0.0.1:8270, log in as demo / demo-password

Building from source works the same way: the repository compiles Dista and k8s.mojo from those checkouts beside it, pixi run test runs the tests, and pixi run orca builds the scheduler to build/orca. It targets Mojo 1.2.0, with a nightly environment alongside. Its state goes through postgres-mojo, which loads libpq; the tests start a throwaway Postgres server of their own. orca.tables, the Iceberg part, has its tests in the data environment (pixi run -e data test-tables).

Steps

A step is a top-level def that takes a Ctx and one input and returns one output. It's declared once, with its name, its input and output types, the function, its kind, and the step it follows:

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

@fieldwise_init
struct Measured(Copyable, Movable):
    var month: Month
    var days: Int

def days(mut ctx: Ctx, m: Month) raises -> Measured: ...
def measure(mut ctx: Ctx, m: Measured) raises -> Measured: ...
def total(mut ctx: Ctx, ms: Seq[Measured]) raises -> Int: ...

comptime Days = Step["days", Month, Measured, days, Mapped]
comptime Measure = Step["measure", Measured, Measured, measure, Mapped, after=Days]
comptime Total = Step["total", Seq[Measured], Int, total, Gather, after=Measure]

A run is over a list of partition keys, months or days. A Mapped step runs once per key; a Gather step runs once per run and takes every key's output as a Seq. retries=2 in the declaration retries a failing step twice, and ctx.attempt says which attempt this is. ctx.log writes a line tagged with the step and partition, to standard output and to the attempt's log file.

Step outputs should be small values, such as paths, snapshot IDs and counts. They're stored in Postgres through Dista's codec, so a resumed run hands a step exactly the input its first attempt had. The data itself goes in tables.

Pipelines

A pipeline is a name and its steps, and the binary's main hands it to orca_main:

comptime Toy = Pipeline["toy", Days, Measure, Total]

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

Step and Pipeline check their edges with comptime asserts, so a pipeline that breaks one of these rules doesn't compile, and the compiler says which:

• A first mapped step takes the partition key, Month or Day; a first gather step takes a Seq of it.
• A mapped step takes the output of the mapped step it follows, and can't follow a gather step.
• A gather step after a mapped step takes a Seq of its output; after a gather step, its output.
• Steps are listed in run order, a step after the step it follows, with unique names.
• A step's after is the pipeline's own step of that name, compared by function.

The repository's tests/compile_fail/ holds one pipeline per rule, each checked to fail with its message. A pipeline's registry hash, over its steps' functions, kinds and edges, identifies the code a run ran.

Partitions and schedules

A pipeline says which partitions exist, and optionally when they're due, as parameters next to its steps:

comptime Toy = Pipeline[
    "toy", Days, Measure, Total,
    partitions=Monthly["2024-01"],       # every month from January 2024
    schedule=Cron["0 3 2 * *"],          # 03:00 UTC on the 2nd
    fills=Previous,                      # ... for the month before
    max_runs=2,                          # at most two runs at once
    run_retries=3,                       # a failed run is started again 3 times
]

Monthly[start] and Daily[start] are the partition sets; the first step's input must be their key, Month or Day, which the compiler checks. Cron takes the five usual fields, with lists, ranges, steps, names and the @daily-style shorthands, and a time zone: Cron["30 2 * * *", "America/New_York"]. Times are matched on that zone's wall clock. A time a spring-forward skips fires at the end of the gap, and a time a fall-back repeats fires once, as Vixie cron and systemd timers do. An expression that can never match, such as 0 0 30 2 *, doesn't parse. A fire fills the partition before the one it falls in (fills=Previous) or the one it falls in (Current), and catchup=False makes only the latest of the fires missed while the scheduler was down. All of it is in describe.

Partitions by name. Data that isn't divided by date, such as 20,000 recordings named by ID, is partitioned by the values of a column in a dataset:

from orca import Name
from orca.keyed import Keyed

comptime Label = Step["label", Name, Int, label, Mapped]
comptime Labels = Pipeline[
    "labels", Label, Score,
    partitions=Keyed["av2.logs", "log_id"],   # the keys are log_id's values
    keys_per_run=100,                         # a run covers 100 keys
    sensor=True,                              # new keys start runs
]

The first step takes a Name: 1 to 128 bytes of letters, digits and ._:-. The keys are the distinct values of the column in the dataset's current snapshot, which <binary> keys prints. A keyed pipeline has no schedule= or fills=, which the compiler checks; it runs by hand, by backfill, or as a sensor. --partitions takes names separated by commas, @all for every current key, or @file:<path> for a list, one per line; a range or a name that isn't a key is refused. A key is done once every mapped step has succeeded for it with the pipeline's current code, so a backfill queues runs of keys_per_run keys for the ones that aren't, and with sensor=True a new snapshot of the keys dataset starts runs for its new keys only.

The command line

Every pipeline binary gets the same commands:

toy describe                              the DAG, partitions and schedule as JSON
toy run --partitions 2024-01..2024-03     starts a run and runs it
toy resume <run> [--allow-new-code]       continues a run at what's unfinished
toy status [<run>]                        the latest runs, or one run's steps
toy fires [<count>] [--after <time>]      the next fires and the partitions they fill
toy sense                                 the current snapshot of each dataset it reads
labels keys                               a keyed pipeline's keys snapshot, then its keys

--partitions takes keys of the pipeline's kind, months (2024-01) or days (2024-01-31), and ranges of them, separated by commas, or a keyed pipeline's names. run and resume print one line per step and partition, then the run's status. Without ORCA_DATABASE_URL, run runs in memory and records nothing, and resume and status aren't available. The exit code says how it went:

The scheduler

orca is a separate binary that links no pipeline. It knows a pipeline by the describe it recorded and starts its runs as <binary> resume <run>, so the run records its own progress exactly as a run started by hand does. It needs ORCA_DATABASE_URL.

./build/orca register ./build/taxi
./build/orca backfill taxi 2011-01..2024-12
./build/orca scheduler

Each fire is a row with a unique (pipeline, time), inserted with its queued run in one transaction, so two schedulers ticking at once, or one killed mid-tick, never fire twice. Fires missed while it was down are made on its next tick, once each. A queued run starts only while fewer than max_runs of the pipeline's runs are live, counted under a lock on the pipeline's row. A scheduled, backfilled or sensed run that fails is started again after ORCA_RETRY_SECONDS, doubled with each start, plus up to a tenth as jitter, up to run_retries times; one whose process died is started again as soon as its heartbeat is stale. The scheduler holds nothing in memory, so it can be stopped and restarted at any time. Registering a new build moves the pipeline's queued runs to the new code.

On Kubernetes and EC2

By default the scheduler starts a run as a process on its own machine, and the run forks its Dista workers there. ORCA_RUNNER starts each run on a cluster of its own instead, from the image the pipeline was registered with:

orca register --image 123456789012.dkr.ecr.us-east-2.amazonaws.com/orca:data \
    --head-type c7g.large --worker-type c7g.2xlarge --workers 3 ./build/taxi
ORCA_RUNNER=ec2 orca scheduler

ORCA_RUNNER=ec2 starts a head instance, on demand, running the Dista head and the driver (<binary> resume <run>), and --workers worker instances pointed at it; --spot makes the workers Spot instances, where the account allows them. Each instance is tagged with its run, told to terminate when it shuts down, and recorded with its type, when it started and ended, and why it ended (it shut itself down, it was terminated, or a Spot interruption). ORCA_RUNNER=kubernetes creates a driver Job, a head Pod and Service and the worker Pods in ORCA_RUNS_NAMESPACE, all labelled with the run and owned by the Job. A pipeline without Dista tasks gets only its driver.

Both clean up in layers. A run's instances or pods end when it does; every tick the scheduler removes those of runs that aren't running any more and those older than ORCA_RUN_DEADLINE, so nothing a killed scheduler left behind keeps running. A run whose head goes is started again on new instances. Before making a cluster, the scheduler adds up the vCPUs of every live run's pods or instances, and a run that would take them past ORCA_VCPU_BUDGET waits.

The UI

orca ui serves pipelines and runs as web pages, from the same orca binary, on 127.0.0.1:8270 unless ORCA_UI_ADDR says otherwise. It holds no state: it reads the tables the scheduler and the runs write, connecting to Postgres per request, so it can run beside the scheduler or anywhere else that reaches the database. Nothing is fetched from elsewhere, so it works offline.

ORCA_DATABASE_URL=postgresql:///orca ./build/orca ui
curl -H 'Accept: application/json' http://127.0.0.1:8270/runs/7

Every path also answers JSON, to Accept: application/json or ?format=json, built from the same rows, so a script and a person see the same thing. While a run or an attempt is going on, its page refreshes its DAG, steps or log every two seconds, and stops once it has ended.

The UI never starts a pipeline binary. An action writes rows in Postgres, and the scheduler starts the queued runs on its next tick, within max_runs, so the UI does nothing without a scheduler running. It binds to localhost unless told otherwise, and refuses a POST whose Origin, or Referer without one, isn't its own, so another site's page can't post to it through a browser.

Login. Anyone who can reach the UI can read it. Its actions need a login once there's a user, or with ORCA_UI_LOGIN=required before there is one:

orca user add alice          # reads the password from standard input
orca user list
orca user remove alice       # and ends their sessions

Passwords are kept as bcrypt hashes made by Postgres. A login is a session cookie, HttpOnly and SameSite=Strict, and Secure with ORCA_UI_SECURE_COOKIES=1 behind HTTPS. Every action is recorded with its user and shown on its run's page, and so is every failed login: after 5 for one name, or 20 from one address, in 15 minutes, /login answers 429 without checking the password. Behind a proxy, ORCA_UI_TRUST_PROXY=1 takes the address from X-Forwarded-For.

On a run's page, a run with Dista tasks draws its timeline: one lane per worker, each task a bar coloured by its function. With ORCA_UI_URL set to the UI's public address, an Open in Perfetto link opens the full trace there. A run on EC2 lists its instances, with the number of workers over the run, why each instance ended, and each one's log. The step names, the pipeline and its Dista tasks link to their code (below), in a new tab, and the headings link to this page's sections (ORCA_DOCS_URL changes where).

Versions and code links

A pipeline binary can carry the repository and commit it was built from, which each run records and the UI links to:

mojo build -D ORCA_SOURCE=$(git remote get-url origin) \
    -D ORCA_COMMIT=$(git rev-parse HEAD) \
    -D ORCA_DIRTY=$(git diff --quiet HEAD && echo 0 || echo 1) \
    lake.mojo -o build/lake
orca locate build/lake --root .       # writes build/lake.source.json
orca register build/lake

describe reports the source; the registry hash doesn't change with it. orca register and every run and resume record a version: the registry hash, the source, the commit, whether the tree had changes, and the image. A pipeline's page lists its versions with their runs, and a run's page links the commit it ran, at <repo>/tree/<commit>. orca locate finds, in the source tree, the line of each step's function, each Dista task and the pipeline's own declaration, and writes them beside the binary, where orca register reads them; the UI then links each to its line at the commit. A binary built without these just has no links.

Runs in Postgres

With ORCA_DATABASE_URL set, the binary creates and migrates an orca schema at start-up and records every run, every step of every partition, and every attempt, with the inputs and outputs of each. Each change of state is one transaction.

ORCA_DATABASE_URL=postgresql:///orca ./build/toy run --partitions 2024-01..2024-03
./build/toy status          # the latest runs
./build/toy status 7        # run 7's steps and attempts
./build/toy resume 7

resume reruns every step that hasn't succeeded, in order, with its recorded input; steps that succeeded are read back, not run again. A run started by a binary with a different registry hash is refused unless --allow-new-code is passed.

One process owns a run at a time. It holds the run with a heartbeat, written from a thread with its own connection, and resume refuses a run whose heartbeat is younger than ORCA_STALE_SECONDS. A run whose process was killed goes stale and can be resumed; a run that ended can be resumed at once. Each attempt also checks that its process still owns the run, so a process that lost its run stops at its next step rather than racing the new owner. tests/kill.sh kills a run at a random step with kill -9, resumes it, and checks every step ran to success exactly once.

Postgres must keep synchronous_commit on: with it off, an immediate restart of the server can lose commits it already acknowledged. The chaos test from PR #6 found this when a run's row vanished.

Iceberg tables

A step reads and writes Iceberg tables through orca.tables, which is what makes a rerun safe:

from orca.tables import append, scan

def load(mut ctx: Ctx, raw: RawFile) raises -> Loaded:
    var src = Table.load_local(raw.table)
    var batches = scan(ctx, src).to_batches()
    var dst = Table.load_local(OUT)
    var id = append(ctx, dst, transform(batches))
    return Loaded(raw.month, id)

scan reads a table at a pinned snapshot: its current one on the step's first attempt, which is recorded, and the same one on every retry and resume, whatever was committed in between. A table that had no snapshot when the step first read it stays empty for that step.

append writes batches as one append snapshot, and commit commits data files already written, added and removed, as a snapshot of any operation, such as replace for a compaction. Both put orca.* properties in the snapshot's summary: pipeline, run, step, partition, attempt, commit number, code, and the snapshots read. Before writing anything they look for a snapshot this step already committed in this run. If an attempt committed and was killed before Postgres heard of it, the retry finds that snapshot and returns it, so a month isn't loaded twice. A table is known by its location.

Reads and commits are recorded in Postgres as they happen, not when the attempt ends. A run in memory has no run number of its own, so it pins reads across its retries but recognises no commits.

Datasets and lineage

A dataset is a table's name, namespace.table, that means the same table in every pipeline. ORCA_WAREHOUSE is the root of a filesystem catalog: lake.clean is the table at <warehouse>/lake/clean. A step declares the datasets it reads and writes, and loads them through its context:

from orca.tables import append, ensure_table, scan, table

def clean(mut ctx: Ctx, m: Month) raises -> Int64:
    var src = scan(ctx, table(ctx, "lake.raw")).to_batches()
    var dst = ensure_table(ctx, "lake.clean", SCHEMA)
    return append(ctx, dst, transform(src))

comptime Clean = Step[
    "clean", Month, Int64, clean, Mapped,
    reads=["lake.raw"], writes=["lake.clean"],
]

scan fails the attempt on a dataset the step didn't declare in reads=, and append and commit on one it didn't declare in writes=, naming the step and the dataset, so the graph of pipelines and datasets is known before anything runs and can't drift from what the steps do. describe lists each step's datasets, and orca register refuses a pipeline that would close a cycle across pipelines. A table outside the warehouse is still read and written by location, and isn't part of the graph.

Every read and commit is recorded with its dataset, so lineage is two walks over them. Back from a snapshot: the run, step, partition and code that made it, the snapshots that attempt read, what made each of those, and so on, to sources no Orca run made. Forward: everything made from it. Both stop after 20 steps.

$ ./build/orca lineage lake.report
lake.report@3141592653
made from:
  0 lake.report@3141592653 by report run 9 report code 0x5e1f... from lake.clean@2718281828
  1 lake.clean@2718281828 by clean run 8 clean 2024-02 code 0x9a0c... from lake.raw@1618033988
  2 lake.raw@1618033988 by ingest run 6 load 2024-02 code 0x77d2...
made from it:
  nothing

Sensors and rebuilds

sensor=True on a pipeline runs it when a dataset it reads, and doesn't write itself, gets a new snapshot:

comptime Report = Pipeline["report", Summarise, partitions=Monthly["2024-01"], sensor=True]

The scheduler finds snapshots in two ways. A snapshot a succeeded run of another pipeline committed is in Postgres, and is seen on the next tick. One that something other than Orca committed is seen by polling: every ORCA_SENSE_SECONDS the scheduler runs <binary> sense, which reads each dataset's current snapshot from its table's metadata, so the scheduler itself still reads no table. Each snapshot starts one run, trigger sensor, however many ticks or schedulers see it, and a run's page names the snapshots that started it. Registering a sensor records what its datasets hold then as already seen, so it doesn't rebuild history; a backfill or a rebuild does that.

The partitions a run covers come from the snapshot: the partition of the step that committed it, or the run's partitions for a gather step. A writer outside Orca can say which it holds with orca.partitions in its snapshot's summary; without that, the pipeline runs the partition its schedule would fill now, or the one today falls in. A day upstream is the month that contains it downstream, and a month upstream is each of its days.

$ ./build/orca rebuild raw.events 2024-03
rebuild 1
  level 0 clean 2024-03
  level 1 report 2024-03

A rebuild runs every pipeline downstream of a dataset, sensor or not, over the partitions given. Each pipeline is placed at its longest distance from the dataset, so one that reads two rebuilt datasets runs after both, and the scheduler queues a level once every run before it has succeeded. A run that fails stops the rebuild; resuming it from the UI carries on. The rebuild's own commits don't also start the sensors it reruns, so each pipeline runs once.

Dista inside steps

A step runs in the run's driver process and calls Dista for anything distributed. The pipeline names the Dista tasks its steps use, and a step gets the run's runtime from its Ctx:

from dista import ObjectRef, Task
from orca import Tasks

comptime Trips = Task[Day, Int, trips]
comptime FanoutTasks = Tasks[Trips]

def month_trips(mut ctx: Ctx, m: Month) raises -> Int:
    ref rt = ctx.runtime[FanoutTasks]()
    var refs = List[ObjectRef[Int]]()
    for d in range(1, days_in(m) + 1):
        refs.append(rt.remote[Trips](Day(m.copy(), d)))
    var total = 0
    for r in refs:
        total += rt.get(r)
    return total

comptime Fanout = Pipeline["fanout", MonthTrips, AllTrips, tasks=FanoutTasks]

Each run gets one Dista runtime, started before anything else so that the head and workers it forks inherit no database connection or thread. Locally it forks ORCA_WORKERS workers; with DISTA_HEAD set it joins that cluster instead, and with DISTA_ROLE set the pipeline binary is itself a Dista head or worker. Asking for a runtime of tasks the pipeline didn't declare is refused. The tasks are part of describe and the registry hash.

Logs

With ORCA_LOGS set to a directory, each attempt writes <pipeline>/<run>/<step>/<partition>/<attempt>.log: its start, its end, its error, and its ctx.log lines with UTC timestamps. The path is recorded with the attempt and shown by status <run>. A run with Dista tasks writes the runtime's timeline beside them as timeline.json, in Chrome's trace format, which Perfetto opens. The scheduler keeps a run's driver output as driver.log, and an EC2 run's head and workers upload theirs to cluster/<role>-<instance>.log every ORCA_CLUSTER_LOG_SECONDS while they run, as an instance can go without notice.

ORCA_LOGS and ORCA_WAREHOUSE can be locations in object storage, s3://bucket/prefix, configured from the AWS_* environment or the instance's role. An object can't be appended to, so a log is written to a file under ORCA_LOG_STAGE and uploaded when it's complete: an attempt's when the attempt ends, a timeline when the run does. The UI reads them back from wherever they are.

Environment

When things fail

A step that raises is retried up to its retries, then fails. The steps downstream of it are skipped, while other partitions' steps that don't depend on it carry on, so one bad month doesn't stop the others; the run ends failed and can be resumed. A process that dies leaves its run running with a stale heartbeat; resume marks the interrupted attempt failed and continues from there. A commit that landed before the crash is found by the retry rather than made again. Fuzzing, and a chaos test that kills runs, restarts Postgres and races two resumes over many seeds, run in CI.