Running Data Workloads on Kubernetes

Consider a hypothetical nightly job that loads the previous day’s orders into a warehouse table. On one server it is a cron entry and a script. On a Kubernetes cluster the same program becomes a container image that the platform starts on a node it chooses, may stop when that node is drained, and may start again after a failure. The platform supplies scheduling, retries, and isolation. It also supplies new ways for the program to run twice, to be killed without writing anything to its own log, or to wait unscheduled while its input ages.

Three agreements decide whether such a job is dependable. The first is what Kubernetes promises about starting and restarting the work, which is set by the Pod, Job, and CronJob objects. The second is what the manifest declares about the resources the work needs, because those numbers decide both placement and the way the job fails. The third is what the program itself does so that a second start leaves the same data behind as one start.

The job is assumed to be packaged as an image, as in Containerizing Data Jobs with Docker, and the declared-state and scheduler model from Kubernetes Controllers and Workload Lifecycle applies throughout. That earlier article centers on long-running services. This one covers the run-to-completion side. The manifests and kubectl commands below are illustrative and were not run against a cluster. The Python programs model the documented rules and were run, and their output is shown as obtained.

A run-to-completion workload is not a service

A Deployment keeps a number of Pods alive and replaces any that exit. A job wants the opposite: the process should exit with status 0 when its work is done, and the platform should treat that exit as success rather than as a crash to repair. The Kubernetes object for this is the Job. Its controller creates Pods from a template, watches how they end, creates replacements for Pods that fail, and records the Job as complete when the required number of Pods have succeeded. A Job with the default completion count of one is satisfied by one successful Pod.

The following manifest is a Job for one logical day of the order load. It is not run here. The name carries the day, because the Job is created for a specific piece of input.

apiVersion: batch/v1
kind: Job
metadata:
  name: orders-load-2024-10-26          # one Job per logical day; the name carries the input identity
spec:
  backoffLimit: 3                       # up to 3 retries after the first Pod, so at most 4 Pods
  activeDeadlineSeconds: 7200           # wall-clock budget for the whole Job, retries and waits included
  ttlSecondsAfterFinished: 86400        # delete the Job and its Pods one day after it finishes
  podFailurePolicy:
    rules:
    - action: FailJob                   # exit code 64 is this program's "input is invalid, do not retry"
      onExitCodes:
        containerName: load
        operator: In
        values: [64]
    - action: Ignore                    # preemption, eviction and node drain do not consume retries
      onPodConditions:
      - type: DisruptionTarget
        status: "True"
  template:
    spec:
      restartPolicy: Never              # a failed container is not restarted inside this Pod
      terminationGracePeriodSeconds: 60
      containers:
      - name: load
        image: registry.example.com/data/orders-load@sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
        args: ["--day", "2024-10-26", "--dest", "s3://example-lake/orders/day=2024-10-26/"]
        resources:
          requests:
            cpu: 500m
            memory: 2Gi
            ephemeral-storage: 4Gi
          limits:
            memory: 2Gi                 # same as the request: memory cannot be borrowed back
            ephemeral-storage: 8Gi      # no CPU limit in this example

Each field answers a separate question about the Pod that will run.

• restartPolicy: Never says what the kubelet does when the container exits non-zero. With Never the failed container is not restarted inside the same Pod. The Pod ends in the Failed phase and the Job controller decides whether to create another Pod. With OnFailure the kubelet restarts the container inside the same Pod and the Job counts those restarts. A Job accepts only these two values, and a podFailurePolicy requires Never.
• backoffLimit bounds replacement. The Kubernetes documentation describes it as the number of retries before the Job is considered failed, with a default of 6, so a limit of 3 allows the first Pod plus three retries. Replacement Pods are created after an exponential back-off delay of 10 seconds, 20 seconds, 40 seconds, and so on, capped at six minutes.
• activeDeadlineSeconds bounds the whole Job. It applies to the duration of the Job regardless of how many Pods were created, and the documentation states that it takes precedence over backoffLimit.
• podFailurePolicy classifies failures. The first rule fails the Job at once when the container exits with code 64, a code this hypothetical program uses for invalid input, because a retry cannot fix bad input. The second rule ignores failures caused by disruptions such as preemption, eviction, or a node drain, so that they do not consume the retry budget.
• ttlSecondsAfterFinished deletes the finished Job, and with it the Pods, after the stated time. The Pods’ logs go with them.

The documentation gives one more fact that shapes the rest of the article. Even with parallelism and completions both set to 1 and restartPolicy: Never, the same program may sometimes be started twice. A Job is a retry mechanism with at-least-once start semantics, not a transaction. The glossary entry on Kubernetes Job and CronJob states the same boundary: the controller’s state describes how execution is managed, and says nothing about whether the destination holds the right data.

One consequence is easy to miss. Kubernetes judges a Pod by its exit status. A program that logs an error, writes nothing, and then exits with status 0 produces a successful Pod and a complete Job. The program has to exit non-zero when its work failed, and the check that the output is complete belongs inside the program or in a validation step after it.

The Pod as one execution attempt

Each retry is a new Pod, so it helps to be precise about what a Pod is. A Pod holds one or more containers that are scheduled together onto one node. The containers share a network namespace, which means one IP address and one port space, and they reach one another on localhost. Two containers in a Pod therefore cannot both listen on the same port. The containers do not share a filesystem: each starts from its own image, and files written in one container’s writable layer are invisible to the others. The mechanisms behind these facts, namespaces and mounts, are covered in Containers: Images, Isolation, and Resource Boundaries, and the glossary entry on the Kubernetes Pod gives the short version.

Containers that must exchange files share a volume. The simplest one for batch work is emptyDir, which is created empty when the Pod is assigned to a node and can be mounted at the same or different paths in each container. The documentation notes that a container crash does not remove the data, because the Pod stays on the node, and that the data is deleted permanently when the Pod is removed for any reason. That makes emptyDir a good scratch area and a poor place for output.

The next manifest shows when putting several containers in one Pod is justified. It is also not applied to a cluster. The Job enriches orders using a read replica of the orders database and a reference file set.

apiVersion: batch/v1
kind: Job
metadata:
  name: orders-enrich-2024-10-26
spec:
  backoffLimit: 2
  template:
    spec:
      restartPolicy: Never
      volumes:
      - name: scratch
        emptyDir:
          sizeLimit: 4Gi                # counts against the Pod's local ephemeral storage
      initContainers:
      - name: fetch-reference           # runs to completion before the others start
        image: registry.example.com/data/ref-fetch:2.0.1
        volumeMounts:
        - {name: scratch, mountPath: /scratch}
      - name: db-proxy                  # restartPolicy: Always makes this init container a sidecar
        image: registry.example.com/platform/db-proxy:1.8.0
        restartPolicy: Always
        args: ["--listen", "127.0.0.1:5432", "--upstream", "orders-replica.internal:5432"]
      containers:
      - name: enrich
        image: registry.example.com/data/orders-enrich:3.1.0
        env:
        - {name: DB_URL, value: "postgresql://127.0.0.1:5432/orders"}   # localhost reaches the sidecar
        - {name: REFERENCE_DIR, value: /scratch/reference}
        volumeMounts:
        - {name: scratch, mountPath: /scratch}
        resources:
          requests: {cpu: "1", memory: 3Gi}
          limits: {memory: 3Gi}

Three roles appear. The init container fetch-reference runs to completion before anything else starts, and it leaves files in the shared scratch volume. The container db-proxy is declared under initContainers with restartPolicy: Always, which Kubernetes treats as a sidecar: it starts before the main container and keeps running next to it. Sidecar containers have been stable since Kubernetes v1.33. The documentation states that a sidecar in a Job’s Pod does not prevent the Job from completing. The main container enrich connects to 127.0.0.1:5432 and reaches the proxy because both share the Pod’s network namespace.

These containers share a Pod because they need what a Pod provides: the same node, the same lifetime, a local address, and a shared scratch directory. Stages of a pipeline usually need none of that. An extract stage and a load stage have different sizes, different failure behavior, and different retry needs, and the output of the first should survive the second failing. Putting both in one Pod ties their requests together and makes a failure in either replace both. The scheduler places a Pod by its effective request, which is the sum over its app containers, not less than the largest init container’s request. A Pod that bundles a 6 GiB memory stage with a 500 MiB stage reserves the sum on one node for the whole run. Separate stages communicate through durable storage or an orchestrator, and each is a Job of its own.

Restarting a container is not replacing a Pod

Both words, restart and retry, describe different events in this model, and the Pod’s contents depend on which one happened.

Event	What the system does	emptyDir contents	Where it runs	Effect on the Job
Container exits non-zero, restartPolicy: OnFailure	The kubelet restarts the container in the same Pod.	Kept.	Same node.	Container restarts count toward backoffLimit.
Container exits non-zero, restartPolicy: Never	The Pod ends as Failed. The Job controller creates a new Pod after the back-off delay.	Lost with the old Pod.	Possibly another node.	One Failed Pod counts toward backoffLimit.
Container killed for exceeding its memory limit	Same as the two rows above, according to restartPolicy. The container status shows reason OOMKilled, usually with exit code 137.	As above.	As above.	Counted as a failure.
Node drained, Pod evicted or preempted	The Pod is terminated with a grace period and the Job controller creates a replacement.	Lost.	Another node.	Counted, unless a podFailurePolicy rule ignores DisruptionTarget.

Neither a restart nor a replacement proves that the previous attempt made no external write. The old container may have written half of its output before it was killed, and it may still be running if the kubelet could not reach it. Treat the Pod as a boundary around one attempt, and keep anything that must outlive the attempt, such as output, progress, and the evidence needed to replay, in storage with its own lifecycle. The glossary entry on graceful shutdown covers the termination side.

Reading a Pod’s state

When a job misbehaves, the Pod’s conditions and each container’s status say different things, and both are needed. The phase and conditions describe the Pod (scheduled, initialized, containers ready). The container statuses describe each container’s current state, its last terminated state with a reason and exit code, and its restart count. These commands show them.

kubectl get pod <pod> -o wide                      # phase, node, restarts
kubectl describe pod <pod>                         # conditions, container states, last state, events
kubectl get pod <pod> -o jsonpath='{.status.containerStatuses[*].lastState}'
kubectl logs <pod> --previous                      # logs of the previous container instance, if it restarted
kubectl get events --field-selector involvedObject.name=<pod>

What you see	Where the evidence is	What it means
Pending, event FailedScheduling with insufficient cpu or insufficient memory	Events from describe pod	No node has enough unreserved requests. Nothing has run yet, and the Job’s clocks are already ticking.
Init status Error or CrashLoopBackOff	initContainerStatuses	An init container failed. With restartPolicy: Never on the Pod, a failed init container makes the whole Pod fail.
Container terminated, reason OOMKilled, exit code 137	lastState.terminated	The memory limit was exceeded. The program was killed without a chance to log.
Pod Failed, reason Evicted	Pod status and node events	The node ran short of memory or disk, or the Pod exceeded its ephemeral-storage limit. This is a different event from an OOM kill inside a surviving Pod.
Job Failed, reason DeadlineExceeded or BackoffLimitExceeded	kubectl describe job conditions	A Job-level clock or counter ended the Job. The Pods may all have been healthy individually.

The time limits on a scheduled Job

A scheduled load involves several time limits that are easy to confuse because each is a number of seconds. They belong to different objects and answer different questions.

Setting	Belongs to	Question it answers	What it does not do
startingDeadlineSeconds	CronJob	Is a late scheduled start still worth starting?	Limit how long a started Job runs.
concurrencyPolicy	CronJob	What happens when the previous Job of this CronJob is still running (Allow, Forbid, or Replace)?	Lock a table, or constrain another CronJob or a manually created Job.
backoffLimit and the back-off delay	Job	How many replacement Pods may be created, and how long to wait before each?	Bound total time. Seven failed Pods with long delays can take far longer than the Pods’ own runtimes.
activeDeadlineSeconds	Job	How long may the Job be active, whatever number of Pods it uses?	Undo output already written. It ends the Job even when retries remain.
terminationGracePeriodSeconds	Pod	How long between the TERM signal and the KILL signal?	Count as a Job retry or extend the Job’s deadline.
ttlSecondsAfterFinished, successfulJobsHistoryLimit, failedJobsHistoryLimit	Job, CronJob	How long do finished Jobs and their Pods remain visible?	Retain business data or output. It deletes the Pods’ logs with them.

Retries and deadlines on a Job

The following program models the Job rules for restartPolicy: Never as documented: a Job fails when the number of failed Pods exceeds backoffLimit, replacement waits follow the 10-second doubling delay capped at 360 seconds, and a Job whose active time reaches activeDeadlineSeconds is failed. It simulates those rules and does not contact a cluster. The scenarios use invented runtimes.

"""Model of the Job retry and deadline rules documented for Kubernetes Jobs (restartPolicy: Never).
It simulates the documented rules; it does not talk to a cluster."""

def backoff_delay(failures_so_far):
    # documented: exponential back-off delay 10s, 20s, 40s ... capped at six minutes
    return min(10 * 2 ** (failures_so_far - 1), 360)

def run_job(name, backoff_limit, deadline, attempts):
    """attempts: list of (runtime_seconds, outcome) the program would have, outcome 'ok' or 'fail'."""
    t, failures, n = 0, 0, 0
    print(f"{name}: backoffLimit={backoff_limit}, activeDeadlineSeconds={deadline if deadline else "unset"}")
    while True:
        runtime, outcome = attempts[min(n, len(attempts) - 1)]
        n += 1
        start, end = t, t + runtime
        if deadline is not None and end > deadline:
            print(f"  pod {n}: start {start:>4}s, still running at {deadline}s -> killed")
            print(f"  Job Failed: DeadlineExceeded at {deadline}s after {failures} failed pod(s)")
            return
        if outcome == "ok":
            print(f"  pod {n}: start {start:>4}s, end {end:>4}s, succeeded")
            print(f"  Job Complete at {end}s")
            return
        failures += 1
        print(f"  pod {n}: start {start:>4}s, end {end:>4}s, failed (failed pods: {failures})")
        if failures > backoff_limit:
            print(f"  Job Failed: BackoffLimitExceeded at {end}s")
            return
        wait = backoff_delay(failures)
        print(f"           controller waits {wait}s before the replacement pod")
        t = end + wait

run_job("A", 3, None, [(120, "fail"), (200, "fail"), (300, "ok")])
run_job("C", 3, 900, [(120, "fail"), (200, "fail"), (300, "fail"), (300, "ok")])
run_job("D", 6, None, [(5, "fail")])

Program output:

A: backoffLimit=3, activeDeadlineSeconds=unset
  pod 1: start    0s, end  120s, failed (failed pods: 1)
           controller waits 10s before the replacement pod
  pod 2: start  130s, end  330s, failed (failed pods: 2)
           controller waits 20s before the replacement pod
  pod 3: start  350s, end  650s, succeeded
  Job Complete at 650s
C: backoffLimit=3, activeDeadlineSeconds=900
  pod 1: start    0s, end  120s, failed (failed pods: 1)
           controller waits 10s before the replacement pod
  pod 2: start  130s, end  330s, failed (failed pods: 2)
           controller waits 20s before the replacement pod
  pod 3: start  350s, end  650s, failed (failed pods: 3)
           controller waits 40s before the replacement pod
  pod 4: start  690s, still running at 900s -> killed
  Job Failed: DeadlineExceeded at 900s after 3 failed pod(s)
D: backoffLimit=6, activeDeadlineSeconds=unset
  pod 1: start    0s, end    5s, failed (failed pods: 1)
           controller waits 10s before the replacement pod
  pod 2: start   15s, end   20s, failed (failed pods: 2)
           controller waits 20s before the replacement pod
  pod 3: start   40s, end   45s, failed (failed pods: 3)
           controller waits 40s before the replacement pod
  pod 4: start   85s, end   90s, failed (failed pods: 4)
           controller waits 80s before the replacement pod
  pod 5: start  170s, end  175s, failed (failed pods: 5)
           controller waits 160s before the replacement pod
  pod 6: start  335s, end  340s, failed (failed pods: 6)
           controller waits 320s before the replacement pod
  pod 7: start  660s, end  665s, failed (failed pods: 7)
  Job Failed: BackoffLimitExceeded at 665s

Scenario A shows an ordinary recovery: two failures, waits of 10 and 20 seconds, and a third Pod that succeeds at 650 seconds. Scenario C has the same failures plus a fourth Pod that would need 300 more seconds. The deadline of 900 seconds arrives while that Pod is running, and the Job fails with DeadlineExceeded after three failed Pods. The retry budget was not used up, and the deadline won, as the documentation says it will. Scenario D uses the default backoffLimit of 6 with a program that fails after 5 seconds every time. Seven Pods run for 35 seconds in total, but the Job fails only at 665 seconds, because the waits grow from 10 seconds to 320 seconds. A fast-failing program with generous retries can therefore keep a Job retrying for more than ten minutes before the Job reports failure, and the deadline is the control that bounds that time.

Choose the two limits together. backoffLimit expresses how many independent failures the work tolerates. activeDeadlineSeconds expresses when a result stops being useful, for example when the output is needed before a morning report. A deadline shorter than the expected runtime plus the waits for the allowed retries means some retries can never happen, so make that choice on purpose.

Starting late, and starting twice, on a CronJob

A CronJob creates Jobs on a schedule. The documentation states that it creates a Job approximately once per scheduled time, and that in some circumstances two Jobs might be created or none, so the Jobs it defines should be idempotent. Two settings control what happens when a scheduled time cannot be honored on time. If startingDeadlineSeconds is set, the controller measures the time between when a Job was expected and now, and skips the run when the difference exceeds the limit. Separately, the controller counts how many schedules were missed since the last one, or only within the last startingDeadlineSeconds if that is set, and when more than 100 were missed it does not start the Job and logs too many missed start times. The documentation illustrates this with a CronJob that runs every minute, a controller outage from 08:29 to 10:21, and no deadline: more than 100 schedules were missed, so nothing starts. With a deadline of 200 seconds the same outage is followed by a start at 10:22.

The program below models those eligibility rules, plus concurrencyPolicy: Forbid. It follows the documented rules, simplifies the controller’s behavior to a single Job on recovery, and does not contact a cluster.

"""CronJob start-eligibility rules as the Kubernetes documentation states them (a model, not the controller's code):
 - a schedule time is eligible if startingDeadlineSeconds is unset, or now - scheduled <= startingDeadlineSeconds;
 - missed schedules are counted from the last scheduled time (or only the last startingDeadlineSeconds if it is set);
   more than 100 missed schedules means no Job is started."""
from datetime import datetime, timedelta

def schedule_times(first, every, upto):
    t = first
    while t <= upto:
        yield t
        t += every

def on_recovery(first, every, last_ok, now, deadline):
    """Controller was down; at `now` it looks at what it missed after `last_ok`."""
    missed = [t for t in schedule_times(first, every, now) if t > last_ok]
    window = missed if deadline is None else [t for t in missed if (now - t).total_seconds() <= deadline]
    if len(window) > 100:
        return f"missed={len(window)} > 100 -> no Job started ('too many missed start times')"
    if not window:
        return f"missed={len(missed)}, none within the deadline -> all skipped, wait for next schedule"
    newest = max(window)
    return f"missed={len(missed)}, eligible={len(window)} -> starts one Job for scheduled time {newest:%H:%M}"

d = lambda h, m: datetime(2024, 10, 27, h, m)
# The documentation's example: every minute from 08:30, controller down 08:29-10:21.
print("every minute, no deadline    :", on_recovery(d(8, 30), timedelta(minutes=1), d(8, 29), d(10, 22), None))
print("every minute, deadline 200 s :", on_recovery(d(8, 30), timedelta(minutes=1), d(8, 29), d(10, 22), 200))
# A daily-style example on an hourly schedule: controller down 01:05-06:20.
print("hourly, down 5 h, no deadline:", on_recovery(d(1, 0), timedelta(hours=1), d(0, 59), d(6, 20), None))
print("hourly, deadline 2 h         :", on_recovery(d(1, 0), timedelta(hours=1), d(0, 59), d(6, 20), 7200))
print("hourly, deadline 10 min      :", on_recovery(d(1, 0), timedelta(hours=1), d(0, 59), d(6, 20), 600))

# concurrencyPolicy: Forbid with a Job that sometimes outlasts its period (schedule every 30 min).
def forbid(duration_min, period=30, n=6):
    busy_until, out = 0, []
    for i in range(n):
        t = i * period
        if t < busy_until:
            out.append(f"{t:>3}m skipped")
        else:
            out.append(f"{t:>3}m starts"); busy_until = t + duration_min
    return ", ".join(out)
print("Forbid, runtime 20 min:", forbid(20))
print("Forbid, runtime 40 min:", forbid(40))

Program output:

every minute, no deadline    : missed=113 > 100 -> no Job started ('too many missed start times')
every minute, deadline 200 s : missed=113, eligible=4 -> starts one Job for scheduled time 10:22
hourly, down 5 h, no deadline: missed=6, eligible=6 -> starts one Job for scheduled time 06:00
hourly, deadline 2 h         : missed=6, eligible=2 -> starts one Job for scheduled time 06:00
hourly, deadline 10 min      : missed=6, none within the deadline -> all skipped, wait for next schedule
Forbid, runtime 20 min:   0m starts,  30m starts,  60m starts,  90m starts, 120m starts, 150m starts
Forbid, runtime 40 min:   0m starts,  30m skipped,  60m starts,  90m skipped, 120m starts, 150m skipped

The first two lines reproduce the documented example, with the model’s own count of eligible schedules in the second. With no deadline and 113 missed minutes, nothing starts. The hourly examples show the trade-off a deadline expresses. After a five-hour outage, a deadline of 2 hours starts one Job for the newest eligible time, a deadline of 10 minutes starts nothing and waits for the next schedule, and no deadline starts the newest of six missed times. The model does not replay the six missed hours one by one, and a CronJob should not be relied on to do so: it is a trigger, not a backfill engine. The glossary entry on a data backfill describes the replay of history that a pipeline needs separately. The Forbid lines show a second kind of skipped run: when a Job takes 40 minutes on a 30-minute schedule, every other start is skipped. Two other documented details matter for operations. A startingDeadlineSeconds below 10 seconds may leave the CronJob never scheduled, because the controller checks every 10 seconds. And scheduled times that pass while suspend is true count as missed, so on resuming a CronJob that has no deadline the missed Jobs are scheduled immediately.

apiVersion: batch/v1
kind: CronJob
metadata:
  name: orders-load
spec:
  schedule: "30 2 * * *"
  timeZone: "Etc/UTC"                   # stable since Kubernetes v1.27
  concurrencyPolicy: Forbid             # skip a start while the previous Job of this CronJob is running
  startingDeadlineSeconds: 3600         # a start more than 1 hour late is skipped
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 5
  jobTemplate:
    spec:
      backoffLimit: 3
      activeDeadlineSeconds: 7200
      ttlSecondsAfterFinished: 86400
      template:
        spec:
          restartPolicy: Never
          containers:
          - name: load
            image: registry.example.com/data/orders-load@sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
            args: ["--day", "yesterday-utc"]    # the template is static: the program resolves "yesterday" itself

The Job template inside a CronJob is static, and that has a consequence for data work. The same template is used for every run, so nothing in it says which day’s data this Job should load. The manifest above leaves the program to resolve “yesterday” from the clock, which makes a late or repeated start load a different day than the one scheduled. Starting with Kubernetes v1.32, the CronJob controller adds the annotation batch.kubernetes.io/cronjob-scheduled-timestamp to each Job it creates, recording the originally scheduled time in RFC 3339 format. That annotation is on the Job object, not on its Pods, so a program does not receive it automatically. It can read it from its own Job through the API server, which requires a service account with permission to do so. For pipelines whose correctness depends on the data interval, such as daily partitions and backfills, an orchestrator that passes the interval to the job as an argument is the better design: the interval then comes from the schedule, not from when the Pod happened to start. The Airflow articles on DAGs, tasks, and scheduling and on retries, backfills, and safe recovery describe that model, and the last section of this article shows how the two layers combine.

Programs that can be started twice

A retried or duplicate attempt is normal behavior on Kubernetes, and the platform cannot undo what an earlier attempt wrote to a database or an object store. The Job documentation tells the program author to handle temporary files, locks, and incomplete output left by previous runs, and to tolerate concurrency when more than one Pod runs at once. The design principles are covered in Building Incremental and Idempotent Pipelines and the glossary entry on idempotency. This section shows what they do to a Job’s retries, using toy data and SQLite as a stand-in destination.

The same attempt twice, written two ways

The data is one logical day of 12 orders in 3 batches of 4. The first Pod is killed after its second batch has been written, and the Job controller starts a replacement for the same day. The first variant appends each batch and commits it. The second deletes the day’s rows and inserts them again inside a single transaction, so a killed attempt leaves nothing behind.

"""Three ways a retried batch attempt can write to a destination. SQLite stands in for the destination table.
Toy data: 3 batches of 4 rows for one logical day. Attempt 1 is killed after its second batch is written."""
import sqlite3

DAY = "2024-10-26"
BATCHES = [[(DAY, b * 4 + i, 10 * (b * 4 + i)) for i in range(4)] for b in range(3)]  # (day, order_id, amount)

class Killed(Exception):
    pass

def fresh():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE orders (day TEXT, order_id INT, amount INT)")
    return db

def summary(db):
    n, total = db.execute("SELECT COUNT(*), SUM(amount) FROM orders WHERE day=?", (DAY,)).fetchone()
    return f"rows={n} amount_sum={total}"

# Variant 1: append each batch and commit it; no identity for the run.
def append_attempt(db, die_after_batch=None):
    for k, batch in enumerate(BATCHES):
        db.executemany("INSERT INTO orders VALUES (?,?,?)", batch)
        db.commit()
        if die_after_batch == k:
            raise Killed

# Variant 2: replace the whole day inside one transaction (delete the day's rows, insert them again).
def replace_attempt(db, die_after_batch=None):
    with db:  # one transaction; rolled back if an exception escapes
        db.execute("DELETE FROM orders WHERE day=?", (DAY,))
        for k, batch in enumerate(BATCHES):
            db.executemany("INSERT INTO orders VALUES (?,?,?)", batch)
            if die_after_batch == k:
                raise Killed

