Workflow Orchestration and Scheduling Questions

Direct answer

A minimal scheduler for retraining jobs needs three pieces working together from a JSON-described dependency graph: a topological sort that resolves a valid execution order, the same sort doubling as cycle detection since a graph with a cycle can never be fully ordered, and a backfill layer that replays that resolved order once per missed date, oldest date first. The sketch below implements exactly this with Kahn's algorithm and explains how to extend it toward distributed execution and retries without changing its core shape.

Structured elaboration

Approach. resolve_order builds a successor list and an in-degree count for every task from the declared dependency pairs, then repeatedly pulls tasks with zero remaining in-degree, in a stable, sorted order for determinism, appending each to the result and decrementing its successors' in-degree. If every task is eventually appended, the order is valid; if any task never reaches zero in-degree, the remaining tasks form a cycle, and resolve_order raises CycleError naming exactly which tasks never resolved, rather than looping forever or silently returning a partial, incorrect order.

Key points. Backfilling is deliberately not a separate scheduling algorithm: backfill calls the exact same resolve_order once per missed date, sorted oldest first, so the DAG's own dependency logic never has two different code paths for "normal run" versus "backfill run." This sketch does not model a depends_on_past-style dependency between different dates' runs of the same task; each date's replay is fully independent of every other date's, which is called out explicitly rather than silently assumed to be handled.

Worked example

"""
Minimal scheduler for retraining jobs: accepts a JSON DAG, resolves execution order via
Kahn's algorithm (which doubles as cycle detection), and supports backfilling missed runs
for specific dates by replaying that same resolved order once per missed date.
"""
import json
from collections import deque


class CycleError(Exception):
    pass


def resolve_order(dag_json):
    """dag_json: {"tasks": [str, ...], "deps": [[upstream, downstream], ...]}"""
    tasks = dag_json["tasks"]
    deps = dag_json["deps"]

    succs = {t: [] for t in tasks}
    indeg = {t: 0 for t in tasks}
    for a, b in deps:
        succs[a].append(b)
        indeg[b] += 1

    queue = deque(sorted(t for t in tasks if indeg[t] == 0))
    order = []
    while queue:
        t = queue.popleft()
        order.append(t)
        for s in sorted(succs[t]):
            indeg[s] -= 1
            if indeg[s] == 0:
                queue.append(s)

    if len(order) != len(tasks):
        unresolved = [t for t in tasks if t not in order]
        raise CycleError(f"cycle detected, tasks never reach indegree 0: {unresolved}")
    return order


def run_dag_for_date(dag_json, run_date, executed_log):
    order = resolve_order(dag_json)
    for task in order:
        executed_log.append((run_date, task))


def backfill(dag_json, missed_dates):
    """Replays the DAG's resolved order once per missed date, oldest date first, so a
    downstream date-partitioned dependency a task might have on 'yesterday's output' is
    honored at the RUN level even though this minimal sketch does not model that dependency
    explicitly (see the answer's extension notes for depends_on_past)."""
    executed_log = []
    for d in sorted(missed_dates):
        run_dag_for_date(dag_json, d, executed_log)
    return executed_log


if __name__ == "__main__":
    dag_json_text = """
    {
      "tasks": ["fetch_features", "validate", "train", "evaluate", "register"],
      "deps": [
        ["fetch_features", "validate"],
        ["validate", "train"],
        ["train", "evaluate"],
        ["evaluate", "register"]
      ]
    }
    """
    dag = json.loads(dag_json_text)

    print("=== resolved order (single run) ===")
    order = resolve_order(dag)
    print(order)
    assert order == ["fetch_features", "validate", "train", "evaluate", "register"]

    print()
    print("=== backfill 3 missed dates, out of order input ===")
    missed = ["2026-07-22", "2026-07-20", "2026-07-21"]
    log = backfill(dag, missed)
    for run_date, task in log:
        print(f"  {run_date}: {task}")
    assert [d for d, _ in log][:5] == ["2026-07-20"] * 5, "must process oldest date first"
    assert len(log) == len(missed) * len(dag["tasks"])
    print("total steps executed:", len(log), "=", len(missed), "dates x", len(dag["tasks"]), "tasks")

    print()
    print("=== cycle detection ===")
    bad_dag = {
        "tasks": ["a", "b", "c"],
        "deps": [["a", "b"], ["b", "c"], ["c", "a"]],
    }
    try:
        resolve_order(bad_dag)
        print("ERROR: should have raised")
    except CycleError as e:
        print("raised CycleError as expected:", e)

Output (actual, from running the block above with python3, stdlib only):

=== resolved order (single run) ===
['fetch_features', 'validate', 'train', 'evaluate', 'register']

=== backfill 3 missed dates, out of order input ===
  2026-07-20: fetch_features
  2026-07-20: validate
  2026-07-20: train
  2026-07-20: evaluate
  2026-07-20: register
  2026-07-21: fetch_features
  2026-07-21: validate
  2026-07-21: train
  2026-07-21: evaluate
  2026-07-21: register
  2026-07-22: fetch_features
  2026-07-22: validate
  2026-07-22: train
  2026-07-22: evaluate
  2026-07-22: register
total steps executed: 15 = 3 dates x 5 tasks

=== cycle detection ===
raised CycleError as expected: cycle detected, tasks never reach indegree 0: ['a', 'b', 'c']

The single-run order matches the linear chain declared in deps exactly. The backfill case is given missed_dates deliberately out of order (2026-07-22, 2026-07-20, 2026-07-21) and the output confirms 2026-07-20 is processed first regardless of input order, proving the sorted(missed_dates) step actually enforces oldest-first replay rather than merely happening to look right. The cycle case, a 3-node cycle with no valid starting point at all, correctly raises CycleError naming all three unresolved tasks, rather than hanging in an infinite loop or silently returning an incomplete order.

Key points, complexity, and edge cases

Complexity: resolve_order is Kahn's algorithm, $O(V + E)$ in the number of tasks and dependency edges, dominated by the sorted insertions into the queue; backfill runs it once per missed date, so a backfill over $m$ dates costs $O(m \times (V + E))$.

Edge cases: an empty tasks list with no deps resolves to an empty order without error, since the queue starts with every zero-in-degree task, all of them, and immediately empties. A task named in deps but not listed in tasks would raise a KeyError from the dictionary comprehensions building succs and indeg, a fail-loud behavior appropriate for a malformed input rather than silently ignoring the dangling reference. The cycle case demonstrates that CycleError correctly reports every unresolved task, not just one, which matters for diagnosing a real malformed DAG definition where identifying the full cycle, not just a single node in it, is what a human actually needs to fix it.

Trade-offs and pitfalls

This scheduler resolves and replays a full DAG once per missed date sequentially; a genuinely distributed extension would need to submit each date's resolved task order to a real worker pool, ideally allowing multiple dates to run their independent stages concurrently rather than one full date finishing before the next starts, since nothing in this dependency structure inherently requires sequential dates to be processed one at a time, only that within a single date, tasks respect their own topological order.

Extending toward real retries needs per-task state (pending, running, succeeded, failed) tracked externally, not just an in-memory executed_log list; a crash partway through a backfill with no persisted state would have to restart every date from scratch.

The explicit choice not to model depends_on_past here is a real scoping decision, not an oversight: adding it would mean a date's train task cannot start until the previous date's train task has succeeded, which changes the scheduler from "N independent per-date replays" into a single larger dependency graph spanning dates, a genuinely different and more complex problem than this sketch solves.

Direct answer

The DAG skeleton below uses Airflow's @dag decorator to declare the DAG-level configuration, its identifier, nightly schedule, start date, and catchup behavior, as a factory function, while each of the seven pipeline stages named in the question, ingest, validate, preprocess, train, evaluate, register_model, and deploy, is a classic PythonOperator instance with a placeholder callable, wired together as a single linear chain with the >> dependency operator. This matches exactly the mix the question asks for: the @dag decorator paired with PythonOperator placeholders, not the newer TaskFlow @task style, which passes data between tasks differently.

Structured elaboration

Approach. The @dag decorator wraps a plain Python function whose body defines the DAG's tasks and their dependencies; calling that decorated function once, nightly_training_pipeline(), produces the actual DAG object. Each of the seven stages is created as its own PythonOperator, all pointed at the same placeholder callable (since the question only asks for the skeleton, not the real implementations), and chained in the exact declared order with a single >> expression.

Key points. Naming each Python variable identically to its task_id string (ingest = PythonOperator(task_id="ingest", ...)) is a deliberate convention here, not an accident, since it keeps the code and the rendered DAG graph in the orchestrator's user interface trivially easy to cross-reference. The placeholder callable inspects context["task"].task_id so a single shared function can still report which stage actually ran, useful for a skeleton meant to be filled in incrementally, one stage at a time, without needing seven separate placeholder functions up front.

Worked example

from datetime import datetime

from airflow.decorators import dag
from airflow.operators.python import PythonOperator


def _placeholder(**context):
    """Stand-in for the task's real logic; only used so the operator has a callable."""
    task_id = context["task"].task_id
    print(f"[placeholder] would run: {task_id}")


@dag(
    dag_id="nightly_training_pipeline",
    schedule="0 2 * * *",       # nightly at 02:00
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["ml", "training"],
)
def nightly_training_pipeline():
    ingest = PythonOperator(task_id="ingest", python_callable=_placeholder)
    validate = PythonOperator(task_id="validate", python_callable=_placeholder)
    preprocess = PythonOperator(task_id="preprocess", python_callable=_placeholder)
    train = PythonOperator(task_id="train", python_callable=_placeholder)
    evaluate = PythonOperator(task_id="evaluate", python_callable=_placeholder)
    register_model = PythonOperator(task_id="register_model", python_callable=_placeholder)
    deploy = PythonOperator(task_id="deploy", python_callable=_placeholder)

    ingest >> validate >> preprocess >> train >> evaluate >> register_model >> deploy


dag_obj = nightly_training_pipeline()

if __name__ == "__main__":
    print("dag_id:", dag_obj.dag_id)
    print("schedule:", dag_obj.timetable.summary)
    print("task_count:", len(dag_obj.tasks))
    order = dag_obj.topological_sort()
    print("topological_order:", [t.task_id for t in order])
    for t in dag_obj.tasks:
        print(f"  {t.task_id}: downstream={sorted(t.downstream_task_ids)}")

Output (actual, from running the block above against a real Apache Airflow 2.11.2 installation):

dag_id: nightly_training_pipeline
schedule: 0 2 * * *
task_count: 7
topological_order: ['ingest', 'validate', 'preprocess', 'train', 'evaluate', 'register_model', 'deploy']
  ingest: downstream=['validate']
  validate: downstream=['preprocess']
  preprocess: downstream=['train']
  train: downstream=['evaluate']
  evaluate: downstream=['register_model']
  register_model: downstream=['deploy']
  deploy: downstream=[]

The topological order matches the declared >> chain exactly, confirming the seven operators are wired as one straight-line dependency chain with no accidental branching, and each task's downstream_task_ids shows exactly one successor apart from the terminal deploy task, which correctly has none.

Key points, complexity, and edge cases

Complexity: building the DAG is $O(n)$ in the number of tasks ($n=7$ here), and topological_sort() runs in $O(V+E)$ time, linear in the number of tasks plus dependency edges, since a straight chain has exactly $n-1$ edges.

Edge cases: catchup=False is a deliberate choice for this pipeline, not a default left untouched: a nightly training pipeline generally should not backfill and retrain once for every historical day between when the DAG was first deployed and today, which is what catchup=True would trigger. If a future stage needs to branch (training several model variants in parallel before evaluation, for instance), the single >> chain shown here would need to become a fan-out and fan-in instead of a straight line, which is a structural change, not just a configuration one. The shared _placeholder callable works correctly for a skeleton precisely because it reads its own task_id from context rather than being hardcoded per task; a real implementation would replace it with seven distinct functions, each doing that stage's actual work.

Trade-offs and pitfalls

A single linear chain of PythonOperator placeholders is the simplest possible skeleton, but it does not pass data between stages: a real ingest task's output would need an explicit hand-off, either through Airflow's cross-communication (XCom) mechanism or an external store such as object storage, since plain PythonOperator return values are not automatically threaded into the next task's arguments the way TaskFlow's @task-decorated functions do. The question explicitly asks for PythonOperator placeholders rather than TaskFlow, so that data-passing convenience is intentionally left out of this skeleton, not overlooked.

Keeping every operator's Python variable name identical to its task_id string, as done here, is a small discipline that pays off the moment the DAG grows past a handful of tasks; letting the two drift (a variable named preprocess_step with task_id="preprocess") is a common, easy-to-introduce source of confusion when reading the rendered graph against the code side by side.

Direct answer

The architectural difference underneath every practical difference between the two: Airflow models a pipeline as a graph of tasks (units of execution), while Dagster models it as a graph of assets (the actual data objects a pipeline produces), with tasks as the mechanism that materializes them. That single choice ripples into developer experience, observability, testing, and deployment. Recommend Airflow when the team's mental model is genuinely task-centric (orchestrate arbitrary steps, many of which are not really "data assets" per se) and when the ecosystem's breadth of existing integrations matters most; recommend Dagster when the pipeline's real unit of value is the data it produces, and being able to reason about, test, and monitor at that level pays off, which is common for feature pipelines and other ML-adjacent data products.

Structured elaboration

Developer experience, in more depth. Airflow's DAG-as-Python-file model is simple to start with and has an enormous operator ecosystem (a pre-built integration for nearly any system a pipeline might need to talk to), but that same breadth means operator quality and API consistency vary across providers, since they are maintained by many different contributors. Dagster's asset-first API is more opinionated and type-aware from the start, which raises the initial number of concepts a new user has to learn (assets, ops, resources, jobs, schedules, sensors are all distinct, related concepts), but tends to produce more consistent, more testable code once a team is past that initial learning curve.

Observability, in more depth. Airflow's UI is built around DAG runs and task instances: it answers "did this task succeed, and when." Dagster's UI is built around the asset graph: it answers "is this specific piece of data fresh, and what upstream asset is stale or missing if it's not," which is a more directly useful question for a data consumer who cares about a specific table or feature set, not about which internal tasks happened to produce it.

Dataset/type awareness. This is the core architectural distinction stated above, made concrete: in Dagster, an asset can carry a declared type, and Dagster can check that what a downstream asset receives actually matches what it expects, catching a class of bug (a schema or shape mismatch between pipeline stages) that Airflow's task-centric model has no native mechanism to catch, since Airflow tasks pass data through XCom or external storage with no built-in typing.

Testing support. Both frameworks let you unit-test the business logic inside a task or op as plain Python, which is the majority of what actually needs testing. The difference is at the pipeline-structure level: Dagster's asset graph is a first-class object that can be validated and partially executed in tests (materialize just this asset and its direct dependencies, in isolation), while testing an Airflow DAG's structure (are the dependencies wired correctly, does the schedule behave as expected) typically requires more custom test scaffolding, since the DAG object itself is not designed around being partially executed for testing.

Deployment model. Airflow has the larger, more mature ecosystem of both managed offerings and self-hosted deployment patterns, reflecting its longer track record and larger install base; teams needing a specific compliance posture, a specific cloud provider's managed service, or deep operational familiarity already present on the team often lean this way by default. Dagster's deployment ecosystem (Dagster Cloud, plus self-hosted) is smaller but has matured significantly, and Dagster's software-defined-assets model is a genuine design differentiator, not just a smaller Airflow.

Worked example

A general-purpose ETL platform team, many source systems, mostly straightforward extract/load/transform sequencing, existing Airflow expertise on the team. Recommend Airflow: the pipeline's actual complexity is in sequencing many heterogeneous sources, which Airflow's operator ecosystem directly supports, and the team's existing operational familiarity with Airflow is a real, non-trivial advantage that a rewrite in a different tool would spend real time and risk to obtain only a marginal architectural benefit for a workload that is not particularly asset-model-shaped.

A machine learning feature pipeline: raw events to engineered features to a feature store, consumed by multiple downstream training jobs, where "is this feature fresh and correctly typed" is the actual question data scientists ask daily. Recommend Dagster: the pipeline's real unit of value is explicitly the data assets (each engineered feature), not the tasks that produce them, and Dagster's asset-centric observability directly answers the question data scientists actually have ("is feature_x fresh as of today, and does it match the schema training expects") without them needing to translate from "which tasks succeeded" to "is my feature usable." The type-awareness also catches a real, common ML pipeline failure mode early: a feature whose shape or type silently drifted between the pipeline and what a training job expects, which Dagster's typed asset graph can surface as a build-time or run-time check rather than a downstream training failure discovered much later.

