# Pipeline orchestration in Mojo

> morca is a pipeline orchestrator written in Mojo: pipelines checked by the compiler, steps run across a cluster by Dista, over Iceberg tables, with every run recorded in Postgres so it resumes and retries.

morca runs data pipelines written in Mojo and checked by the compiler. Dista runs their steps across a cluster, over Iceberg tables, with every run recorded in Postgres.

[Get started](https://morca.dev/getting-started) [Read the docs](https://morca.dev/docs) [Dista](https://morca.dev/dista) [GitHub](https://github.com/morca-dev)

## Checked at compile time

A morca pipeline is a Mojo type. Its steps, their inputs and outputs, its partitions and its schedule are compile-time parameters, so the compiler checks every edge before anything runs, and each step declares the Iceberg tables it reads and writes. [Writing a pipeline](https://morca.dev/docs/build) shows how.

## Distributed by Dista

Steps hand their work to [Dista](https://morca.dev/dista), which runs Mojo functions across a cluster as tasks and actors, keeping each result on the worker that made it. On AWS, `orca aws` gives each run workers of its own, scaling up from none while tasks wait.

## Over Iceberg and Postgres

The data lives in Apache Iceberg tables in object storage. Every run, attempt and log is recorded in Postgres: a failed step retries, a killed run resumes at the steps that didn't succeed, and a retried commit is recognised rather than written twice. Sensors start a pipeline when a table it reads changes, and lineage traces each snapshot back to the runs and code that made it.

## Small and quick to build

On three nodes, eight queries over 1.4 billion taxi trips in S3 take 12.0 s a pass on Dista, against 28.7 s for Rust and 38.6 s for Ray. Dista's container image is 32 MB compressed, against 41 MB for Rust and 204 MB for Ray; [Dista's performance page](https://morca.dev/dista/performance) has the whole build.

## One example per layer

Orchestration:

```
from orca import Cron, Ctx, Mapped, Month, Monthly, Pipeline, Step, orca_main
from orca.tables import append

def load(mut ctx: Ctx, m: Month) raises -> Int64:
    var table = trips()
    return append(ctx, table, read_month(download(m)))

comptime Load = Step["load", Month, Int64, load, Mapped, retries=2]
comptime Taxi = Pipeline[
    "taxi", Load,
    partitions=Monthly["2011-01"],
    schedule=Cron["0 6 2 * *"],
    max_runs=4,
]

def main() raises:
    orca_main[Taxi]()
```

morca: the compiler checks every edge, the binary gets run, resume and status, and the scheduler fires it monthly and backfills 2011–2024 four runs at a time. [More](https://morca.dev/)

Distributed compute:

```
from dista import ObjectRef, Runtime, Task
from parquet import ParquetReader

def rows(path: String) raises -> Int:
    return ParquetReader.open(path).num_rows()

comptime Rows = Task[String, Int, rows]

def main() raises:
    var rt = Runtime[Rows](workers=16)
    var refs = List[ObjectRef[Int]]()
    for path in files:
        refs.append(rt.remote[Rows](path))
    var total = 0
    for r in refs:
        total += rt.get(r)
```

Dista: one compiled binary is the driver, the head and every worker, on a laptop or a cluster; large results stay on the worker that made them. [More](https://morca.dev/dista)

Data processing:

```
from iceberg import Catalog

var table = Catalog.rest("https://polaris.example/api/catalog").load("db.events")
var rows = table.scan().filter('["=", "region", "eu"]').select(["id", "amount"]).to_table()
```

Under both sits magmalake's native Iceberg stack: metadata, manifests, scan planning, Parquet decode, S3 IO and the REST catalog, all of it Mojo. [More](https://magmalake.dev/s/iceberg)

## The census runs on morca

morca and the [magmalake tins](https://magmalake.dev) are used to build [mojourney.dev](https://mojourney.dev), the census of Mojo on GitHub.

The [census](https://github.com/morca-dev/census) is the [Mojourney](https://mojourney.dev) project's app, built on morca: it imports every day of GH Archive into an Iceberg table, finds the public GitHub repositories that contain Mojo, reads each one's git history, and serves the console that shows how the Mojo ecosystem grows. Three pipelines write its five tables, on one small server, and its runs are public.

[The census console](https://mojourney.dev) [Its runs in morca's UI](https://orca.mojourney.dev) [How it counts](https://morca.dev/blog/introducing-mojourney) [Build a lake like it](https://morca.dev/getting-started)

## One machine, then AWS

### On one machine

- One pixi project holds everything: Mojo, the tins, and Postgres and git from conda-forge.
- The local runner forks a run's Dista workers on the same machine.
- No database to start with: with `ORCA_DATABASE_URL` unset, a run is kept in memory. Add a Postgres under `build/` for the scheduler, retries and the UI.
- The warehouse is a directory: with `ORCA_WAREHOUSE=$PWD/build/warehouse`, the Iceberg table `lake.events` is `build/warehouse/lake/events`.

```
export ORCA_WAREHOUSE=$PWD/build/warehouse
pixi run build
pixi run ./build/gharchive run --partitions 2024-06-12
```

### Then on AWS

- `orca aws up` deploys the project as one CloudFormation stack, and `orca aws push` builds one image of the same binaries.
- Each run gets EC2 workers of its own, added while its tasks wait and drained once they're idle; `--workers 0..4` starts from none.
- The warehouse and the logs move to `s3://`. The pipeline code doesn't change.

```
./build/orca aws up --email you@example.com
./build/orca aws push
./build/orca aws register --workers 0..4 gharchive
```

[Getting started](https://morca.dev/getting-started/step-by-step#local) walks through both with a real project.

## Docs, Dista, the blog

The [docs](https://morca.dev/docs) cover writing, running, deploying and operating pipelines, and the [API reference](https://morca.dev/docs/api) lists the Mojo package. [Dista](https://morca.dev/dista) has docs and measurements of its own. The [introduction](https://morca.dev/blog/introducing-orca) is on the [blog](https://morca.dev/blog).

Underneath, Parquet, Iceberg, the codecs, Postgres and Kubernetes clients are Mojo libraries published as tins on mojoshelf. They live on [magmalake.dev](https://magmalake.dev), with their docs and benchmarks.