Celery Canvas Workflows: Chain, Group, Chord

How to express sequential, parallel, and aggregating task flows declaratively instead of wiring callbacks together by hand.

The limits of wiring tasks by hand

Dispatching a single task to an asynchronous task queue is easy. The trouble starts when tasks have to be composed: A finishes, then B, then C runs in parallel over B's result, and D is called once everything is collected. Wiring that flow with callbacks scatters the result-passing code across workers.

Canvas is Celery's workflow composition system. It groups tasks into Chain (sequential), Group (parallel), and Chord (scatter-gather) so that multi-step work is expressed declaratively. Instead of scattered callbacks, the flow itself becomes a single value you can pass around.

Signature, the unit of composition

A signature wraps a task invocation in a serializable dictionary. Rather than running the function, it carries only the plan: call this task with these arguments. Every Canvas construct composes signatures.

python

from celery import signature
 
# explicit declaration
sig = signature('myapp.tasks.add', args=(2, 2), options={'queue': 'default'})
 
# shorthand (the common form)
sig = add.s(2, 2)

.s() creates a partial signature. The empty first argument slot is filled with the previous task's result. .si() creates an immutable signature instead, ignoring the previous result and using only its fixed arguments.

python

# partial: receives the previous result as the first argument
sig = mul.s(8)        # mul(previous_result, 8)
 
# immutable: ignores the previous result, always fixed
sig = add.si(2, 2)    # always add(2, 2) = 4

This distinction determines whether results propagate through a chain or stop there. Use .s() when a step consumes the previous result, and .si() for steps such as notifications that do not care about it.

Sequential execution with Chain

A chain runs tasks in order and passes each result as the first argument of the next task. Link them with the pipe operator | or wrap them in chain().

python

from celery import chain
 
# pipe operator (preferred)
workflow = add.s(4, 4) | mul.s(8) | mul.s(10)
 
# explicit chain()
workflow = chain(add.s(4, 4), mul.s(8), mul.s(10))
 
result = workflow()        # same as workflow.apply_async()
print(result.get())        # 640

The computation runs as ((4 + 4) × 8) × 10 = 640. Intermediate results are reachable by walking back through parent.

python

print(result.get())                 # 640  (final)
print(result.parent.get())          # 64   (mul(8, 8))
print(result.parent.parent.get())   # 8    (add(4, 4))

If one step fails, the chain stops there and no later step runs. That makes chains a fit for flows where each result is a precondition for the next step.

Parallel execution with Group

A group dispatches several tasks at once and collects the results into a list. Use it when the tasks have no dependency on one another.

python

from celery import group
 
# run 10 add tasks in parallel
tasks = group(add.s(i, i) for i in range(10))
result = tasks()
 
print(result.get())   # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

Results come back in dispatch order, so items can be matched by index. Individual task states are available through result.results.

python

for res in result.results:
    print(res.state, res.result)

A group only collects results; it does not include a step that processes them once collected. Adding that "and then" step to the flow is what Chord does.

Scatter and gather with Chord

A chord runs a group in parallel and then executes a single callback once every result has arrived. It is the map-reduce pattern: split the work and scatter it, then gather the results in one place.

The diagram shows ten add tasks running in parallel and their results converging into a single summing callback. A chord is the combination of a header (the group) and a body (the callback), and the two forms below behave identically.

python

from celery import chord, group
 
@app.task
def xsum(values):
    return sum(values)
 
header = group(add.s(i, i) for i in range(10))
 
# 1) explicit chord
workflow = chord(header)(xsum.s())
 
# 2) pipe (group | callback)
workflow = (group(add.s(i, i) for i in range(10)) | xsum.s())()
 
print(workflow.get())   # 90

The sum of 90 matches the sum of [0, 2, ..., 18]. Unlike a group, a chord folds the collecting callback into the same flow.

Chord synchronization and its constraints