def run_with_retry(attempt):
    db = fresh()
    try:
        attempt(db, die_after_batch=1)      # pod 1: killed after batch index 1
    except Killed:
        pass
    attempt(db)                              # pod 2: replacement pod runs the same logical day
    return db

print("expected: rows=12 amount_sum=660")
print("append + retry      :", summary(run_with_retry(append_attempt)))
print("replace-day + retry :", summary(run_with_retry(replace_attempt)))

# Two pods for the same day running one after another to completion (a duplicate start), both with replace-day.
db = fresh()
replace_attempt(db); replace_attempt(db)
print("replace-day, 2 full runs:", summary(db))
db2 = fresh()
append_attempt(db2); append_attempt(db2)
print("append, 2 full runs     :", summary(db2))

Program output:

expected: rows=12 amount_sum=660
append + retry      : rows=20 amount_sum=940
replace-day + retry : rows=12 amount_sum=660
replace-day, 2 full runs: rows=12 amount_sum=660
append, 2 full runs     : rows=24 amount_sum=1320

The append variant ends with 20 rows and an amount sum of 940 instead of 12 rows and 660, because the killed Pod’s 8 committed rows remain and the replacement adds all 12 again. The failure is silent: the Job reports success. The replace-the-day variant ends correct after the retry and after two complete runs, because the destination operation is defined in terms of the logical day, not in terms of this attempt’s progress. The two complete runs at the bottom model the duplicate start that the documentation warns about. The append variant doubles the day, and the replace variant is unchanged.

The logical day is the identity of the work, and it comes from the Job’s input, passed here as --day. It does not come from the Pod name, which differs for every attempt, or from the clock at the moment of execution. A destination operation keyed by the Pod name cannot recognize its own previous attempt.

When the destination has no transaction

Object storage rarely offers a transaction across several files. The usual substitute is to let each attempt write to its own staging location and to make one small commit operation decide which attempt’s files count. The program below uses a local directory as the store, a pointer file as the commit record, and os.replace, which replaces a file atomically on a single file system, as the commit. Real object stores provide different commit primitives, such as a table-format commit or a manifest object, and the design must use the one the destination actually has. The point to take from the model is the structure.

"""Attempt-scoped staging plus one atomic publish step. A local directory stands in for object storage;
os.replace on a pointer file stands in for the single commit operation a real destination offers
(a table commit, or a manifest/marker object). Toy data: one logical day split into 3 part files."""
import json, os, shutil, tempfile, threading

root = tempfile.mkdtemp()
RUN = "orders/day=2024-10-26"            # identity of the work, derived from the input interval, not from the Pod
PARTS = {f"part-{i}.csv": "\n".join(f"{i * 4 + j},{10 * (i * 4 + j)}" for j in range(4)) for i in range(3)}

def attempt(pod_name, die_after=None):
    stage = os.path.join(root, RUN, "_staging", pod_name)       # unique per attempt
    os.makedirs(stage)
    for k, (name, body) in enumerate(PARTS.items()):
        with open(os.path.join(stage, name), "w") as f:
            f.write(body)
        if die_after == k:
            return "killed"                                       # leaves a partial staging directory behind
    # commit: a pointer file naming the complete attempt directory; replaced atomically
    tmp = os.path.join(root, RUN, f"_pointer.{pod_name}.tmp")
    with open(tmp, "w") as f:
        json.dump({"dir": os.path.relpath(stage, os.path.join(root, RUN)), "parts": sorted(PARTS)}, f)
    os.replace(tmp, os.path.join(root, RUN, "_pointer.json"))
    return "published"

def read_published():
    base = os.path.join(root, RUN)
    p = os.path.join(base, "_pointer.json")
    if not os.path.exists(p):
        return None
    ptr = json.load(open(p))
    return sum(len(open(os.path.join(base, ptr["dir"], n)).read().splitlines()) for n in ptr["parts"]), ptr["dir"]

print("before any attempt:", read_published())
print("pod-a-1:", attempt("pod-a-1", die_after=1), "| reader sees:", read_published())
res = {}
t1 = threading.Thread(target=lambda: res.update(b=attempt("pod-b-2")))
t2 = threading.Thread(target=lambda: res.update(c=attempt("pod-c-3")))   # duplicate start for the same day
t1.start(); t2.start(); t1.join(); t2.join()
print("pod-b-2 / pod-c-3:", res["b"], res["c"], "| reader sees (rows, dir):", read_published())
print("staging dirs left:", sorted(os.listdir(os.path.join(root, RUN, "_staging"))))
shutil.rmtree(root)

Program output:

before any attempt: None
pod-a-1: killed | reader sees: None
pod-b-2 / pod-c-3: published published | reader sees (rows, dir): (12, '_staging/pod-b-2')
staging dirs left: ['pod-a-1', 'pod-b-2', 'pod-c-3']

The first attempt is killed after two of three part files and never writes the pointer, so a reader that follows the pointer sees nothing. Two further attempts for the same day run concurrently, as in a duplicate start, and each publishes a complete directory. The reader sees 12 rows from whichever pointer was written last, and both directories hold identical content, so either answer is right. The thread race decides which attempt’s directory the pointer names, so a rerun can show the other Pod. Three staging directories remain, including the partial one. Staging directories need a cleanup step that is separate from the commit and runs only after the pointer no longer names them, and it should be separate from the Job’s own cleanup, because ttlSecondsAfterFinished deletes Pods, not objects in a bucket.

Indexed Jobs and static partitioning

When the work divides naturally, a Job can run several Pods in parallel with completionMode: Indexed. Each Pod receives an index from 0 to completions - 1 through the JOB_COMPLETION_INDEX environment variable and the annotation batch.kubernetes.io/job-completion-index, and the Job is complete when one Pod has succeeded for each index. The documentation notes that, although rare, more than one Pod can be started for the same index after node failures, kubelet restarts, or evictions, and that only the first to succeed counts. Per-index failures can be limited separately with backoffLimitPerIndex, which is stable since Kubernetes v1.33.

The program has to turn an index into a set of inputs. A stable hash of the object name modulo the number of completions is one way.

"""Static work assignment for an Indexed Job: pod with index i handles the inputs whose stable hash maps to i.
Toy input: 12 invented object names for one day."""
import zlib

FILES = [f"events/day=2024-10-26/part-{i:03d}.json" for i in range(12)]

def my_files(index, completions):
    return [f for f in FILES if zlib.crc32(f.encode()) % completions == index]   # crc32 is stable across processes; hash() is not

