magmalake Orca

Docs

Writing a pipeline

A pipeline is one Mojo file: its steps, the partitions they run over, when they're due, and the tables they read and write. You compile it into a binary of its own, and the compiler checks how the steps fit together.

Steps

A step is a top-level def that takes a Ctx and one input and returns one output. You declare it 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 Lake = Pipeline["lake", Days, Measure, Total]

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

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 Lake = Pipeline[
    "lake", 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 the binary's describe, and fires prints the next ones (the command line).

Keyed pipelines

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. This is the repository's keyed example, labels: 

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. A keyed pipeline's binary has one command more, keys, which prints them: 

./build/labels keys                              # the snapshot of av2.logs it read, then one key per line
./build/labels run --partitions log-0001,log-0002
./build/labels run --partitions @file:keys.txt   # one key per line
./build/orca backfill labels @all                # runs of 100 keys for every key that isn't done

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 scheduler asks the binary for its keys by running keys itself.

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

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, which is what lineage walks.

Sensors

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.

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.

Building the binaries

Each pipeline is a Mojo file you compile into its own binary, conventionally one per file under pipelines/, with -I src for the project's own modules: 

pixi run mojo build -I src pipelines/gharchive.mojo -o build/gharchive
./build/gharchive run --partitions 2025-01-01

The orca command, which registers pipelines, schedules their runs and serves the UI, links no pipeline. It's in the package as orca.command, and 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 -I src bin/scheduler.mojo -o build/orca
./build/orca register ./build/gharchive

Neither is installed anywhere: you run them from build/. On AWS, orca aws push compiles both into the project's image, the orca command at /opt/orca/bin/orca and each pipeline at /opt/orca/pipelines/<name>, both on the image's PATH. The command line lists what each takes.

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: 

pixi run mojo build -I src -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) \
    pipelines/gharchive.mojo -o build/gharchive
./build/orca locate build/gharchive --root .     # writes build/gharchive.source.json
./build/orca register build/gharchive

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. The image orca aws push builds passes the source and commit and runs orca locate for you.