Trade-offs and pitfalls

Choosing Dagster purely because "asset-based" sounds like the more modern architecture, for a pipeline that is genuinely task-shaped (many heterogeneous, loosely-related steps with no strong shared data-asset identity), adds conceptual overhead without a corresponding benefit; the asset model earns its complexity specifically when the pipeline's actual product is a well-defined set of data assets, not universally.

Staying on Airflow purely out of inertia for a pipeline that is genuinely asset-shaped (a feature pipeline, a data product with many downstream consumers who care about freshness and type correctness) means building, by hand, observability and validation that Dagster provides natively, which is a real, recurring engineering cost paid on every such pipeline, not a one-time migration cost avoided.

Finally, underestimating Airflow's ecosystem breadth when evaluating Dagster is a common mistake in the other direction: a source system with a mature, well-tested Airflow provider but no equivalent first-class Dagster integration can mean building custom integration code in Dagster that Airflow would have gotten for free, which is a real cost that should be weighed against the asset-model benefits on a case-by-case basis, not assumed away.

Direct answer

A task is idempotent if running it multiple times with the same input produces the same end state as running it exactly once: no duplicated rows, no double-counted totals, no repeated side effects. An orchestrator depends on this because its entire safety net, retries, replays, and backfills, works by re-executing a task, sometimes more than once for the same logical unit of work; if a task is not idempotent, every one of those safety mechanisms becomes a correctness risk instead of a recovery tool. Turning a non-idempotent write into an idempotent one is usually a matter of replacing "add to what's there" with "set what should be there," and validating it in CI means actually running the task twice against the same input and asserting the output is identical both times, not just asserting it by inspection.

Structured elaboration

What idempotency means precisely for an ETL task. Given the same input data and the same logical run identifier (the same logical date, the same partition), executing the task 1 time or 5 times leaves the target system in the same observable state. This is a property of the task's write behavior specifically; a task that only reads and computes, with no external side effect, is trivially idempotent, since re-running it just recomputes the same answer without touching anything.

Why the orchestrator depends on it. Three of the orchestrator's core mechanisms all boil down to "run this task again":

• Retries. A task that fails partway through (after some side effect already happened) gets re-executed automatically. If the write is not idempotent, the retry compounds the partial side effect from the failed attempt instead of correcting it.
• Replays, meaning a manual rerun of a specific task instance, for example after diagnosing and fixing a bug, without waiting for a full backfill.
• Backfills. Re-running a historical logical interval, sometimes because the interval genuinely never ran, but often because it DID run and produced wrong output that needs replacing, not adding to.
  In every one of these cases, the orchestrator has no way to know or enforce whether the underlying write is safe to repeat; it can only trigger the task again. Idempotency is what makes "trigger it again" a safe, general-purpose recovery mechanism instead of something that has to be reasoned about case by case, task by task, every time something goes wrong.

Turning a non-idempotent write into an idempotent one. A common non-idempotent shape is an append-only insert: a task that runs INSERT INTO daily_totals (date, region, total) VALUES (...) for each computed row. Run it twice for the same date, and the table now has two rows for that date and region, silently doubling anything downstream that sums the table. The idempotent version replaces that pattern with one of two mechanisms:

1. Delete-then-insert, scoped to the exact logical unit being written. Before inserting, delete any existing rows for that specific (date, region) key, then insert fresh. Running it N times leaves exactly the rows from the last run, since each run first clears its own prior output.
2. Upsert (insert-or-replace) keyed on the natural identifier. INSERT ... ON CONFLICT (date, region) DO UPDATE SET total = EXCLUDED.total (or the equivalent MERGE statement) achieves the same end state in one statement, without a separate delete step, and is generally preferred when the underlying table supports it, since it avoids a window where the delete has happened but the insert has not.
   Both approaches share the same principle: the write is scoped to overwrite exactly the rows this task run is responsible for, identified by a stable key, rather than blindly adding new rows every time.

Trade-offs the fix introduces. Delete-then-insert needs the delete's scope to be exactly right; too narrow (missing a key column) leaves stale rows behind after a schema or logic change, too broad (deleting more than this run's own output) can destroy another task's data if two tasks share the same table. Both delete-then-insert and upsert require a stable natural key to scope or conflict on, which is not always available for genuinely append-only event data (a raw event log with no natural deduplication key); in that case, idempotency has to be achieved differently, typically by making the destination path or partition itself unique per run (so a rerun overwrites the same file/partition rather than needing row-level conflict resolution) rather than by upserting individual rows. There is also a performance cost: an upsert or a scoped delete-then-insert is generally more expensive than a blind append, since the database has to check for existing matches, which matters at high write volume.

Validating idempotency in CI. Rather than reasoning about it by reading the code, run the task twice against the same fixed input in a test database or table and assert the resulting state is identical after both runs, not just that the second run did not error. Concretely: seed a test table, run the write function once, capture the full resulting state (row count, and ideally a hash or sorted dump of the actual row contents, not just the count, since a delete-then-insert bug that leaves the row count right but the values stale would pass a count-only check), run the write function a second time with the same input, and assert the captured state is unchanged. This is a stronger test than checking the function does not raise on a second call, since a non-idempotent function can run successfully twice while still producing wrong (duplicated or doubled) data.

Worked example

Take a preprocessing step in a feature pipeline that computes a daily aggregate and writes it to a daily_features table, originally implemented as a non-idempotent append:

Non-idempotent version: the task computes feature_value for (user_id, feature_date) and executes a plain INSERT for every row. Run once: 10,000 rows, one per user for that date. Retried after a transient failure partway through a 10,000-row batch (say, 6,000 rows already inserted before the failure): the retry reruns the whole computation and inserts all 10,000 rows again, so the 6,000 users caught in the partial first attempt now have 2 rows each and the other 4,000 have 1, a silently inconsistent table where a naive AVG(feature_value) for those 6,000 users is now computed from duplicated identical values (which happens not to change the average, but does change any query that instead uses COUNT or SUM, both of which are now wrong for exactly the users caught in the partial retry).

Idempotent version: the write becomes INSERT INTO daily_features (user_id, feature_date, feature_value) VALUES (...) ON CONFLICT (user_id, feature_date) DO UPDATE SET feature_value = EXCLUDED.feature_value. The same retry scenario now: the first, partial attempt inserts 6,000 rows; the retry recomputes and writes all 10,000, of which the first 6,000 hit the ON CONFLICT path and are updated in place (not duplicated), and the remaining 4,000 are freshly inserted. End state after the retry: exactly 10,000 rows, one per user, matching what a single clean run would have produced, regardless of exactly where the original attempt failed.

CI validation for this task: seed a test database with a small fixture of 50 users, run the write function once, capture a sorted dump of (user_id, feature_date, feature_value) for all rows, run the write function a second time with the identical input, capture the same dump again, and assert the two dumps are byte-identical and the row count is still 50. This test would fail loudly against the non-idempotent version (100 rows the second time, dump mismatch) and pass against the upsert version (50 rows both times, identical dump), which is exactly the distinction a row-count-only or "did it error" check would miss.

Trade-offs and pitfalls

The most common mistake is validating idempotency by re-reading the code and reasoning "this looks like it should be safe" rather than actually running it twice; a delete-then-insert whose delete clause is missing one key column looks correct on a read-through and only fails when actually exercised with a second run against real data.

A second common mistake is testing idempotency by checking that the row count is unchanged after a second run, without checking the row values; a bug where the delete step correctly removes stale rows but the insert step recomputes with a subtly different formula on the second pass produces the right count with wrong values, and a count-only test gives false confidence.

Finally, upsert-based idempotency assumes the natural key used for conflict resolution is genuinely stable and unique for the task's logical scope; if two different upstream sources can legitimately both produce a row for the same (user_id, feature_date) key for different reasons, an upsert silently overwrites one with the other instead of raising the conflict as an error, which trades away a signal that might have indicated two pipelines are stepping on the same output when they should not be.

Direct answer

Running short extract-transform-load (ETL) jobs, medium Spark jobs, and heavy graphics-processing-unit (GPU) model training on one Kubernetes cluster under Airflow requires the cluster's scheduler, not just Airflow, to recognize these as fundamentally different resource shapes. That means separate node pools sized to each workload's real profile, priority classes that protect critical work from being starved by cheaper and more numerous jobs, and preemption applied only where the workload can actually tolerate it, so the cluster can burst without permanently paying for idle headroom.

Structured elaboration

Node pools. At minimum three pools matching the three workload shapes: a general-purpose pool of many small nodes for short ETL tasks; a memory- and central-processing-unit-heavier pool for medium Spark jobs; and a GPU-equipped pool reserved for heavy training. GPU nodes are the cluster's most expensive capacity, so a short ETL task accidentally scheduled onto one wastes it on the cheapest workload in the mix. Enforce this with Kubernetes node taints on the GPU pool and matching tolerations only on GPU-requesting pods, a structural guarantee rather than a scheduling convention that can silently be violated.

Priority classes. Define at least a critical, standard, and best-effort tier, assigned per Airflow task or DAG according to actual business impact. When capacity is contended, the scheduler prefers admitting a critical pod over a best-effort one, and can preempt a lower-priority pod if that is the only way to schedule a higher-priority one.

Preemption and spot-like instances. Use preemptible or spot instances for the ETL and Spark pools, workloads that are naturally retryable at the orchestrator level, since Airflow already retries a failed task automatically. Avoid spot capacity for the GPU training pool by default: losing hours of an in-progress training run to preemption is far more costly than losing a short ETL task, unless the training job itself checkpoints its own progress. If GPU spot capacity is used anyway for cost reasons, pair it explicitly with the training job's own checkpointing, since Airflow simply retrying the whole task would mean restarting the entire run from scratch.

GPU scheduling. Request GPUs through Kubernetes' nvidia.com/gpu resource requests and limits on the pod specification, for example through a KubernetesPodOperator. Size the GPU pool's autoscaler minimum above zero if training happens often enough that scaling up from a true zero adds unacceptable startup latency, and bound the maximum by actual budget. Avoid GPU fragmentation, a job requesting one GPU landing on an eight-GPU node and stranding the other seven, through bin-packing-aware scheduling or by dedicating specific node sizes to the pool's typical GPU-count requests.

Preventing starvation of critical tasks. Combine priority classes, so critical work can preempt when genuinely necessary, with a reserved minimum capacity carved out per pool specifically for critical workloads, capacity that best-effort work cannot consume even when it looks idle. "Looks idle" can flip to "needed" faster than a running best-effort pod can be evicted and rescheduled elsewhere, so the reservation, not preemption alone, is what actually guarantees availability in the common case.

Worked example

The GPU pool has 4 nodes with 8 GPUs each, 32 GPUs total. A critical fraud-model retraining job needs 4 GPUs and must start within 10 minutes of being triggered. A best-effort batch of 20 experimental training jobs, each requesting 2 GPUs, 40 GPUs of total demand against a 32-GPU pool, is already running when the critical job arrives.

A reserved-quota carve-out sets aside 4 GPUs specifically for critical-priority pods, so at most:

$32 - 4 = 28 \text{ GPUs}$

are ever schedulable to best-effort work, guaranteeing the critical job's 4-GPU request always has room without needing preemption in the common case. If those reserved 4 GPUs somehow end up occupied by a stuck best-effort job, since a quota alone does not prevent misconfiguration, priority-based preemption is the second line of defense, evicting the lowest-priority pods holding GPUs specifically within that reserved range.

Trade-offs and pitfalls

Reserving capacity for critical workloads means that capacity sits partially idle whenever no critical job is actually running, a direct cost paid for guaranteed availability, not a free guarantee.

Preempting a best-effort GPU job that had already run for 3 hours of a planned 4-hour training run wastes that entire 3 hours unless the job checkpointed its own progress; preemption policy and the preempted workload's checkpointing discipline have to be designed together, not assumed to work out independently.

Mixing spot or preemptible capacity into the same priority tier as steady, on-demand capacity can produce a surprising failure mode: a critical pod landing on a spot node that then gets reclaimed by the cloud provider itself, entirely outside Kubernetes' own priority and preemption model. Critical workloads should generally run on stable, non-spot capacity regardless of how their Kubernetes priority is configured.