Task Groups and Chords in Celery
In this tutorial, you will learn about Task Groups and Chords in Celery. We cover key concepts, practical examples, and best practices to help you master this topic.
Execute Celery tasks in parallel using groups, handle fan-out patterns with chords, and combine parallel results into a single aggregated output.
What You Learn
You will learn how to create parallel task groups, use chords to aggregate group results, combine groups with chains, and handle errors in parallel executions.
Why It Matters
Many workloads are embarrassingly parallel: processing 1000 images, fetching 50 URLs, analyzing 10 log files. Sequential processing takes 10x longer. Groups execute these in parallel, and chords combine the results into a single output.
Real-World Use
Doda Browser's batch file analysis uses a chord: 50 files are scanned in parallel by separate workers, and the chord callback aggregates all results into a single report. This reduces total analysis time from 50 minutes to 1 minute.
Basic Group
from celery import Celery, group
app = Celery('group_demo', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def process_item(item):
return f"Processed: {item}"
# Execute tasks in parallel
tasks = [process_item.s(f"item_{i}") for i in range(5)]
job = group(tasks)
result = job.delay()
print(f"Group ID: {result.id}")
print(f"Results: {result.get(timeout=10)}")
Expected output:
Group ID: 550e8400-...
Results: ['Processed: item_0', 'Processed: item_1', 'Processed: item_2', 'Processed: item_3', 'Processed: item_4']
Group with Variable Input
from celery import Celery, group
app = Celery('group_var', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def fetch_url(url):
import requests
return f"Fetched {url}: {len(url)} chars"
@app.task
def process_file(filepath):
return f"File {filepath}: processed"
# Fetch multiple URLs in parallel
urls = [
"https://example.com",
"https://httpbin.org/get",
]
tasks = [fetch_url.s(url) for url in urls]
result = group(tasks).delay()
print(result.get(timeout=30))
Expected output:
['Fetched https://example.com: 19 chars', 'Fetched https://httpbin.org/get: 23 chars']
Chords: Group with Callback
A chord is a group followed by a callback that receives the list of results:
from celery import Celery, chord
app = Celery('chord_demo', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def process_chunk(chunk):
return f"chunk_{chunk}"
@app.task
def aggregate(results):
total = len(results)
return f"Aggregated {total} results: {', '.join(results)}"
# Chord: process chunks in parallel, then aggregate results
callback = aggregate.s()
header = [process_chunk.s(i) for i in range(5)]
result = chord(header)(callback)
print(f"Chord result: {result.get(timeout=10)}")
Expected output:
Chord result: Aggregated 5 results: chunk_0, chunk_1, chunk_2, chunk_3, chunk_4
Real-World Chord: Batch Processing
from celery import Celery, chord
import time
import random
app = Celery('batch', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task(bind=True)
def scan_file(self, file_id, file_path):
print(f"Scanning {file_id}: {file_path}")
time.sleep(random.uniform(0.5, 2.0))
threat = random.choice(['clean', 'clean', 'clean', 'suspicious'])
return {'file_id': file_id, 'path': file_path, 'threat': threat}
@app.task
def generate_report(results):
total = len(results)
threats = [r for r in results if r['threat'] != 'clean']
clean = total - len(threats)
report = f"""
Scan Report:
Total files: {total}
Clean: {clean}
Threats found: {len(threats)}
"""
for t in threats:
report += f" - {t['path']}: {t['threat']}\n"
print(report)
return {'total': total, 'clean': clean, 'threats': len(threats)}
# Batch scan 10 files
files = [(i, f"/data/file_{i}.exe") for i in range(10)]
header = [scan_file.s(fid, path) for fid, path in files]
result = chord(header)(generate_report.s())
print(result.get(timeout=60))
Expected output:
Scan Report:
Total files: 10
Clean: 8
Threats found: 2
- /data/file_3.exe: suspicious
- /data/file_7.exe: suspicious
{'total': 10, 'clean': 8, 'threats': 2}
Groups within Chains
Combine groups and chains for complex workflows:
from celery import Celery, chain, group, chord
app = Celery('complex_flow', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def fetch_data(query):
return [f"item_{i}" for i in range(5)]
@app.task
def process_single(item):
return f"processed_{item}"
@app.task
def combine(results):
return f"Combined {len(results)} items"
@app.task
def finalize(data):
return f"Final: {data}"
# Chain: fetch -> process each in parallel -> combine -> finalize
workflow = (
fetch_data.s("select * from items") |
group([process_single.s() for _ in range(5)]) |
combine.s() |
finalize.s()
)
result = workflow.delay()
print(result.get(timeout=30))
Expected output:
Final: Combined 5 items
Error Handling in Groups
from celery import Celery, group
app = Celery('group_errors', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task(bind=True)
def safe_process(self, item):
if item == 'fail':
raise ValueError(f"Cannot process {item}")
return f"ok_{item}"
# Group with some failing tasks
tasks = [safe_process.s(f"item_{i}") for i in range(5)]
tasks.append(safe_process.s("fail"))
result = group(tasks).delay()
# Get results with propagate=False to see all results (including errors)
outputs = result.get(propagate=False)
for i, out in enumerate(outputs):
status = "OK" if not isinstance(out, Exception) else f"FAIL: {out}"
print(f" Task {i}: {status}")
Expected output:
Task 0: OK
Task 1: OK
Task 2: OK
Task 3: OK
Task 4: OK
Task 5: FAIL: Cannot process fail
Common Mistakes
1. Assuming Group Execution Order
Groups run in parallel. Task 5 may complete before Task 0. Results are ordered by submission, not completion time.
2. Using Group When Chain Is Needed
If Task B needs Task A's result, use chain not group. Groups are for independent parallel tasks.
3. Not Handling Group Errors
A group does not stop on individual task failures. Failed tasks return exceptions. Check each result with propagate=False or handle exceptions in the chord callback.
4. Creating Groups for Single Tasks
A group with one task adds overhead. Use a single task instead. Groups are for 2+ parallel operations.
5. Chord with Empty Header
A chord with an empty header hangs because the callback never fires. Always verify the header has at least one task.
Practice Questions
1. What is the difference between group and chord?
A group executes tasks in parallel and returns a list of results. A chord is a group plus a callback task that receives the aggregated results.
2. How do you get results from a group?
Call result.get(). Returns a list of individual task results in submission order, not completion order.
3. When would you use a chord instead of a group?
When you need to do something after all parallel tasks complete. The chord callback runs after the entire group finishes.
4. How do you handle errors in a group?
Use result.get(propagate=False) to get exceptions as results instead of raising them. Check each result for errors.
Challenge
Design a chord-based system for batch malware analysis: 100 files submitted for scanning in parallel. The chord callback aggregates results, counts clean vs infected files, generates a summary report, and triggers alerts for any threats found. Implement retry for individual scan failures.
FAQ
Mini Project: Parallel Report Generator
# parallel_reports.py
from celery import Celery, group, chord
import time
import random
app = Celery('reports', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task(bind=True)
def generate_chart(self, chart_type, data_id):
print(f"Generating {chart_type} chart for data {data_id}")
time.sleep(random.uniform(0.5, 2.0))
return f"{chart_type}_chart_{data_id}.png"
@app.task(bind=True)
def generate_table(self, data_id):
print(f"Generating table for data {data_id}")
time.sleep(random.uniform(0.3, 1.5))
return f"table_{data_id}.html"
@app.task(bind=True)
def generate_summary(self, data_id):
print(f"Generating text summary for data {data_id}")
time.sleep(random.uniform(0.2, 1.0))
return f"summary_{data_id}.txt"
@app.task
def compile_report(components):
report = {
'charts': [c for c in components if 'chart' in c],
'tables': [c for c in components if 'table' in c],
'summaries': [c for c in components if 'summary' in c],
}
print(f"Report compiled: {len(report['charts'])} charts, "
f"{len(report['tables'])} tables, "
f"{len(report['summaries'])} summaries")
return report
def generate_report(data_id):
header = [
generate_chart.s('bar', data_id),
generate_chart.s('pie', data_id),
generate_table.s(data_id),
generate_summary.s(data_id),
]
result = chord(header)(compile_report.s())
return result.delay()
if __name__ == '__main__':
result = generate_report(42)
report = result.get(timeout=30)
print(f"Generated: {report}")
Expected output:
Generating bar chart for data 42
Generating pie chart for data 42
Generating table for data 42
Generating text summary for data 42
Report compiled: 2 charts, 1 tables, 1 summaries
Generated: {'charts': ['bar_chart_42.png', 'pie_chart_42.png'], 'tables': ['table_42.html'], 'summaries': ['summary_42.txt']}
What's Next
Now that you understand groups and chords, explore monitoring with Flower for tracking task execution, then learn about error handling patterns for production robustness.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro