Job Dependencies and Chains — Complete Guide
In this tutorial, you will learn about Job Dependencies and Chains. We cover key concepts, practical examples, and best practices to help you master this topic.
Create job dependency graphs, execution chains, and directed acyclic graphs to enforce ordered processing and parallel execution of related background tasks.
What You Learn
You will learn how to model job dependencies as directed acyclic graphs, implement sequential chains and fan-out workflows, handle dependency failures, and use Redis for dependency tracking.
Why It Matters
Many workflows require ordered execution: generate thumbnail after image upload, send notification after payment succeeds, run cleanup after all workers finish. Without dependency management, you must write complex Orchestration code that is brittle and hard to maintain.
Real-World Use
DodaTech's file processing pipeline has three stages: scan for malware, generate previews, archive to cold storage. Each stage depends on the previous. If scanning fails, preview generation and archiving are skipped. The dependency engine handles this automatically.
Simple Job Chain
import time
import json
import redis
r = redis.Redis()
class JobChain:
def __init__(self):
self.chain_key = 'job_chain'
def create_chain(self, chain_id, steps):
chain = {
'id': chain_id,
'steps': steps,
'current_step': 0,
'status': 'pending',
'results': [],
}
r.set(f"{self.chain_key}:{chain_id}", json.dumps(chain))
return chain_id
def next_step(self, chain_id):
chain_data = r.get(f"{self.chain_key}:{chain_id}")
if not chain_data:
return None
chain = json.loads(chain_data)
if chain['current_step'] >= len(chain['steps']):
chain['status'] = 'completed'
r.set(f"{self.chain_key}:{chain_id}", json.dumps(chain))
return None
step = chain['steps'][chain['current_step']]
return step
def complete_step(self, chain_id, result):
chain_data = r.get(f"{self.chain_key}:{chain_id}")
if not chain_data:
return
chain = json.loads(chain_data)
chain['results'].append(result)
chain['current_step'] += 1
r.set(f"{self.chain_key}:{chain_id}", json.dumps(chain))
return chain['current_step']
def fail_chain(self, chain_id, error):
chain_data = r.get(f"{self.chain_key}:{chain_id}")
if not chain_data:
return
chain = json.loads(chain_data)
chain['status'] = 'failed'
chain['error'] = error
r.set(f"{self.chain_key}:{chain_id}", json.dumps(chain))
chain = JobChain()
chain.create_chain('process-file', ['scan', 'thumbnail', 'archive'])
step = chain.next_step('process-file')
while step:
print(f"Executing step: {step}")
time.sleep(0.5)
chain.complete_step('process-file', f"{step}-ok")
step = chain.next_step('process-file')
print("Chain completed")
Expected output:
Executing step: scan
Executing step: thumbnail
Executing step: archive
Chain completed
Directed Acyclic Graph (DAG) Dependencies
import json
import redis
import time
r = redis.Redis()
class DAGExecutor:
def __init__(self):
self.dag_key = 'dag_jobs'
def create_dag(self, dag_id, tasks, dependencies):
graph = {
'id': dag_id,
'tasks': tasks,
'deps': dependencies,
'completed': [],
'failed': [],
'status': 'running',
}
r.set(f"{self.dag_key}:{dag_id}", json.dumps(graph))
return dag_id
def get_ready_tasks(self, dag_id):
data = r.get(f"{self.dag_key}:{dag_id}")
if not data:
return []
graph = json.loads(data)
if graph['status'] != 'running':
return []
ready = []
for task in graph['tasks']:
if task in graph['completed'] or task in graph['failed']:
continue
deps = graph['deps'].get(task, [])
if all(d in graph['completed'] for d in deps):
ready.append(task)
return ready
def complete_task(self, dag_id, task, result):
data = r.get(f"{self.dag_key}:{dag_id}")
if not data:
return
graph = json.loads(data)
graph['completed'].append(task)
if len(graph['completed']) == len(graph['tasks']):
graph['status'] = 'completed'
r.set(f"{self.dag_key}:{dag_id}", json.dumps(graph))
def fail_task(self, dag_id, task, error):
data = r.get(f"{self.dag_key}:{dag_id}")
if not data:
return
graph = json.loads(data)
graph['failed'].append(task)
graph['status'] = 'failed'
r.set(f"{self.dag_key}:{dag_id}", json.dumps(graph))
dag = DAGExecutor()
dag.create_dag(
'build-pipeline',
['lint', 'test', 'build', 'deploy', 'notify'],
{'test': ['lint'], 'build': ['lint', 'test'], 'deploy': ['build'], 'notify': ['deploy']}
)
ready = dag.get_ready_tasks('build-pipeline')
while ready:
for task in ready:
print(f"Running: {task}")
dag.complete_task('build-pipeline', task, 'ok')
time.sleep(0.5)
ready = dag.get_ready_tasks('build-pipeline')
print("DAG completed")
Expected output:
Running: lint
Running: test
Running: build
Running: deploy
Running: notify
DAG completed
Fan-Out / Fan-In Pattern
import threading
import time
import random
class FanOutFanIn:
def __init__(self):
self.results = []
def fan_out(self, tasks, worker_func):
threads = []
for task in tasks:
thread = threading.Thread(target=lambda t: self.results.append(worker_func(t)), args=(task,))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
def fan_in(self, combine_func):
return combine_func(self.results)
def process_chunk(chunk):
time.sleep(random.uniform(0.1, 0.5))
result = chunk * 2
print(f" Processed chunk {chunk}: {result}")
return result
def combine(chunks):
return sum(chunks)
pipeline = FanOutFanIn()
chunks = [1, 2, 3, 4, 5]
print("Fanning out...")
pipeline.fan_out(chunks, process_chunk)
print("Fanning in...")
total = pipeline.fan_in(combine)
print(f"Total: {total}")
Expected output:
Fanning out...
Processed chunk 1: 2
Processed chunk 3: 6
Processed chunk 2: 4
Processed chunk 5: 10
Processed chunk 4: 8
Fanning in...
Total: 30
Dependency with Timeout
import time
import threading
class DependencyTimeout:
def __init__(self, timeout=10):
self.timeout = timeout
def wait_for_dependency(self, dep_name, check_func, interval=0.5):
start = time.time()
while time.time() - start < self.timeout:
if check_func():
print(f"Dependency {dep_name} satisfied")
return True
time.sleep(interval)
print(f"Dependency {dep_name} timed out after {self.timeout}s")
return False
def execute_after(self, dep_name, check_func, work_func):
if self.wait_for_dependency(dep_name, check_func):
return work_func()
return None
def is_upload_complete():
return random.random() < 0.3
def process_upload():
return "Upload processed"
dt = DependencyTimeout(timeout=5)
result = dt.execute_after('upload', is_upload_complete, process_upload)
print(f"Result: {result}")
Celery Chains and Groups
# celery_workflow.py
from celery import Celery, group, chain, chord
app = Celery('tasks', broker='redis://localhost:6379')
@app.task
def scan_file(path):
print(f"Scanning: {path}")
return {'path': path, 'clean': True}
@app.task
def generate_thumbnail(result):
print(f"Generating thumbnail for: {result['path']}")
return {'path': result['path'], 'thumbnail': 'preview.jpg'}
@app.task
def archive_file(result):
print(f"Archiving: {result['path']}")
return {'path': result['path'], 'archived': True}
@app.task
def notify_user(results):
print(f"All tasks completed: {results}")
return {'notified': True}
# Sequential chain
# chain(scan_file.s('/data/file.pdf'),
# generate_thumbnail.s(),
# archive_file.s())()
#
# Parallel group
# group(
# scan_file.s('/data/file1.pdf'),
# scan_file.s('/data/file2.pdf'),
# )()
#
# Chord (group + callback)
# chord(
# group(scan_file.s('/data/file1.pdf'),
# scan_file.s('/data/file2.pdf')),
# notify_user.s()
# )()
Common Mistakes
1. Circular Dependencies
Task A depends on B, B depends on C, C depends on A. The graph can never resolve. Always validate DAGs for cycles before execution.
2. Not Handling Dependency Failures
When a dependency fails, dependent tasks should either skip or fail. Without cascading failure handling, dependent tasks hang forever.
3. Overly Complex Dependency Graphs
Deep dependency chains increase latency and failure points. Keep graphs shallow. Prefer 2-3 levels over 10-level deep pipelines.
4. No Timeout on Dependencies
A missing dependency blocks execution indefinitely. Always set timeouts for waiting tasks.
5. Ignoring Partial Results
When a task in a fan-in pattern fails, partial results from completed tasks are lost. Collect and report partial results even on failure.
Practice Questions
1. What is a DAG in job scheduling?
A Directed Acyclic Graph where tasks are nodes and dependencies are edges. Direction indicates order. Acyclic means no circular dependencies.
2. What is the fan-out/fan-in pattern?
Fan-out splits work into parallel subtasks. Each subtask executes independently. Fan-in collects and combines all results.
3. How do you handle a failed dependency?
Propagate the failure: skip dependent tasks, mark the workflow as failed, and trigger rollback or cleanup tasks.
4. What is a chord in Celery?
A chord is a group of parallel tasks followed by a callback task that runs after all group tasks complete. It implements fan-out/fan-in.
Challenge
Build a deployment pipeline with these dependencies: lint (no deps), test (depends on lint), build (depends on lint and test), integration-test (depends on build), deploy-staging (depends on integration-test), deploy-production (depends on deploy-staging and requires manual approval), notify (depends on deploy-production or deploy-staging failure).
FAQ
Mini Project: Dependency Pipeline
import json
import redis
import time
r = redis.Redis()
class DependencyPipeline:
def __init__(self):
self.pipeline_key = 'pipeline'
def create(self, pipeline_id, stages):
pipeline = {
'id': pipeline_id,
'stages': stages,
'status': {s['name']: 'pending' for s in stages},
'results': {},
'errors': {},
}
r.set(f"{self.pipeline_key}:{pipeline_id}", json.dumps(pipeline))
return pipeline_id
def get_ready(self, pipeline_id):
data = r.get(f"{self.pipeline_key}:{pipeline_id}")
if not data:
return []
pipeline = json.loads(data)
ready = []
for stage in pipeline['stages']:
if pipeline['status'][stage['name']] != 'pending':
continue
deps = stage.get('depends_on', [])
if all(pipeline['status'][d] == 'completed' for d in deps):
ready.append(stage['name'])
return ready
def mark_completed(self, pipeline_id, stage_name, result=None):
data = r.get(f"{self.pipeline_key}:{pipeline_id}")
if not data:
return
pipeline = json.loads(data)
pipeline['status'][stage_name] = 'completed'
if result:
pipeline['results'][stage_name] = result
r.set(f"{self.pipeline_key}:{pipeline_id}", json.dumps(pipeline))
def mark_failed(self, pipeline_id, stage_name, error):
data = r.get(f"{self.pipeline_key}:{pipeline_id}")
if not data:
return
pipeline = json.loads(data)
pipeline['status'][stage_name] = 'failed'
pipeline['errors'][stage_name] = error
r.set(f"{self.pipeline_key}:{pipeline_id}", json.dumps(pipeline))
def is_complete(self, pipeline_id):
data = r.get(f"{self.pipeline_key}:{pipeline_id}")
if not data:
return False
pipeline = json.loads(data)
return all(s == 'completed' for s in pipeline['status'].values())
pipeline = DependencyPipeline()
pipeline.create('data-pipeline', [
{'name': 'extract', 'depends_on': []},
{'name': 'transform', 'depends_on': ['extract']},
{'name': 'load', 'depends_on': ['transform']},
{'name': 'notify', 'depends_on': ['load']},
])
ready = pipeline.get_ready('data-pipeline')
while ready:
for stage in ready:
print(f"Running: {stage}")
pipeline.mark_completed('data-pipeline', stage, f"{stage}-ok")
time.sleep(0.5)
ready = pipeline.get_ready('data-pipeline')
print(f"Pipeline complete: {pipeline.is_complete('data-pipeline')}")
Expected output:
Running: extract
Running: transform
Running: load
Running: notify
Pipeline complete: True
What's Next
Now that you understand job dependencies, explore distributed workers for scaling across multiple machines, then learn about worker scalability for handling increased load.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro