DAG-Based Job Workflows — Complete Guide
In this tutorial, you will learn about DAG. We cover key concepts, practical examples, and best practices to help you master this topic.
Design directed acyclic graph (DAG) workflows for complex job Orchestration with parallel execution, dependency resolution, topological sorting, and fault tolerance.
What You Learn
You will learn how to model job workflows as DAGs, resolve dependencies with topological sort, execute independent tasks in parallel, and handle failures in DAG-based execution.
Why It Matters
Simple linear chains cannot model complex workflows. DAGs allow parallel execution, conditional paths, and fan-out/fan-in patterns. Understanding DAG workflows enables building sophisticated job pipelines.
Real-World Use
DodaTech's deployment pipeline uses a DAG workflow: test job and security scan run in parallel, followed by build, then deploy to staging. Approval gates pause the workflow until manual confirmation.
DAG Workflow Model
flowchart TD
A[Start] --> B[Validate Config]
A --> C[Load Dependencies]
B --> D[Run Tests]
B --> E[Security Scan]
C --> D
D --> F[Build Artifact]
E --> F
F --> G[Deploy Staging]
G --> H{Approval}
H -->|Approved| I[Deploy Production]
H -->|Rejected| J[Rollback]
DAG Implementation
import json
from collections import defaultdict, deque
class DAGWorkflow:
def __init__(self):
self.graph = defaultdict(list)
self.nodes = {}
def add_node(self, name, task=None, metadata=None):
self.nodes[name] = {
'task': task,
'metadata': metadata or {},
'dependencies': [],
'status': 'pending',
}
def add_dependency(self, from_node, to_node):
self.graph[from_node].append(to_node)
self.nodes[to_node]['dependencies'].append(from_node)
def topological_sort(self):
in_degree = {node: 0 for node in self.nodes}
for node in self.nodes:
for neighbor in self.graph[node]:
in_degree[neighbor] = in_degree.get(neighbor, 0) + 1
queue = deque([n for n, d in in_degree.items() if d == 0])
result = []
while queue:
node = queue.popleft()
result.append(node)
for neighbor in self.graph[node]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if len(result) != len(self.nodes):
raise ValueError("Graph contains a cycle")
return result
def get_ready_nodes(self):
ready = []
for name, node in self.nodes.items():
if node['status'] == 'pending':
deps_met = all(
self.nodes[dep]['status'] == 'completed'
for dep in node['dependencies']
)
if deps_met:
ready.append(name)
return ready
def mark_completed(self, node_name, result=None):
if node_name in self.nodes:
self.nodes[node_name]['status'] = 'completed'
self.nodes[node_name]['result'] = result
def mark_failed(self, node_name, error=None):
if node_name in self.nodes:
self.nodes[node_name]['status'] = 'failed'
self.nodes[node_name]['error'] = error
def is_complete(self):
return all(
n['status'] in ('completed', 'failed')
for n in self.nodes.values()
)
def describe(self):
return {
'nodes': {n: {'status': s['status'], 'deps': s['dependencies']}
for n, s in self.nodes.items()},
'topological_order': self.topological_sort(),
}
dag = DAGWorkflow()
dag.add_node('fetch_data')
dag.add_node('validate')
dag.add_node('transform')
dag.add_node('load')
dag.add_node('notify')
dag.add_dependency('fetch_data', 'validate')
dag.add_dependency('validate', 'transform')
dag.add_dependency('transform', 'load')
dag.add_dependency('load', 'notify')
dag.mark_completed('fetch_data')
print(f"Ready after fetch: {dag.get_ready_nodes()}")
print(f"Topological order: {dag.topological_sort()}")
Expected output:
Ready after fetch: ['validate']
Topological order: ['fetch_data', 'validate', 'transform', 'load', 'notify']
Parallel DAG Execution
import threading
import time
import random
class ParallelDAG:
def __init__(self):
self.nodes = {}
self.edges = defaultdict(list)
self.reverse_edges = defaultdict(list)
self._lock = threading.Lock()
def add_task(self, name, func, deps=None):
self.nodes[name] = {
'func': func,
'deps': deps or [],
'status': 'pending',
'result': None,
}
for dep in (deps or []):
self.edges[dep].append(name)
self.reverse_edges[name].append(dep)
def get_pending(self):
with self._lock:
ready = []
for name, node in self.nodes.items():
if node['status'] == 'pending':
deps_ok = all(
self.nodes[d]['status'] == 'completed'
for d in node['deps']
)
if deps_ok:
ready.append(name)
return ready
def mark_done(self, name, result):
with self._lock:
self.nodes[name]['status'] = 'completed'
self.nodes[name]['result'] = result
def mark_failed(self, name, error):
with self._lock:
self.nodes[name]['status'] = 'failed'
self.nodes[name]['error'] = str(error)
def execute(self):
threads = []
while not self._all_done():
ready = self.get_pending()
for task_name in ready:
node = self.nodes[task_name]
t = threading.Thread(
target=self._run_task,
args=(task_name, node['func']),
daemon=True
)
t.start()
threads.append(t)
time.sleep(0.1)
for t in threads:
t.join()
return {n: nd['result'] for n, nd in self.nodes.items()
if nd['status'] == 'completed'}
def _run_task(self, name, func):
self.nodes[name]['status'] = 'running'
time.sleep(random.uniform(0.1, 0.3))
try:
result = func()
self.mark_done(name, result)
print(f"Completed: {name}")
except Exception as e:
self.mark_failed(name, e)
print(f"Failed: {name}: {e}")
def _all_done(self):
return all(
n['status'] in ('completed', 'failed')
for n in self.nodes.values()
)
pdag = ParallelDAG()
pdag.add_task('fetch', lambda: 'data_loaded', deps=[])
pdag.add_task('validate', lambda: 'validated', deps=['fetch'])
pdag.add_task('backup', lambda: 'backed_up', deps=['fetch'])
pdag.add_task('transform', lambda: 'transformed', deps=['validate', 'backup'])
pdag.add_task('notify', lambda: 'notified', deps=['transform'])
results = pdag.execute()
print(f"Results: {results}")
Expected output:
Completed: fetch
Completed: validate
Completed: backup
Completed: transform
Completed: notify
Results: {'fetch': 'data_loaded', 'validate': 'validated', ...}
DAG with Conditional Execution
import json
class ConditionalDAG:
def __init__(self):
self.tasks = {}
self.conditions = {}
def task(self, name, deps=None):
def decorator(func):
self.tasks[name] = {'func': func, 'deps': deps or []}
return func
return decorator
def condition(self, task_name, condition_func):
self.conditions[task_name] = condition_func
def execute(self, initial_data=None):
data = initial_data or {}
completed = set()
executed = set()
while len(executed) < len(self.tasks):
for name, tdef in self.tasks.items():
if name in executed:
continue
deps_met = all(d in completed for d in tdef['deps'])
if not deps_met:
continue
if name in self.conditions:
should_run = self.conditions[name](data)
if not should_run:
print(f"Skipping: {name} (condition not met)")
executed.add(name)
completed.add(name)
continue
print(f"Running: {name}")
result = tdef['func'](data)
data.update(result)
executed.add(name)
completed.add(name)
return data
cdag = ConditionalDAG()
@cdag.task('ingest', deps=[])
def ingest(data):
return {'records': 1000, 'format': 'csv'}
@cdag.task('validate', deps=['ingest'])
def validate(data):
if data['records'] > 500:
data['needs_sampling'] = True
return {'valid': True}
@cdag.task('sample', deps=['validate'])
def sample(data):
return {'sampled': True, 'sample_size': 100}
@cdag.task('report', deps=['validate'])
def report(data):
return {'report_generated': True}
cdag.condition('sample', lambda d: d.get('needs_sampling', False))
result = cdag.execute({'input_file': 'data.csv'})
print(f"Executed tasks: {list(result.keys())}")
Expected output:
Running: ingest
Running: validate
Running: sample
Skipping: report (condition not met)
...or...
Running: ingest
Running: validate
Running: report
DAG with Fan-Out/Fan-In
import threading
import time
from concurrent.futures import ThreadPoolExecutor
class FanOutFanInDAG:
def __init__(self):
self.stages = []
def stage(self, name):
def decorator(func):
self.stages.append({'name': name, 'func': func})
return func
return decorator
def execute(self, data, parallelism=4):
results = {}
current_data = data
for stage_info in self.stages:
stage_name = stage_info['name']
stage_func = stage_info['func']
print(f"Stage: {stage_name}")
if hasattr(stage_func, '_fan_out'):
items = stage_func(current_data)
with ThreadPoolExecutor(max_workers=parallelism) as pool:
sub_results = list(pool.map(
lambda item: self._process_item(stage_name, item), items
))
current_data = {'items': sub_results}
elif hasattr(stage_func, '_fan_in'):
current_data = stage_func(current_data)
else:
current_data = stage_func(current_data)
results[stage_name] = current_data
return results
def _process_item(self, stage, item):
time.sleep(0.1)
return f"processed_{item}"
def fan_out(count=5):
def decorator(func):
func._fan_out = count
return func
return decorator
def fan_in(func):
func._fan_in = True
return func
dag = FanOutFanInDAG()
@dag.stage('fetch')
def fetch(data):
return {'source': 's3', 'files': ['a.csv', 'b.csv', 'c.csv']}
@dag.stage('process')
@fan_out(3)
def process(data):
return data['files']
@dag.stage('aggregate')
@fan_in
def aggregate(data):
return {'total_items': len(data['items']), 'items': data['items']}
result = dag.execute({'job': 'etl-001'})
print(f"Aggregated: {result['aggregate']}")
Expected output:
Stage: fetch
Stage: process
Stage: aggregate
Aggregated: {'total_items': 3, 'items': ['processed_a.csv', ...]}
Common Mistakes
1. Circular Dependencies
Cycles in the graph cause infinite loops. Always validate DAGs for cycles before execution using topological sort.
2. Too Fine-Grained Tasks
DAGs with hundreds of tiny nodes are hard to manage. Keep tasks at a meaningful granularity (one task = one logical operation).
3. No Status Persistence
In-memory DAG state is lost on crash. Persist node statuses to Redis or database for recovery and visibility.
4. Ignoring Parallelism Limits
Running all ready nodes in parallel can overwhelm resources. Limit concurrency with a Semaphore or thread pool.
5. No Error Recovery Strategy
A single failed node can block the entire DAG. Define failure strategies: skip downstream, use fallback, or pause for manual intervention.
Practice Questions
1. What is a DAG in job workflows?
A directed acyclic graph where nodes represent tasks and edges represent dependencies. Acyclic means no circular dependencies are allowed.
2. How does topological sort help DAG execution?
It produces a linear ordering where every task appears after its dependencies, ensuring correct execution order and identifying ready tasks.
3. What is fan-out/fan-in in DAGs?
Fan-out splits work into parallel sub-tasks. Fan-in collects results from parallel branches before proceeding to the next stage.
4. How do you handle failures in DAG workflows?
Define per-node failure handlers, allow retries for transient failures, and provide skip or fallback options for non-critical nodes.
Challenge
Build a DAG workflow for a CI/CD pipeline: lint and test run in parallel, security scan runs with tests, build depends on both, deploy to staging after build, integration tests after deploy, production deploy after approval gate.
FAQ
Mini Project: DAG Workflow Engine
import json
import threading
import time
from collections import defaultdict
class DAGEngine:
def __init__(self):
self.nodes = {}
self.edges = defaultdict(list)
self._lock = threading.Lock()
def add_node(self, name, func):
self.nodes[name] = {'func': func, 'status': 'pending', 'result': None}
def add_edge(self, from_node, to_node):
self.edges[from_node].append(to_node)
def get_ready(self):
with self._lock:
ready = []
for name, node in self.nodes.items():
if node['status'] != 'pending':
continue
has_pending_deps = False
for src, targets in self.edges.items():
if name in targets and self.nodes[src]['status'] != 'completed':
has_pending_deps = True
break
if not has_pending_deps:
ready.append(name)
return ready
def run(self):
while True:
ready = self.get_ready()
if not ready:
if all(n['status'] in ('completed', 'failed') for n in self.nodes.values()):
break
time.sleep(0.1)
continue
for name in ready:
self.nodes[name]['status'] = 'running'
thread = threading.Thread(target=self._execute, args=(name,), daemon=True)
thread.start()
def _execute(self, name):
time.sleep(0.1)
try:
result = self.nodes[name]['func']()
with self._lock:
self.nodes[name]['status'] = 'completed'
self.nodes[name]['result'] = result
print(f"OK: {name}")
except Exception as e:
with self._lock:
self.nodes[name]['status'] = 'failed'
self.nodes[name]['error'] = str(e)
print(f"FAIL: {name}: {e}")
engine = DAGEngine()
engine.add_node('fetch', lambda: 'data')
engine.add_node('clean', lambda: 'cleaned')
engine.add_node('validate', lambda: 'valid')
engine.add_node('store', lambda: 'stored')
engine.add_edge('fetch', 'clean')
engine.add_edge('fetch', 'validate')
engine.add_edge('clean', 'store')
engine.add_edge('validate', 'store')
engine.run()
print("DAG complete")
Expected output:
OK: fetch
OK: clean
OK: validate
OK: store
DAG complete
What's Next
Now that you understand DAG workflows, explore job progress websocket for real-time pipeline status, then learn about job timeout for preventing stuck workflows.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro