pipeline
etl_craft.execution.pipeline ¶
run --pipeline_code: run a whole pipeline; --init-only and --finalize-only.
In local mode the engine runs every active task under one run, starting ready tasks after each
completion, at most Orchestration.Max_parallel_tasks at once. Each task goes through
run_task in a process of its own. Gates wait cooperatively without a worker;
a task whose dependencies can never be met is recorded SKIPPED. Execution continues until
every task is settled or none can start. Data tasks determine the run's outcome; failed alerts
are logged and recorded on their own tasks. A pipeline containing only alerts uses those
tasks' outcomes.
A new run starts only when the pipeline's dependencies on other pipelines are satisfied (see
gates); otherwise the run is recorded SKIPPED. An IN-PROGRESS run is resumed instead,
without that check, and its finished tasks are not run again. Before either, the connections the
run uses are tested (see connections).
In remote mode the orchestrator is the only source of truth for scheduling (see remote). It
runs each task with run --task_code, between run --init-only, which refuses rules the
orchestrator does not support, tests connections and starts the run, and run
--finalize-only, which records its decisions and ends the run: a task it never ran is
SKIPPED, and the run is FAILED when a data task failed, SKIPPED when it ran none.
No gate is checked and no upstream run is consumed; the orchestrator's DAGs hold those rules.
An operator may cancel a local run (etl-craft cancel): its running tasks are stopped, no
other task starts, and the run ends CANCELLED.
Every run of a pipeline with SLA_IN_HOURS is marked MET or BREACHED. While a local run
is going, a watcher marks it BREACHED as soon as the SLA passes; otherwise the finalize step
does. RunHooks.on_sla_lapse is called once per run, the first time the breach is seen.
MAX_BACKFILL_DAYS
module-attribute
¶
The most dates one backfill runs; a longer one is split, so a typo cannot start thousands.
NOT_RUN_BY_ORCHESTRATOR
module-attribute
¶
Why --finalize-only records a task SKIPPED in remote mode.
PipelineOutcome
dataclass
¶
How a pipeline run, or one step of it, ended.
status is IN-PROGRESS after --init-only started the run, or when the pipeline
was paused while it ran. pipeline_run_id is None when a paused pipeline started
nothing.
SlaLapse
dataclass
¶
A run that has passed its pipeline's SLA.
RunHooks
dataclass
¶
What other layers do at points of a run.
on_sla_lapse is called once per run, when it is first seen past its SLA.
on_finalized is called after the run has ended. A hook that raises is logged and does
not change how the run ended. A run given no hooks uses default_hooks.
BackfillOutcome
dataclass
¶
How a backfill went: each run it made, and the one that stopped it, if any.
status
property
¶
SUCCESS when every date ran to the end, else the status of the run that stopped.
default_hooks ¶
Return the hooks a run gets unless given others: the SLA email, with Enforce_sla on.
The email goes to the pipeline's EMAIL_RECIPIENTS, else the DAG default recipients,
through the Email settings.
run_pipeline ¶
run_pipeline(engine: Engine, config: ConnectorConfig, pipeline_code: str, **kwargs: Unpack[_PipelineOptions]) -> PipelineOutcome
Run the cooperative local scheduler in the foreground until it settles.
pipeline_steps ¶
pipeline_steps(engine: Engine, config: ConnectorConfig, pipeline_code: str, *, force: bool = False, clock: Clock | None = None, child: ChildOptions | None = None, hooks: RunHooks | None = None, run_date: date | None = None, backfill: str | None = None, selector: RunSelector = runlog.ACTIVE_RUN, stop_dispatch: Event | None = None, graph_data: PipelineGraphData | None = None, pool: ThreadPoolExecutor | None = None) -> Generator[float, None, PipelineOutcome]
Advance ready tasks of pipeline_code cooperatively; local mode only.
force skips the pipeline's gate and runs every task after its upstreams finish, whatever its
status or dependencies. run_date is the date a new run runs as of (today unless given),
and backfill, the reason for one, makes it a backfill run (see backfill). When
interrupted, the running task processes are stopped and recorded FAILED, and the run
stays IN-PROGRESS so the next run resumes it.
backfill ¶
backfill(engine: Engine, config: ConnectorConfig, pipeline_code: str, first: date, last: date, reason: str, *, clock: Clock | None = None, child: ChildOptions | None = None, hooks: RunHooks | None = None) -> BackfillOutcome
Run pipeline_code once for each date from first to last, oldest first.
Each is a run of its own, as of its date (SQL's $$run_date, a script's
task.run_date), marked a backfill run: it checks no dependency on other pipelines,
consumes no upstream run, and its scripts get no offset and store none, so the scheduled
runs go on as before. The backfill stops at the first run that does not end SUCCESS or
SKIPPED; a later backfill from that date takes up again. Local mode only, with a reason,
at most MAX_BACKFILL_DAYS dates, and not while the pipeline has a run in progress.
init_pipeline_run ¶
init_pipeline_run(engine: Engine, config: ConnectorConfig, pipeline_code: str, *, clock: Clock | None = None, hooks: RunHooks | None = None, run_date: date | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> PipelineOutcome
Start or resume the run of pipeline_code, for an orchestrator's first step.
run_date is the date a new run runs as of, such as the orchestrator's logical date.
Tests the run's connections and, for a new run in local mode, checks the pipeline's gate. The
outcome is IN-PROGRESS, or SKIPPED when the gate was not satisfied. In remote mode the
orchestrator applies the pipeline's rules, so RemoteUnsupportedError is raised first when
it has rules the orchestrator does not support, and no gate is checked.
finalize_active_run ¶
finalize_active_run(engine: Engine, config: ConnectorConfig, pipeline_code: str, *, hooks: RunHooks | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> PipelineOutcome
End the active run of pipeline_code from its tasks' statuses, for an orchestrator.
In remote mode this records the orchestrator's decisions (see _record_orchestrated);
in local mode, tasks that can never run are recorded SKIPPED as in a local run. Raises
RunStateError when the pipeline has no active run.
rerun_task ¶
rerun_task(engine: Engine, config: ConnectorConfig, pipeline_code: str, task_code: str, reason: str, *, with_downstream: bool = False, child: ChildOptions | None = None, hooks: RunHooks | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> PipelineOutcome
Run task_code again under the selected run: run --task_code --rerun.
With with_downstream, every task after it runs again too; local mode only. The task
runs as it stands, without its dependencies checked again. Each task after it runs again,
in dependency order, when its dependencies are satisfied by then; the others keep their
rows. A run that had ended is reopened for this and then ended again from its tasks'
statuses; a run still in progress is left for run --pipeline_code to finish.
force_task ¶
force_task(engine: Engine, config: ConnectorConfig, pipeline_code: str, task_code: str, *, child: ChildOptions | None = None, hooks: RunHooks | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> PipelineOutcome
Run task_code past its checks: run --task_code --force; local mode only.
Under the pipeline's active run the task just runs. A run that had ended is reopened for
it and then ended again from its tasks' statuses, so a forced task that fails leaves the
run FAILED.