Kubernetes Jobs for Background Processing
In this tutorial, you will learn about Kubernetes Jobs for Background Processing. We cover key concepts, practical examples, and best practices to help you master this topic.
Run background job workers on Kubernetes as Jobs, Deployments, and CronJobs with proper resource management, scaling, and integration with Redis and message queues.
What You Learn
You will learn how to deploy job workers as Kubernetes Deployments, use Jobs for batch processing, configure autoscaling based on queue depth, and manage configuration with ConfigMaps.
Why It Matters
Kubernetes provides Orchestration for containerized workers: automatic restart, scaling, rolling updates, and resource management. Understanding K8s job patterns is essential for production-grade systems.
Real-World Use
DodaTech's workers run as Kubernetes Deployments with Horizontal Pod Autoscaler based on queue depth. Each pod runs one worker Process. Rolling updates ensure zero-downtime worker replacement.
Kubernetes Worker Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: job-worker
spec:
replicas: 3
selector:
matchLabels:
app: job-worker
template:
metadata:
labels:
app: job-worker
spec:
containers:
- name: worker
image: dodatech/worker:latest
env:
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: redis-secret
key: url
- name: QUEUE_NAME
value: "default"
- name: WORKER_CONCURRENCY
value: "4"
resources:
requests:
cpu: "250m"
memory: "128Mi"
limits:
cpu: "500m"
memory: "256Mi"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 10
periodSeconds: 30
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
periodSeconds: 10
Expected output:
Worker as Python K8s Client
import time
import json
import threading
class KubernetesWorker:
def __init__(self, worker_id, queue='default'):
self.worker_id = worker_id
self.queue = queue
self.processed = 0
self.running = True
def process_job(self, job):
print(f"[{self.worker_id}] Processing: {job.get('task', 'unknown')}")
time.sleep(0.5)
self.processed += 1
return {'status': 'completed', 'worker': self.worker_id}
def run(self):
print(f"[{self.worker_id}] Worker started, polling {self.queue}")
jobs = [
{'task': 'scan', 'file': 'doc1.pdf'},
{'task': 'scan', 'file': 'doc2.pdf'},
{'task': 'email', 'to': 'user@test.com'},
]
for job in jobs:
if not self.running:
break
self.process_job(job)
print(f"[{self.worker_id}] Completed {self.processed} jobs")
def stop(self):
self.running = False
# Simulate multiple worker pods
workers = [KubernetesWorker(f"worker-{i}") for i in range(3)]
threads = [threading.Thread(target=w.run, daemon=True) for w in workers]
for t in threads:
t.start()
for t in threads:
t.join()
total = sum(w.processed for w in workers)
print(f"Total processed across all workers: {total}")
Expected output:
[worker-0] Worker started, polling default
[worker-1] Worker started, polling default
[worker-2] Worker started, polling default
[worker-0] Processing: scan
[worker-1] Processing: scan
[worker-2] Processing: scan
...
[worker-0] Completed 3 jobs
[worker-1] Completed 3 jobs
[worker-2] Completed 3 jobs
Total processed across all workers: 9
HPA Based on Queue Depth
import time
import random
import threading
class QueueBasedAutoscaler:
def __init__(self, min_replicas=2, max_replicas=10):
self.min_replicas = min_replicas
self.max_replicas = max_replicas
self.current_replicas = min_replicas
self.queue_depth = 0
self.target_depth_per_pod = 10
def update_queue_depth(self, depth):
self.queue_depth = depth
def calculate_desired_replicas(self):
if self.queue_depth == 0:
return self.min_replicas
desired = (self.queue_depth + self.target_depth_per_pod - 1) // self.target_depth_per_pod
return max(self.min_replicas, min(self.max_replicas, desired))
def scale(self):
desired = self.calculate_desired_replicas()
if desired != self.current_replicas:
print(f"Scaling: {self.current_replicas} -> {desired} (queue: {self.queue_depth})")
self.current_replicas = desired
return self.current_replicas
def get_status(self):
return {
'queue_depth': self.queue_depth,
'current_replicas': self.current_replicas,
'desired_replicas': self.calculate_desired_replicas(),
'min': self.min_replicas,
'max': self.max_replicas,
}
scaler = QueueBasedAutoscaler()
def simulate_load():
depths = [0, 5, 25, 100, 200, 50, 10, 0]
for d in depths:
scaler.update_queue_depth(d)
scaler.scale()
time.sleep(0.2)
threading.Thread(target=simulate_load, daemon=True).start()
time.sleep(1)
status = scaler.get_status()
print(f"Final status: {status}")
Expected output:
Scaling: 2 -> 2 (queue: 0)
Scaling: 2 -> 2 (queue: 5)
Scaling: 2 -> 3 (queue: 25)
Scaling: 2 -> 10 (queue: 100)
Scaling: 10 -> 10 (queue: 200)
Scaling: 10 -> 5 (queue: 50)
Scaling: 5 -> 2 (queue: 10)
Scaling: 2 -> 2 (queue: 0)
Final status: ...
Rolling Update Strategy
import time
import threading
class RollingUpdateWorker:
def __init__(self, version='1.0', worker_id='w-1'):
self.version = version
self.worker_id = worker_id
self.running = True
self.draining = False
def start_drain(self):
self.draining = True
print(f"[{self.worker_id}] Draining, version {self.version}")
def process_job(self, job_id):
if self.draining:
print(f"[{self.worker_id}] Refusing new job {job_id}, draining")
return False
print(f"[{self.worker_id} v{self.version}] Processing {job_id}")
time.sleep(0.3)
return True
def shutdown(self):
self.running = False
print(f"[{self.worker_id}] Shut down")
def rolling_update_simulation():
old_worker = RollingUpdateWorker('1.0', 'worker-old')
new_worker = RollingUpdateWorker('2.0', 'worker-new')
# Old version is active
old_worker.process_job('job-001')
old_worker.process_job('job-002')
# Start rolling update
print("\n--- Rolling update started ---")
old_worker.start_drain()
old_worker.process_job('job-003')
# New version takes over
new_worker.process_job('job-003')
new_worker.process_job('job-004')
old_worker.shutdown()
print("--- Rolling update complete ---")
rolling_update_simulation()
Expected output:
[worker-old v1.0] Processing job-001
[worker-old v1.0] Processing job-002
--- Rolling update started ---
[worker-old] Draining, version 1.0
[worker-old] Refusing new job job-003, draining
[worker-new v2.0] Processing job-003
[worker-new v2.0] Processing job-004
[worker-old] Shut down
--- Rolling update complete ---
Common Mistakes
1. No Liveness or Readiness Probes
Without probes, Kubernetes cannot detect stuck workers. Configure liveness probe for hung processes and readiness probe for workers that cannot accept jobs.
2. Too Many Worker Replicas
More workers than needed cause contention for Redis connections and resources. Autoscale based on queue depth, not guesswork.
3. Ignoring Pod Disruption Budgets
Node maintenance and updates evict pods. Without PDB, all workers can be evicted simultaneously, causing processing gaps.
4. No Graceful Shutdown
When pods are terminated, workers must finish current jobs. Handle SIGTERM and implement drain logic.
5. Hardcoded Configuration
Configuration in the Docker image requires rebuilds for changes. Use ConfigMaps and Secrets for environment-specific configuration.
Practice Questions
1. What is the difference between K8s Job and Deployment for workers?
Job runs until completion (batch). Deployment keeps pods running continuously (queue workers). Choose based on workload type.
2. How does HPA autoscale workers?
HPA monitors a metric (queue depth) and adjusts replica count. Desired replicas = queue depth / target depth per pod.
3. Why use readiness probes for workers?
Readiness probe tells Kubernetes when a worker can accept jobs. During startup, the worker is not ready until it connects to Redis.
4. What is a Pod Disruption Budget for workers?
It limits how many worker pods can be down simultaneously during voluntary disruptions, ensuring minimum processing capacity.
Challenge
Deploy a worker on Kubernetes with: Deployment with 3 replicas, HPA based on queue depth, readiness and liveness probes, ConfigMap for configuration, rolling update strategy, and graceful shutdown handling.
FAQ
Mini Project: K8s Worker Simulator
import time
import threading
import random
class K8sWorkerSim:
def __init__(self, name, version='1.0'):
self.name = name
self.version = version
self.processed = 0
self.healthy = True
def health_check(self):
return {'pod': self.name, 'version': self.version, 'healthy': self.healthy, 'processed': self.processed}
def process(self, jobs):
for job in jobs:
if not self.healthy:
break
time.sleep(random.uniform(0.1, 0.3))
self.processed += 1
print(f"[{self.name}] {job} done (total: {self.processed})")
pods = [K8sWorkerSim(f"worker-pod-{i}") for i in range(3)]
jobs = [f"job-{j}" for j in range(5)]
threads = [threading.Thread(target=p.process, args=(jobs,), daemon=True) for p in pods]
for t in threads:
t.start()
for t in threads:
t.join()
for p in pods:
print(f" {p.name}: {p.processed} jobs processed (healthy: {p.healthy})")
Expected output:
[worker-pod-0] job-0 done (total: 1)
[worker-pod-1] job-0 done (total: 1)
[worker-pod-2] job-0 done (total: 1)
...
worker-pod-0: 5 jobs processed (healthy: True)
worker-pod-1: 5 jobs processed (healthy: True)
worker-pod-2: 5 jobs processed (healthy: True)
What's Next
Now that you understand Kubernetes Jobs, explore Kubernetes CronJobs for scheduled batch processing, then learn about job security and permissions for securing workers.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro