Celery Workflow Patterns: Advanced Task Orchestration and Composition

In this tutorial, you will learn about Celery Workflow Patterns: Advanced Task Orchestration and Composition. We cover key concepts, practical examples, and best practices to help you master this topic.

Advanced Celery workflow patterns go beyond basic canvas primitives to include dynamic workflow generation based on runtime data, conditional branching, parallel pipelines with different processing paths, and persistent workflows that survive worker restarts.

flowchart TD
    Start[Incoming Data] --> Classify{Classify Data}
    Classify -->|Type A| PipelineA[Pipeline A]
    Classify -->|Type B| PipelineB[Pipeline B]
    Classify -->|Type C| PipelineC[Pipeline C]
    PipelineA --> Aggregate[Aggregate Results]
    PipelineB --> Aggregate
    PipelineC --> Aggregate
    Aggregate --> Notify[Notify User]

What You'll Learn

• Dynamic workflow generation based on runtime conditions
• Conditional branching and routing in workflows
• Parallel pipelines with different processing paths
• Persistent workflows with chord callbacks
• Error recovery and compensating transactions

Why It Matters

Basic canvas workflows handle simple patterns, but real-world applications need dynamic workflows that adapt to data content, handle errors gracefully, and persist across worker restarts. These patterns let you build reliable, adaptable Distributed Systems.

Real-World Use

DodaTech's data ingestion pipeline dynamically generates workflows based on file type. Images go through resize >> thumbnail >> optimize. PDFs go through OCR >> extract >> index. Videos go through transcode >> thumbnail >> analyze. A classifier task determines the file type and generates the appropriate chain at runtime.

Dynamic Workflow Generation

Create workflows based on runtime data:

from celery import Celery, chain, group, chord

app = Celery('workflows', broker='redis://localhost:6379/0')

@app.task
def classify_document(doc_id):
    doc_type = determine_type(doc_id)
    print(f"Document {doc_id} is type: {doc_type}")
    return {"doc_id": doc_id, "type": doc_type}

def determine_type(doc_id):
    types = {1: "invoice", 2: "report", 3: "image"}
    return types.get(doc_id % 3 + 1, "unknown")

@app.task
def process_invoice(doc_info):
    print(f"Processing invoice: {doc_info['doc_id']}")
    return {"status": "invoiced", "doc_id": doc_info["doc_id"]}

@app.task
def process_report(doc_info):
    print(f"Processing report: {doc_info['doc_id']}")
    return {"status": "reported", "doc_id": doc_info["doc_id"]}

@app.task
def process_image(doc_info):
    print(f"Processing image: {doc_info['doc_id']}")
    return {"status": "imaged", "doc_id": doc_info["doc_id"]}

@app.task
def save_result(result):
    print(f"Saved: {result}")
    return result

def build_workflow(doc_id):
    classifier = classify_document.s(doc_id)
    def pipeline_builder(doc_info):
        doc_type = doc_info["type"]
        if doc_type == "invoice":
            return process_invoice.s(doc_info) | save_result.s()
        elif doc_type == "report":
            return process_report.s(doc_info) | save_result.s()
        elif doc_type == "image":
            return process_image.s(doc_info) | save_result.s()
        else:
            return save_result.s({"status": "unknown", "doc_id": doc_info["doc_id"]})
    return classifier | pipeline_builder

for doc_id in range(1, 6):
    workflow = build_workflow(doc_id)
    result = workflow()
    print(f"  Doc {doc_id}: {result}")

Expected output:

Document 1 is type: invoice
Processing invoice: 1
Saved: {'status': 'invoiced', 'doc_id': 1}
  Doc 1: {'status': 'invoiced', 'doc_id': 1}
Document 2 is type: report
Processing report: 2
...

Parallel Pipelines with Different Paths

Run multiple processing pipelines in parallel:

from celery import Celery, group, chord

app = Celery('workflows', broker='redis://localhost:6379/0')

@app.task
def fetch_data(source):
    print(f"Fetching from {source}")
    return {"source": source, "data": f"data_from_{source}"}

@app.task
def transform_a(data_packet):
    result = f"transform_a({data_packet['data']})"
    print(result)
    return result

@app.task
def transform_b(data_packet):
    result = f"transform_b({data_packet['data']})"
    print(result)
    return result