for idx in range(4):
    print(f"completions=4 index={idx}:", [f[-8:-5] for f in my_files(idx, 4)])
moved = sum(1 for f in FILES if zlib.crc32(f.encode()) % 4 != zlib.crc32(f.encode()) % 5)
print(f"changing completions 4 -> 5 moves {moved} of {len(FILES)} files to a different index")

Program output:

completions=4 index=0: ['000', '007', '009', '010']
completions=4 index=1: ['001', '006', '008', '011']
completions=4 index=2: ['003', '004']
completions=4 index=3: ['002', '005']
changing completions 4 -> 5 moves 9 of 12 files to a different index

Two properties of this assignment are worth reading from the output. The shards are uneven: with 12 files and 4 indexes the sizes are 4, 4, 2, and 2, so the Job takes as long as its largest shard, and a skewed input makes one index the straggler for everyone. And the assignment depends on the number of completions: changing it from 4 to 5 moves 9 of the 12 files to a different index. A rerun of the same index with the same completions processes the same files, which is what lets a retried index repeat its own output safely. Changing the shard count between a first attempt and a retry of the same run, or between a run and its replay, does not. The index belongs in the output path and the shard count in the run’s recorded parameters.

Termination, preemption, and non-retryable errors

A Pod is deleted by a node drain, by a scale-down, by preemption, or by the Job controller itself when a deadline passes. The kubelet then sends TERM to the container’s main process and, when the grace period (30 seconds unless terminationGracePeriodSeconds says otherwise) is over, KILL to whatever remains. A batch program can use that window to stop between units of work, flush a durable checkpoint, and skip the commit. The following program plays the kubelet against a child process whose work is divided into batches.

"""What a batch program can do between SIGTERM and SIGKILL. The parent plays the kubelet: it sends SIGTERM,
waits for the grace period, then sends SIGKILL. Times are shortened to fractions of a second."""
import signal, subprocess, sys, time

CHILD = r'''
import signal, sys, time
stop = False
def on_term(signum, frame):
    global stop
    stop = True                                   # finish the current batch, then leave without publishing
signal.signal(signal.SIGTERM, on_term)
batch_seconds = float(sys.argv[1])
for batch in range(10):
    time.sleep(batch_seconds)                     # one unit of work
    print(f"batch {batch} done", flush=True)
    if stop:
        print("SIGTERM seen: stopping between batches, not publishing, exiting 143", flush=True)
        sys.exit(143)
print("all batches done, publishing", flush=True)
'''

def run(batch_seconds, term_after, grace):
    p = subprocess.Popen([sys.executable, "-u", "-c", CHILD, str(batch_seconds)], stdout=subprocess.PIPE, text=True)
    time.sleep(term_after)
    p.send_signal(signal.SIGTERM)                 # kubelet: TERM to the container's main process
    try:
        out, _ = p.communicate(timeout=grace)     # the grace period
    except subprocess.TimeoutExpired:
        p.kill()                                  # kubelet: KILL when the grace period is over
        out, _ = p.communicate()
    print(f"batch={batch_seconds}s grace={grace}s -> exit code {p.returncode}")
    for line in out.strip().splitlines():
        print("   ", line)

run(batch_seconds=0.3, term_after=0.45, grace=1.0)
run(batch_seconds=3.0, term_after=0.45, grace=1.0)

Program output:

batch=0.3s grace=1.0s -> exit code 143
    batch 0 done
    batch 1 done
    SIGTERM seen: stopping between batches, not publishing, exiting 143
batch=3.0s grace=1.0s -> exit code -9

In the first run the child sees TERM during a 0.3-second batch, finishes that batch, does not publish, and exits with 143, which is 128 plus the signal number. In the second run a single batch takes longer than the grace period, so the parent kills the child. The exit status is reported by Python as -9, which a cluster shows as 137, and the child printed nothing. A program whose units of work are longer than the grace period cannot rely on cleanup code, and its output must be safe for a kill at any instruction. That is the reason for the staging design above. The glossary entry on graceful shutdown describes the sizing of grace periods for services.

Disruptions are routine on shared and spot capacity, so a job that retries on every disruption can spend its budget on events unrelated to its own code. The podFailurePolicy in the first manifest handles both ends. An Ignore rule for the DisruptionTarget condition keeps preemptions and evictions from counting toward backoffLimit. A FailJob rule for a specific exit code ends the Job immediately for errors that a retry cannot fix. The pod failure policy has been stable since Kubernetes v1.31, and its rules require restartPolicy: Never. Retrying an invalid input three more times only delays the alert and spends cluster time; the Airflow operations article makes the same point under when a retry is the wrong response.

Declaring resources for work whose size changes

A service’s load is steady enough to tune once. A batch job’s cost follows its input, which varies from day to day, so its resource declaration is a statement about the largest input it should handle. The fundamentals are covered in Kubernetes Controllers and Workload Lifecycle, in the CPU quota model of Containers: Images, Isolation, and Resource Boundaries, and in the glossary entry on resource requests and limits. In short, the scheduler uses requests to place the Pod, and the kubelet and kernel enforce limits while it runs. This section applies those rules to batch jobs.

Placement depends on requests, not on what is idle

A node has an allocatable amount of each resource. The scheduler’s filtering step admits the node only if the new Pod’s request fits within what remains after the requests of the Pods already placed there. Actual usage does not enter that test, and the limit does not either. The program below models only that arithmetic with invented node sizes, using the case from the glossary entry: a node with 2 CPUs allocatable and 1.5 CPUs already requested.

"""Placement by requests, as the scheduler's filtering step does for CPU and memory (a model of the arithmetic only).
Units: CPU in millicores, memory in MiB. Node numbers are invented."""

class Node:
    def __init__(self, name, cpu_alloc, mem_alloc, requests):
        self.name, self.cpu_alloc, self.mem_alloc = name, cpu_alloc, mem_alloc
        self.pods = list(requests)                       # (cpu_request, mem_request) of pods already placed
    def free(self):
        return (self.cpu_alloc - sum(c for c, _ in self.pods), self.mem_alloc - sum(m for _, m in self.pods))
    def fits(self, cpu, mem):
        fc, fm = self.free()
        return cpu <= fc and mem <= fm

def schedule(pod_name, cpu_req, mem_req, nodes):
    ok = [n for n in nodes if n.fits(cpu_req, mem_req)]
    if not ok:
        reasons = []
        for n in nodes:
            fc, fm = n.free()
            r = [x for x, bad in (("Insufficient cpu", cpu_req > fc), ("Insufficient memory", mem_req > fm)) if bad]
            reasons.append(f"{n.name}: {', '.join(r)} (free {fc}m, {fm}Mi)")
        return f"{pod_name} ({cpu_req}m, {mem_req}Mi): Pending, FailedScheduling -> " + "; ".join(reasons)
    best = max(ok, key=lambda n: n.free()[0])             # any rule would do for this demo
    best.pods.append((cpu_req, mem_req))
    return f"{pod_name} ({cpu_req}m, {mem_req}Mi): bound to {best.name}"

# The situation from the glossary example: 2 CPUs allocatable, 1.5 CPUs already requested. Actual use is irrelevant here.
node = Node("node-1", 2000, 8192, [(1000, 2048), (500, 2048)])
print("free after existing requests:", node.free())
for cpu in (750, 250, 500):
    n = Node("node-1", 2000, 8192, [(1000, 2048), (500, 2048)])
    print(schedule(f"job-cpu{cpu}", cpu, 512, [n]))
# A limit of 1 CPU changes nothing about placement: only the request enters the fit test.
n = Node("node-1", 2000, 8192, [(1000, 2048), (500, 2048)])
print("request 250m / limit 1000m ->", schedule("job-limit1000", 250, 512, [n]), "| free now:", n.free())

# Memory decides, CPU is free: a two-node pool and a pod that asks for a lot of memory.
pool = [Node("node-1", 4000, 16384, [(500, 14000)]), Node("node-2", 4000, 16384, [(500, 12000)])]
print(schedule("big-memory", 500, 6144, pool))

Program output:

free after existing requests: (500, 4096)
job-cpu750 (750m, 512Mi): Pending, FailedScheduling -> node-1: Insufficient cpu (free 500m, 4096Mi)
job-cpu250 (250m, 512Mi): bound to node-1
job-cpu500 (500m, 512Mi): bound to node-1
request 250m / limit 1000m -> job-limit1000 (250m, 512Mi): bound to node-1 | free now: (250, 3584)
big-memory (500m, 6144Mi): Pending, FailedScheduling -> node-1: Insufficient memory (free 3500m, 2384Mi); node-2: Insufficient memory (free 3500m, 4384Mi)

A 750m request does not fit in the 500m that remain, even if the existing Pods use almost nothing. A 250m request and a 500m request fit. The fourth line sets a limit of 1000m on the 250m request: the node’s free CPU drops by 250m, not by 1000m, so the limit reserved nothing. The last line shows the Pending case with a message of the form the documentation shows for FailedScheduling, here caused by memory on both nodes while CPU was plentiful. The remedies for a Pending Job live outside the Job: reduce an inflated request, add capacity, or wait. Meanwhile activeDeadlineSeconds is already counting, so a Job that stays unscheduled can fail its deadline without a single container having started.

Memory is the limit that kills

The documentation describes the two limits differently. A CPU limit is enforced by throttling. A memory limit is enforced by the kernel’s out-of-memory killer, reactively: a container that exceeds its limit may be killed, and the kill happens when the kernel detects memory pressure. Because memory cannot be taken back from a running program the way CPU time can be withheld, the sizing question for a batch job is its peak, not its average. The peak is a property of the code as much as of the data. The following measurement runs two ways of aggregating the same synthetic CSV file in fresh processes, and reports the process’s peak resident set size. The data is generated by the script: rows of an order id, a customer id, an amount, and a 40-character note. The measurements were taken on a local macOS machine with Python 3.12, and the exact ratio depends on the language, the library, and the data. The shape is the lesson. The listing shows the two strategies; the harness that generates the files and launches one process per strategy is omitted.

def load_all(path):
    with open(path, newline="") as f:
        rows = list(csv.DictReader(f))               # every row held as a dict at once
    totals = {}
    for r in rows:
        totals[r["customer_id"]] = totals.get(r["customer_id"], 0) + int(r["amount"])
    return len(totals)

def stream(path):
    totals = {}
    with open(path, newline="") as f:
        for r in csv.DictReader(f):                  # one row alive at a time
            totals[r["customer_id"]] = totals.get(r["customer_id"], 0) + int(r["amount"])
    return len(totals)

# in the child process, after the call:
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss   # bytes on macOS, KiB on Linux

Program output:

      rows  file MiB  load_all peak MiB  stream peak MiB
    200000      10.7                 96               11
    400000      21.5                182               11
    800000      43.1                351               11

Loading everything costs about eight times the file size above the interpreter’s baseline, and the cost grows linearly with the rows: 96, 182, and 351 MiB for files of 10.7, 21.5, and 43.1 MiB. The streaming version stays at 11 MiB for every size, because only the running totals and one row are alive. Two conclusions follow for the manifest. A memory limit chosen from yesterday’s input is wrong the day the input doubles, if the program holds the input in memory, and the failure is an OOM kill at the moment of peak, not a gradual slowdown. And the cheapest memory reduction is often a change to the code, not a larger limit. When a program’s peak scales with its input, either bound the input per Pod, for example by partitioning across an Indexed Job, or set the limit from the largest input the job is supposed to handle and alert when inputs approach it.

A workable sizing routine is to run the job on the largest representative input, record the peak memory the platform reports, add headroom for variation, and set the memory request equal to the memory limit, as in the first manifest. Equal values make the reservation match what the Pod can use, so a Pod that is placed does not later push its node into memory pressure. After a failure, read the container’s last terminated state before changing anything: reason OOMKilled with exit code 137 identifies the memory limit, while an Evicted Pod points to node pressure or a storage limit.

CPU limits trade neighbor protection for duration

A CPU request is a scheduling reservation and a weight when containers compete. A CPU limit is a hard ceiling enforced by the kernel’s quota, so a container that wants more is paused until the next period. For a batch job, throttling does not cause errors. It lengthens the run, and a longer run meets the Job’s activeDeadlineSeconds sooner. The first manifest sets a CPU request and no CPU limit: the request reserves capacity and gives the job a fair weight under contention, and without a cap the job can use idle CPU and finish sooner. Setting a limit protects neighbors on a shared node, which matters where the noisy neighbor problem is real, at the price of a longer and less predictable duration. Which choice is better depends on how contended the nodes are, so measure before deciding. The throttled-period counters in the container’s CPU statistics (the cpu.stat file under cgroup v2) show whether a limit is binding.

Libraries that size a thread pool from the machine’s CPU count need attention here. In Python, os.cpu_count() reports the logical CPUs of the machine, and the container’s CPU quota is not part of that number. A job with a 1-CPU limit on a 32-CPU node can start dozens of workers that spend their time throttled. Set the worker count explicitly from the job’s own parameter.

Local storage and Spark’s memory arithmetic

