# Data pipelines in Mojo

> Data pipelines in Mojo: checked by the compiler, distributed with Dista, recorded in Postgres so they resume and retry, documented for coding agents, and built on a native Iceberg stack published as tins on mojoshelf.

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

[Get started](https://magmalake.dev/getting-started) [Browse the org on GitHub](https://github.com/magmalake) [See the tins](https://magmalake.dev/tins)

## Checked at compile time

An [Orca](https://magmalake.dev/s/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](https://magmalake.dev/s/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](https://github.com/magmalake/skills) teach a coding agent Orca, Dista, a whole magmalake app and fast Mojo data code. [Getting started](https://magmalake.dev/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](https://magmalake.dev/s/dista/performance#images) has the whole build. Underneath, Parquet, Iceberg and the codecs are Mojo [tins](https://magmalake.dev/tins), measured against pyarrow on [the performance page](https://magmalake.dev/performance), including where they lose.

## 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](https://magmalake.dev/s/orca)

```
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://magmalake.dev/s/dista)

```
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](https://magmalake.dev/s/iceberg)

## 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://magmalake.dev/getting-started/step-by-step#local) walks through both with a real project.