@app.task
def transform_c(data_packet):
    result = f"transform_c({data_packet['data']})"
    print(result)
    return result

@app.task
def merge_results(all_results):
    merged = ", ".join(all_results)
    print(f"Merged: {merged}")
    return merged

sources = ["api_a", "api_b", "api_c"]

pipelines = group(
    fetch_data.s("api_a") | transform_a.s(),
    fetch_data.s("api_b") | transform_b.s(),
    fetch_data.s("api_c") | transform_c.s(),
)

workflow = chord(pipelines, merge_results.s())

result = workflow()
print(f"\nFinal result: {result}")

Expected output:

Fetching from api_a
Fetching from api_b
Fetching from api_c
transform_a(data_from_api_a)
transform_b(data_from_api_b)
transform_c(data_from_api_c)
Merged: transform_a(data_from_api_a), transform_b(data_from_api_b), transform_c(data_from_api_c)

Final result: transform_a(data_from_api_a), transform_b(data_from_api_b), transform_c(data_from_api_c)

Error Handling in Workflows

Handle failures gracefully in complex workflows:

from celery import Celery, chain, group

app = Celery('workflows', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=2)
def unreliable_task(self, value):
    import random
    if random.random() < 0.5:
        raise ValueError(f"Random failure on {value}")
    result = f"processed_{value}"
    print(f"Success: {result}")
    return result

@app.task
def handle_error(request, exc, traceback, task_name=None):
    print(f"Error handler for {task_name}: {exc}")
    return {"status": "failed", "task": task_name, "error": str(exc)}

@app.task
def fallback_task(value):
    result = f"fallback_{value}"
    print(f"Fallback: {result}")
    return result

@app.task
def final_step(results):
    print(f"Final: {results}")
    return results

tasks = [unreliable_task.s(f"item_{i}") for i in range(5)]
group_with_fallbacks = group(
    task | fallback_task.s() for task in tasks
)

workflow = group_with_fallbacks | final_step.s()

for _ in range(3):
    try:
        result = workflow()
        print(f"Workflow result: {result}")
    except Exception as e:
        print(f"Workflow failed: {e}")
    print("---")

Expected output:

Success: processed_item_0
Fallback: fallback_item_1
Success: processed_item_2
...
Final: ['processed_item_0', 'fallback_item_1', ...]
Workflow result: ['processed_item_0', 'fallback_item_1', ...]
---

Common Mistakes

• Building dynamic workflows without serializability — if a workflow is stored or sent to a worker, all signature arguments must be serializable. Avoid passing live database connections or file handles.
• Not handling errors in sub-workflows — a failure in one branch of a parallel pipeline should not block other branches. Use link_error callbacks or fallback tasks for error recovery.
• Creating circular workflow dependencies — ensure workflows are directed acyclic graphs. Circular chains cause infinite execution loops.
• Ignoring workflow timeout — long workflows can run for hours. Set global workflow timeouts and use progress tracking to detect stalled workflows.
• Making workflows too granular — 50 micro-tasks per workflow adds overhead. Batch related operations into single tasks where appropriate. Aim for 5-15 tasks per workflow.

Practice Questions

1. How do you build a workflow that chooses different task paths based on runtime data?
2. How do you handle errors in one branch of a parallel pipeline without affecting other branches?
3. What Serialization considerations apply to dynamic workflows?
4. How do you make workflows persistent across worker restarts?
5. What is the recommended number of tasks per workflow?

Challenge

Build an order processing workflow with: (1) validation (check inventory, check payment), (2) parallel fulfillment (warehouse pick, payment capture, receipt generation), (3) conditional shipping (if Express, use overnight carrier; else standard), (4) notification (email + SMS in parallel), and (5) compensation actions for failures (refund payment if warehouse fails). The workflow must be dynamically generated based on the order type and persist across restarts.

FAQ

Mini Project

Build a configurable ETL workflow engine. Users define a workflow as JSON: sources, transformations, and destinations. The engine: (1) parses the JSON to build a Celery workflow dynamically, (2) creates parallel extraction tasks (one per source), (3) routes each source through configured transformations (chain), (4) merges transformed data, (5) loads into destinations (parallel), and (6) handles errors with configurable retry and fallback behaviors. The workflow persists across worker restarts.

What's Next

Continue with Celery Rate Limiting to control task execution rates. Then explore Celery Task Sets for managing related task groups.