morca

morca · API

orca.pipeline

Pipelines: a named list of steps, checked and run in order.

comptime Taxi = Pipeline["taxi", Download, Load, Compact]

var result = Taxi.run(parse_months("2024-01..2024-03"))

A pipeline is partitioned by month unless it says otherwise with partitions= (orca.partitions): its first step takes the partition key, or a Seq of them for a gather step.

Steps are listed after the step they follow, so the declared order is a valid run order; check asserts it at compile time, along with every edge's types and unique names. Every function that runs a pipeline calls check, so a pipeline that breaks a rule doesn't compile.

execute is the executor: it runs the steps that haven't succeeded yet and reports each attempt to a Store, either MemoryStore (what run uses: a run lives as long as the process) or orca.state.PgStore, which makes a run resumable by another process. Each step's output is kept encoded by dista's codec, as Postgres holds it. A step that raises is retried up to its retries, then fails its partition: the steps after it for that partition, and any gather step over it, are skipped, while other partitions carry on. One that raises orca.step.defer ends deferred, with no more retries, and its downstream steps are skipped likewise; the run is deferred if no step failed.

Structs

Pipeline

struct Pipeline[name: StringSpan[ImmStaticOrigin], *Ts: StepLike, *, tasks: TaskSet = NoTasks, partitions: Partitions = Monthly, schedule: ScheduleLike = NoSchedule, fills: Fill = Previous, catchup: Bool = True, max_runs: Int = Int(1), run_retries: Int = Int(3), retry_seconds: Int = Int(0), max_retry_seconds: Int = Int(0), sensor: Bool = False, keys_per_run: Int = Int(1), workers: Int = Int(2)]

A pipeline named name of the steps Ts, in run order, whose steps call the dista tasks, over the keys of partitions; run by schedule for the partition its fires fills (orca.schedule), and, with sensor, when a dataset it reads gets a new snapshot (orca.sensors). A backfill puts keys_per_run keys in each run it queues, and so do a fill and, above 1, a sensor (#240; orca.toml's keys_per_run overrides it): one for time keys by default, tens to hundreds for keyed partitions (orca.keyed), whose runs then each go through many keys. On Kubernetes, a run of a pipeline with tasks gets a cluster of workers dista workers (orca.kube); locally it forks ORCA_WORKERS.

The scheduler runs at most max_runs of its runs at once, and starts a failed scheduled or backfilled run again up to run_retries times, with backoff: retry_seconds (0 for the scheduler's ORCA_RETRY_SECONDS) doubling with each start, up to max_retry_seconds if it's set (orca.scheduler). run_retries=72, retry_seconds=3600, max_retry_seconds=3600 tries hourly for three days.

Implements Movable

comptime K = partitions.K

The partition key.

check

def check()

Asserts at compile time that the pipeline is well formed.

registry_hash

def registry_hash() -> UInt64

A hash over the pipeline's name, its steps' names, kinds, edges, functions and datasets, and its dista tasks: two binaries with the same hash run the same DAG.

describe

def describe() -> String

The DAG as JSON: the pipeline's name, its registry hash, its partitions, each step's name, kind, upstream, function, types and datasets in run order, and its dista tasks.

timeouts

def timeouts() -> String

step=seconds for each step with timeout_seconds, comma-separated, for the run's watchdog (orca.heartbeat).

sensed_datasets

def sensed_datasets() -> List[String]

The datasets its steps read and none of them writes, once each, in step order, and a keyed pipeline's keys dataset: what a sensor watches.

scheduled

def scheduled() -> Bool

load_schedule

def load_schedule() -> Schedule

The schedule, parsed, with its zone loaded; raises for a pipeline without one, or one whose expression or zone is bad.

fills_for

def fills_for(fire: Int64, s: Schedule) -> Self.K

The partition a fire at fire fills.

run

def run(keys: List[Self.K], run: Int = Int(0)) -> RunResult

Runs every step over keys, in order, recorded in a MemoryStore of its own, with a dista runtime of its own if the pipeline has tasks.

execute

def execute[S: Store](mut store: S, var prior: RunResult, keys: List[Self.K], run: Int, runtime: Int = Int(0)) -> RunResult

Runs the steps over keys that haven't succeeded in prior, in order, reporting each attempt to store.

prior holds the steps a run had already finished when it's resumed, with their outputs; they aren't run again. A step that raises is retried up to its retries, then fails its partition. An error from the store itself is raised: the run stops where it is, and can be resumed from there. runtime is the address of the run's tasks.RT, which steps reach through Ctx.runtime.