Data pipelines
in Mojo

Mojo from the scheduler down to the Parquet bytes, with no Python or JVM in the path.

Get started Browse the org on GitHub See the tins

Checked at compile time

An Orca 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.

Scalable and distributed

Steps hand their work to 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.

Reliable

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.

Agent friendly

magmalake's skills teach a coding agent Orca, Dista, a whole magmalake app and fast Mojo data code. Getting started has them, and the prompt from which an agent built a lake of GitHub activity that now runs on AWS.

Fast, small, 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, and builds from cold in 122 s, against 170 s for Rust; Dista's performance page has the whole build. Underneath, Parquet, Iceberg and the codecs are Mojo tins, measured against pyarrow on the performance page, including where they lose.

In code

One example per layer

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]()

Orca: the compiler checks every edge, the binary gets run, resume and status, and the orca scheduler fires it monthly and backfills 2011–2024 four runs at a time. More

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

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()

Metadata, manifests, scan planning, Parquet decode, deletes and deletion vectors, S3 IO and the REST catalog — all of it Mojo. More

Where it runs

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 walks through both with a real project.

Data pipelines in Mojo