Skip to content

remote

etl_craft.execution.remote

Remote mode: the orchestrator is the only source of truth for scheduling.

generate-yml writes every scheduling rule into the orchestrator's DAGs, and the engine checks none of them itself: run --task_code runs what it is told. A task's run condition and its dependencies' types become one trigger rule on its upstream steps. A dependency on a task in another pipeline becomes a sensor step that succeeds when the dependency is satisfied, and so does a dependency on another pipeline, unless Global_dag is on and the global DAG triggers the pipelines in dependency order instead. __init__ starts the run before any sensor, so every task the orchestrator runs has a run to bind to.

Some rules have no equivalent in an orchestrator. Rather than drop them, remote mode fails wherever they would be dropped (validate, generate-yml and run --init-only), naming each one with its remedy:

  • run condition N: a trigger rule counts all or one of the upstream steps, never N of them;
  • a HAS_DATA dependency: an orchestrator sees whether a step succeeded, not how many rows it wrote;
  • a task whose dependencies have different types: a trigger rule applies to every upstream step alike (a sensor counts as a SUCCESS dependency, since it succeeds when its dependency is satisfied);
  • with Global_dag on, a pipeline whose dependencies on other pipelines have different types, for the same reason.

TASK_SENSOR_STATES module-attribute 

TASK_SENSOR_STATES: dict[str, tuple[list[str], list[str]]] = {DependencyType.SUCCESS: (['success'], ['failed', 'upstream_failed', 'skipped']), DependencyType.FAILURE: (['failed', 'upstream_failed'], ['success', 'skipped']), DependencyType.ALWAYS: (['success', 'failed', 'upstream_failed', 'skipped'], [])}

For a dependency on another pipeline's task: the upstream task states that satisfy it, and those that never will, as a sensor's allowed_states and failed_states.

DAG_SENSOR_STATES module-attribute 

DAG_SENSOR_STATES: dict[str, tuple[list[str], list[str]]] = {DependencyType.SUCCESS: (['success'], ['failed']), DependencyType.FAILURE: (['failed'], ['success']), DependencyType.ALWAYS: (['success', 'failed'], [])}

For a dependency on another pipeline: the upstream DAG run states that satisfy it, and those that never will.

SENSOR_TYPE module-attribute 

SENSOR_TYPE = DependencyType.SUCCESS

The dependency type a sensor step counts as for the trigger rule of the task after it.

UnsupportedRule dataclass

A rule of a pipeline that a remote orchestrator does not support.

where is PIPELINE or PIPELINE.TASK; rule is the rule as it is configured.

describe 

describe() -> str

Return one line naming the rule, why it fails, and the remedy.

unsupported_rules 

unsupported_rules(conn: Connection, pipeline_id: int, pipeline_code: str, *, global_dag: bool) -> list[UnsupportedRule]

Return every rule of pipeline_id a remote orchestrator does not support.

require_supported 

require_supported(conn: Connection, config: ConnectorConfig, pipeline_code: str) -> None

Raise RemoteUnsupportedError naming every rule of the pipeline the orchestrator lacks.