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. 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.
Structs
StepRun
struct StepRun
One step for one partition (or for the run, for a gather step).
Implements Copyable, Movable
- Fields
step: Stringpartition: String: Empty for a gather step.status: Stringoutput: List[UInt8]: The encoded output, when the step succeeded.error: Stringattempt: Int: The number of the attempt that settled it; 0 for a skipped step.
__init__
def __init__(out self, step: String, partition: String, status: String, var output: List[UInt8], error: String, attempt: Int = Int(0))
RunResult
struct RunResult
What a run did: every step and partition, in the order they ran.
Implements Movable
- Fields
steps: List[StepRun]
succeeded
def succeeded(self) -> Bool
find
def find(self, step: StringSpan, partition: StringSpan) -> Int
The index of step for partition, or -1.
status
def status(self, step: StringSpan, partition: StringSpan) -> String
output
def output[T: Movable](self, step: StringSpan, partition: StringSpan = "") -> T
The decoded output of step for partition.
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), 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: 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 (orca.scheduler).
Implements Movable
comptime K = partitions.K
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.
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.
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, recording nothing, 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.