A chord has to decide when every header task has finished before firing the callback. How that decision is made depends on the result backend, the store that holds task results.

Result backend	How completion is detected	Cost
Redis	RPUSH results onto a list, check length with LLEN	Atomic counter, no polling
Database	celery.chord_unlock polls every second	Repeated DB queries, relatively slow

Redis pushes each result onto a list as a header task finishes, counts the length, and fires the callback the moment it reaches the expected count. With a database backend the celery.chord_unlock task checks every second whether the result set is complete, which adds latency and load. The difference comes from Celery implementing chord synchronization differently per result backend.

Chords therefore have preconditions. Violate any one of the three below and they will not work.

result_backend must be configured.
Header tasks must have ignore_result=False for their results to be collected.
The RPC backend (rpc://) does not support chords.

python

# chord fails on the RPC backend
app.conf.result_backend = 'rpc://'
workflow = (group(tasks) | callback)()   # fails
 
# works on a real backend such as Redis
app.conf.result_backend = 'redis://redis:6379/1'

Checking the configuration catches most common chord failures before they happen. Watch out for ignore_result=True mixed into a header, which leaves the gather step waiting forever.

Common composition patterns

The three primitives nest to build larger flows. A common shape is a group inside a chain, the ETL (extract-transform-load) pattern.

python

from celery import chain, group
 
workflow = chain(
    group(fetch.s(url) for url in urls),   # parallel extract
    transform.s(),                          # transform
    load.s(),                               # load
)
print(workflow().get())   # {'count': 3}

Wrapping several chains in a group does the opposite: independent pipelines run in parallel. Each URL goes through fetch → parse → validate → store in sequence, while different URLs are processed concurrently.

python

workflows = group(
    chain(fetch.s(url), parse.s(), validate.s(), store.s())
    for url in urls
)

For branching, self.replace() swaps the workflow at runtime. Celery has no if/else primitive, so a task builds the matching chain internally and replaces itself with raise self.replace(...).

python

@app.task(bind=True)
def route_by_type(self, data):
    if data['type'] == 'urgent':
        workflow = chain(process_urgent.s(data), notify_admin.s())
    else:
        workflow = chain(process_normal.s(data), queue_for_review.s())
    raise self.replace(workflow)   # must be invoked with raise

self.replace() only replaces the current task with the new workflow when it is used with raise. Conditional branching that static composition cannot express is handled this way.

Error handling and retries

The longer the composition, the more it matters how a single step's failure is handled. Success and failure callbacks attach through link and link_error.

python

@app.task
def on_error(request, exc, traceback):
    """Receives the failed task's request, exception, and traceback."""
    print(f"failed: {exc}")
 
# attach an errback to the whole chain
workflow = (task1.s() | task2.s() | task3.s())
result = workflow.apply_async(link_error=on_error.s())

For transient errors, retrying comes before callbacks. Work that will succeed on a second attempt, such as a network call, gets automatic retries.

python

@app.task(
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,        # wait 1, 2, 4, 8 seconds
    retry_backoff_max=600,     # up to 10 minutes
    retry_jitter=True,         # randomize the wait
    max_retries=5,
)
def call_api(url):
    return requests.get(url, timeout=10).json()

retry_backoff grows the retry interval exponentially, and retry_jitter spreads those intervals randomly so many tasks do not retry at the same instant. As a workflow grows, do not bind it into one enormous composition. Dispatching stage by stage and passing IDs instead of objects reduces serialization risk and makes recovery easier.

Summary

Chain runs tasks in sequence passing results forward, Group runs independent tasks in parallel, and Chord gathers parallel results into a callback. The unit of composition is the signature, and the distinction between .s() taking the previous result and .si() ignoring it determines how results propagate. Chord completion is decided by the result backend, so a configured result_backend and ignore_result=False on header tasks are preconditions. Large workflows recover better when they are decomposed into stages that pass IDs rather than composed all at once.