Batch programs write temporary files, sort spills, and downloaded inputs. Local ephemeral storage is a resource with its own request and limit, and the kubelet counts writes to the container’s writable layer, writes to local emptyDir volumes, and the Pod’s own logs. When a Pod’s local ephemeral usage exceeds its limit, the kubelet marks it for eviction. The Pod ends as Failed with reason Evicted, which looks different from an OOM kill. An emptyDir with medium: Memory is a different trade-off: it is RAM-backed, and the documentation states that files written there count against the container’s memory limit. Without a sizeLimit, such a volume may consume up to the Pod’s memory limit. Large scratch space in RAM therefore competes with the program’s own memory. For data that must survive the Pod, use a persistent volume claim, described in the glossary entry on the persistent volume claim and in Kubernetes Networking, Storage, and Stateful Workloads, or object storage as in Designing an Object-Storage Data Lake.

Spark on Kubernetes shows how a framework’s memory model meets the container limit. The Spark documentation states that the driver and executor Pods’ memory request and limit are set by summing spark.{driver,executor}.memory and spark.{driver,executor}.memoryOverhead, where the overhead covers non-heap memory such as native allocations and, for PySpark, the Python worker processes. The overhead defaults to the larger of 384 MiB and a fraction of the heap, and the documented fraction is 0.10 for JVM jobs and 0.40 for non-JVM jobs on Kubernetes. As arithmetic on those documented defaults, an executor with an 8 GiB heap gets a 0.8 GiB overhead for a JVM job, and a container limit of about 8.8 GiB, while a PySpark job gets 3.2 GiB, and a limit of about 11.2 GiB. A job that sets only the heap and is killed at the container limit is usually short of overhead, not of heap. The driver Pod also differs from a normal Job Pod: the documentation states that executor Pods are cleaned up when the application completes, while the driver Pod persists in the completed state until it is garbage collected or removed. Plan the cleanup of finished driver Pods, and see Apache Spark Architecture and Execution and the glossary entry on the Spark driver and executor for how the two roles divide the work.

Getting data and credentials in and out

A Pod’s writable layer and emptyDir are local and temporary, so a Job’s inputs and outputs live in systems reached over the network: object storage, a warehouse, a database, a queue. That has three practical consequences.

First, the parallelism of the Job is a load setting for those systems. A hypothetical Job with 20 parallel Pods that each hold a 5-connection pool to the source database asks it for up to 100 connections at once, and a retry storm adds more. The Job’s parallelism and the program’s pool size should be chosen from the source’s capacity, and backpressure belongs in the design if the source cannot slow producers down.

Second, credentials must not be part of the image. Environment variables and files mounted from Secret objects are the basic mechanisms for supplying them. A Secret is an API object, and whether its values are encrypted at rest in the cluster’s datastore is a cluster setting, so a Secret is a delivery mechanism and not a vault. Where the platform supports it, short-lived credentials bound to the Pod’s service account are preferable to long-lived keys in a Secret, and each Job’s service account should hold only the permissions its work needs, as the glossary entries on secrets management and least privilege describe.

Third, the evidence about a run has to be kept somewhere that outlives the Pod. With ttlSecondsAfterFinished and the CronJob history limits, Jobs, Pods, and their logs disappear on a schedule, and the events that explain a failure expire even sooner. The run’s own record should therefore live with the data: a row per run with the logical day or shard index, the input identity (for example the object versions or the high-water mark that were read), the Pod name of the attempt that committed, the output location, and row counts. Ship container logs to a log store as well. The Pod name is useful in that record as an attempt label, but, as the earlier sections showed, not as the key that makes the output safe to repeat.

Where Airflow and Spark fit

Most teams do not write CronJobs for every pipeline. They run an orchestrator and use Kubernetes as the place where tasks execute. The Airflow provider for Kubernetes offers a KubernetesPodOperator, which the provider documentation describes as using the Kubernetes API to launch a Pod for a task, with task-level resource configuration, and which allows Airflow to act as a job orchestrator for jobs in any language. It creates a Pod directly, not a Job, so no Job controller replaces a failed Pod. Retries come from Airflow’s task retries, and the operator’s documented durable mode, which requires Airflow 3.3 or newer, lets a retried task reconnect to a Pod that is still running instead of creating a duplicate when its worker died. The provider also lists a KubernetesJobOperator that creates a Job.

This produces two retry layers whose attempts multiply when both are active. Consider a hypothetical task with 2 Airflow retries and a Job with backoffLimit: 3: the worst case is 3 task attempts times 4 Pods per attempt, 12 Pods that may each write to the destination. Pick one layer to do the retrying and have the other fail fast, or make the output safe for the product. Airflow also supplies what a CronJob lacks. It passes the data interval to the task, so the job’s input identity comes from the schedule, and it gives backfills a defined meaning. The task instance and retry entry and the Airflow operations article describe how Airflow’s attempts relate to a logical run.

Spark on Kubernetes follows the same split in other terms. The Spark driver runs in a Pod and requests executor Pods from the Kubernetes API. Pod placement, requests, limits, and preemption are Kubernetes concerns, while task retries and stage recomputation remain Spark’s. A lost executor Pod is a Spark event, a failed driver Pod is a failed application, and an application that is restarted from the outside starts from scratch unless its output was designed for that. The GPU variant of these scheduling questions, quotas and fragmentation, is the subject of the article on scheduling GPU workloads.

When Kubernetes is the wrong place for the job

A cluster is a good home for batch jobs when it already exists, when several teams share its capacity, or when jobs need different shapes of resources, such as large memory nodes or GPUs. It adds work when it does not exist. Someone has to operate the control plane or choose a managed one, patch nodes, size node pools, manage quotas, and answer questions about Pending Pods. For a handful of nightly jobs, a managed container service, a serverless container product, an orchestrator’s own workers, or the warehouse’s native scheduling can meet the need with less to operate. The trade-offs are covered in Choosing Kubernetes, Managed Services, or Simpler Runtime Options. The Job, CronJob, and resource rules above still apply to jobs that stay on Kubernetes, and the repeat-safe output design applies to every runtime, because every scheduler can start a job twice.

References

• Kubernetes documentation, Jobs
• Kubernetes documentation, CronJob
• Kubernetes documentation, Pods
• Kubernetes documentation, Init Containers
• Kubernetes documentation, Sidecar Containers
• Kubernetes documentation, Pod Lifecycle
• Kubernetes documentation, Resource Management for Pods and Containers
• Kubernetes documentation, Local ephemeral storage
• Kubernetes documentation, Volumes (emptyDir)
• Apache Spark documentation, Running Spark on Kubernetes
• Apache Spark documentation, Configuration (memory overhead settings)
• Apache Airflow cncf.kubernetes provider, KubernetesPodOperator