Skip to content
DrToller.Framework

How a run executes

Overview

Section titled “Overview”

Something outside the step starts a run: Streamlit, Airflow, the local CLI, or the Run API. It builds a RunContext (who, which step, which params) and calls dispatch. From there the framework loads DRTML, optionally arms metrics, picks a runner, and always finalizes metrics when the job ends — success or failure.

flowchart TD
  INT["Integration: Streamlit / Airflow / CLI / HTTP"] --> RC["RunContext"]
  RC --> RFM["run_from_manifest"]
  RFM --> PLG["load_plugins"]
  PLG --> LOAD["load + merge params"]
  LOAD --> START["emit_run_start"]
  START --> MET{"metrics needed?"}
  MET -->|yes| SETUP["compile + setup_metrics"]
  MET -->|no| DISP["Dispatch"]
  SETUP --> DISP
  DISP --> RUN["pattern runner or execution.entry"]
  RUN --> FIN["finally: finalize_metrics"]
  FIN --> END["emit_run_end"]

from drtoller.framework.processing.dispatch import run_from_manifest
outcome = run_from_manifest(manifest_path, ctx, user_params=…)

Or a thin step hook (run_partition_loop, run_mapping, …) that hits the same runners. Concepts: Core concepts. Runner choice: Dispatch.

1. Load + merge params

Section titled “1. Load + merge params”
  1. load_manifest_path — YAML + include fragments (DRTML v4 → lowered internal manifest).
  2. merged_params_from_manifest — params_defaults ⊕ UI/Airflow overlay.
  3. Assert manifest.step_id == ctx.step_id.

2. Metrics setup (when needed)

Section titled “2. Metrics setup (when needed)”

With runtime.metrics_enabled and a prometheus plan / mode:

  • compile_step_manifest → StoragePlan;
  • setup_metrics → RunMetrics + resource sampler.

3. Dispatch

Section titled “3. Dispatch”
flowchart TD
  START["run_from_manifest"] --> V{"drtml_version >= 3?"}
  V -->|yes| PAT{"pattern registered?"}
  PAT -->|yes| PR["pattern runner"]
  PAT -->|no| ENT{"execution.entry set?"}
  ENT -->|yes| IMP["import_symbol entry"]
  ENT -->|no| ERR["ValueError"]
  V -->|legacy| MODE{"execution.mode registered?"}
  MODE -->|yes| MR["mode runner"]
  MODE -->|no| LEG["import_symbol runner"]
  PR --> DONE["return outcome"]
  IMP --> DONE
  MR --> DONE
  LEG --> DONE
  DONE --> FIN["finalize_metrics"]

v3 / v4 (canonical):

  1. If get_pattern_runner(pattern) hits → that runner (mapping, reduce_merge, dataset_metrics, evaluation). Registry tables live in runtime/pattern_runners.py (ensure_builtin_runners).
  2. Else execution.entry is required → import_symbol(entry)(ctx, manifest_path=…).

How public run.type lowers (see Modes):

run.type Internal Path
mapping / reduce / dataset_metrics / evaluation pattern + builtin entry pattern registry
partition / embed_train execution.entry only dynamic import
custom run.entry → execution.entry dynamic import

Legacy: get_mode_runner(execution.mode) or manifest.runner.

Extend without touching the core: register_pattern_runner / register_mode_runner (prefer drtoller.framework.ext; dispatch re-exports).
Full recipe: New execution mode.

4. Mode internals (partition_loop)

Section titled “4. Mode internals (partition_loop)”
flowchart TD
  C["compile + ParallelPlan"] --> F["feed shard metrics"]
  F --> B{"parallelism.backend"}
  B -->|inprocess| IP["3-stage: reader / processor / writer"]
  B -->|processes or ray| PW["workers + static assign or pull queue + merge"]
  IP --> CK["registry checkpoint from DRTML"]
  PW --> CK
  CK --> DM["metrics poller to Prometheus"]
  DM --> CL["finalize manifests / close session"]

The processor receives already materialized rows and returns ProcessResult.
Write: session.write → write_backend_for(spec.backend).

5. Finalize

Section titled “5. Finalize”

finally: finalize_metrics(…, outcome=finished|failed).

Other execution paths

Section titled “Other execution paths”
  • mapping — compile merge plan, mapping.build, output manifests JSON.
  • reduce_merge — sum/merge per DRTML; shuffle-bucket (or flat) work units over job_map + write lanes + manifest merge; metrics ticks via callback (storage does not import processing).
  • dataset_metrics — run offline corpus-growth jobs with the Python or PostgreSQL executor.
  • evaluation — stable-hash sample of vector, cluster, or graph artifacts (vector_job / cluster_job / graph_job); emit normalized metric rows (optional corpus-growth-shaped PG observation points).
  • embed_train — stateful W fold; shared embed_train/engine/training/train_loop.run_train_over_groups; Ray workers≥2 uses honest W shards; backends are transport only.
  • cluster fit/assign — streaming centroid fit + chunked assign under processing/cluster (optional sticky actors / shard parallel); K choice stays in domain / probe.
  • probe — N+1 candidate prepare/evaluate/decide/publish; evaluate/pipeline fan-out via job_map; nested evaluate shares the parallelism budget.

Responsibility matrix

Section titled “Responsibility matrix”
Layer Does Does not
Domain transform, in-RAM registry, choose K I/O, metrics emit
Processing lifecycle, queues, hooks parquet math, raw SQL
Storage read/write/feed Prometheus, Airflow
Telemetry emit / codegen runtime psycopg
DB transport domain column defaults
Section titled “Read next”