# Canvas Workflows (/integrations/celery/how-to/canvas-workflows)



## Overview [#overview]

Celery's **canvas** lets you compose tasks into larger workflows — sequential pipelines, parallel fan-outs, fan-in callbacks, and batched maps — using signatures (`.s()`) and the primitives `chain`, `group`, `chord`, `starmap`, `map`, and `chunks`. Each composed task still rides KubeMQ [Queues](/learn/queues) under the hood, and because `kubemq-celery` is a Kombu transport that registers the `kubemq://` URL scheme, **every canvas primitive works unchanged on KubeMQ**. There is no canvas-specific API to learn and no code to rewrite: the same workflow that ran on Redis or RabbitMQ runs on KubeMQ once you switch the broker URL.

<Callout type="info">
  Canvas composition is a Celery-level feature, not a broker feature. The transport's job is to deliver task messages and (optionally) store results — chaining, grouping, and callback orchestration all happen in Celery itself.
</Callout>

The only behavioral difference is **chord unlocking**. Redis provides an O(1) native chord unlock; KubeMQ uses Celery's portable **polling fallback** (the `chord_unlock` task polls for group completion). This is covered in detail in the [Chord](#chord-group--callback) section below.

All examples on this page configure the broker and result backend the same way — point both at a `kubemq://` URL:

```python title="app.py"
import os

from celery import Celery

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport

app = Celery("canvas_app")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
        "task_serializer": "json",
        "result_serializer": "json",
        "accept_content": ["json"],
    }
)
```

You also need a reachable KubeMQ broker. The fastest way to get one locally is Docker:

<RunKubeMQ ports="[50000, 9090]" />

Port `50000` is the gRPC port the transport connects to; port `9090` is the shared HTTP server (REST and the `/health` probe) — `curl http://localhost:9090/health` confirms the broker is up as tasks flow through your workflow.

<Callout type="warn">
  `import kubemq_celery` must run before Celery resolves the broker URL — the import is what registers the `kubemq://` scheme with Kombu. Keep it at the top of your app module even if your editor flags it as unused. See [Getting Started](/integrations/celery/tutorials/getting-started) for the full setup walkthrough.
</Callout>

## Chain: sequential pipeline [#chain-sequential-pipeline]

A `chain` runs tasks one after another, passing the result of each task as the **first argument** of the next. Build the pipeline from signatures and call it — the output of `add(2, 3)` becomes the `x` argument of `multiply`, and so on.

```python title="chain.py"
from celery import chain


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def multiply(x: int, y: int) -> int:
    return x * y


@app.task
def subtract(x: int, y: int) -> int:
    return x - y


# Pipeline: add(2, 3) -> multiply(result, 10) -> subtract(result, 5)
# Expected: (2 + 3) * 10 - 5 = 45
workflow = chain(
    add.s(2, 3),
    multiply.s(10),
    subtract.s(5),
)

result = workflow.apply_async()
value = result.get(timeout=10)
print(f"Result: {value}")  # 45
```

Chains can be arbitrarily long. Each step waits for the previous one to complete, and the final `AsyncResult` resolves to the output of the last task:

```python title="chain.py"
# add(1,1)=2 -> multiply(2,3)=6 -> add(6,4)=10 -> multiply(10,2)=20
workflow = chain(
    add.s(1, 1),
    multiply.s(3),
    add.s(4),
    multiply.s(2),
)

value = workflow.apply_async().get(timeout=10)
print(f"Result: {value}")  # 20
```

## Group: parallel fan-out [#group-parallel-fan-out]

A `group` dispatches its members in parallel and collects all of their results. Calling `.get()` on the returned `GroupResult` blocks until every member completes, then returns the list of results.

```python title="group.py"
from celery import group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def square(n: int) -> int:
    return n * n


# Static group: four additions in parallel
workflow = group(
    add.s(1, 2),
    add.s(3, 4),
    add.s(5, 6),
    add.s(7, 8),
)
values = workflow.apply_async().get(timeout=10)
print(f"Results: {values}")  # [3, 7, 11, 15]
```

Groups can also be built dynamically from an iterable of signatures — useful when the number of parallel tasks is data-driven:

```python title="group.py"
# Dynamic group: square each value in 1..5
workflow = group(square.s(i) for i in range(1, 6))
values = workflow.apply_async().get(timeout=10)
print(f"Results: {values}")  # [1, 4, 9, 16, 25]

# Large group: 20 parallel additions
workflow = group(add.s(i, i) for i in range(20))
values = workflow.apply_async().get(timeout=30)
print(len(values))  # 20
```

Group result collection is backed by the result backend. On KubeMQ, group metadata is stored on `celery-group-{group_id}` Queue channels, and member results on per-task `celery-result-{task_id}` channels read back with a non-destructive peek. See [Result Backend](/integrations/celery/how-to/result-backend) for how this works.

## Chord: group + callback [#chord-group--callback]

A `chord` runs a group of tasks in parallel and then invokes a **callback** with the list of all group results. It is the canonical fan-out / fan-in pattern: parallel work followed by an aggregation step.

```python title="chord.py"
from celery import chord, group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def sum_results(values: list[int]) -> int:
    """Chord callback — sums the group results."""
    return sum(values)


# Run three additions in parallel, then sum the results.
workflow = chord(
    group(
        add.s(1, 2),
        add.s(3, 4),
        add.s(5, 6),
    ),
    sum_results.s(),
)
value = workflow.apply_async().get(timeout=15)
print(f"sum([3, 7, 11]) = {value}")  # 21
```

### Chord uses the polling fallback on KubeMQ [#chord-uses-the-polling-fallback-on-kubemq]

This is the one place where KubeMQ behaves differently from Redis. Redis provides a **native O(1) chord unlock** — the broker atomically decrements a counter as each group member completes and fires the callback immediately when the count reaches zero. KubeMQ does not implement that broker-side primitive, so `kubemq-celery` uses **Celery's portable polling fallback**: a built-in `chord_unlock` task periodically polls the result backend for group completion and dispatches the callback once all members have finished.

<Callout type="warn">
  **Chord requires a result backend.** The `chord_unlock` task polls group results to detect completion, so a chord will not unlock unless `result_backend` is configured. Set `result_backend="kubemq://..."` (or any supported backend). See [Result Backend](/integrations/celery/how-to/result-backend).
</Callout>

Two practical consequences:

* **Latency** — the callback fires on the next poll after the last member completes, adding roughly **1–2 seconds** versus Redis's immediate unlock. Use generous timeouts on `result.get()` for chord-based workflows.
* **Correctness is unchanged** — the polling fallback is a standard Celery mechanism; results and callback semantics are identical, only the unlock timing differs.

```python title="chord.py"
# Chord built from a dynamic group — same polling-fallback behavior.
workflow = chord(
    group(add.s(i, i * 2) for i in range(1, 6)),
    sum_results.s(),
)
value = workflow.apply_async().get(timeout=15)
print(f"sum([3, 6, 9, 12, 15]) = {value}")  # 45
```

### Chord error handling [#chord-error-handling]

If **any** chord member fails, the callback is **not** executed. Attach an error callback with `link_error` to observe the failure, and propagate the exception to the caller with `result.get(..., propagate=True)`.

```python title="chord_error_handling.py"
from celery import Task, chord, group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def failing_task(x: int) -> int:
    raise ValueError(f"Intentional failure for input {x}")


@app.task
def sum_results(values: list[int]) -> int:
    return sum(values)


@app.task
def on_chord_error(request: Task, exc: Exception, traceback: str | None) -> None:
    """Error callback — runs when a chord member fails."""
    print(f"Chord error! Task {request.id} failed: {exc}")


workflow = chord(
    group(
        add.s(1, 2),
        failing_task.s(99),
        add.s(5, 6),
    ),
    sum_results.s(),
)
workflow.link_error(on_chord_error.s())

try:
    result = workflow.apply_async()
    result.get(timeout=15, propagate=True)
except Exception as exc:
    print(f"Chord failed as expected: {type(exc).__name__}: {exc}")
```

For broker-level failure handling — dead-letter queues, retries, and acknowledgment semantics — see [Error Handling](/integrations/celery/how-to/error-handling).

## Chains of groups and multi-stage workflows [#chains-of-groups-and-multi-stage-workflows]

Canvas primitives nest. A common pattern is a `chain` whose first step is a `chord` (a parallel stage that collects into a single result), followed by sequential transformation steps. Each stage waits for the previous group to fully complete before the next begins.

```python title="chain_of_groups.py"
from celery import chain, chord, group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def collect(values: list[int]) -> list[int]:
    """Pass-through collector — fans the group results into the chain."""
    return values


@app.task
def double_all(values: list[int]) -> list[int]:
    return [v * 2 for v in values]


@app.task
def sum_all(values: list[int]) -> int:
    return sum(values)


# Stage 1 (chord): parallel additions -> collect           => [3, 7, 11]
# Stage 2 (task):  double every element                    => [6, 14, 22]
# Stage 3 (task):  sum everything                           => 42
workflow = chain(
    chord(
        group(add.s(1, 2), add.s(3, 4), add.s(5, 6)),
        collect.s(),
    ),
    double_all.s(),
    sum_all.s(),
)
value = workflow.apply_async().get(timeout=20)
print(f"Result: {value}")  # 42
```

The same composition scales to realistic pipelines. The example below models an **ETL** workflow — extract data from several sources in parallel, transform the merged result through sequential steps, then load it:

```python title="complex_workflow.py"
import time

from celery import chain, chord, group


@app.task
def fetch_sales_data(region: str) -> dict:
    """Extract — simulate fetching from a regional database."""
    time.sleep(0.3)
    data = {
        "us": {"region": "us", "revenue": 50000, "orders": 120},
        "eu": {"region": "eu", "revenue": 35000, "orders": 85},
        "apac": {"region": "apac", "revenue": 28000, "orders": 65},
    }
    return data.get(region, {"region": region, "revenue": 0, "orders": 0})


@app.task
def merge_datasets(datasets: list[dict]) -> dict:
    """Transform — chord callback that merges the extracted datasets."""
    return {
        "regions": [d["region"] for d in datasets],
        "total_revenue": sum(d["revenue"] for d in datasets),
        "total_orders": sum(d["orders"] for d in datasets),
    }


@app.task
def enrich_summary(summary: dict) -> dict:
    summary["avg_order_value"] = round(
        summary["total_revenue"] / max(summary["total_orders"], 1), 2
    )
    summary["region_count"] = len(summary["regions"])
    return summary


@app.task
def save_report(report: dict) -> dict:
    """Load — simulate writing to a data warehouse."""
    time.sleep(0.2)
    return {"status": "saved", **report}


workflow = chain(
    # Extract: parallel fetch from three regions, merged by the chord callback.
    chord(
        group(
            fetch_sales_data.s("us"),
            fetch_sales_data.s("eu"),
            fetch_sales_data.s("apac"),
        ),
        merge_datasets.s(),
    ),
    # Transform: enrich the merged summary.
    enrich_summary.s(),
    # Load: persist the final report.
    save_report.s(),
)
report = workflow.apply_async().get(timeout=30)
print(report["status"])  # saved
```

Because the parallel stages use chords, these multi-stage workflows rely on the polling fallback described above — budget for the extra unlock latency at each chord boundary when setting timeouts.

<Mermaid
  chart="flowchart LR
    subgraph Extract
        US[&#x22;fetch_sales_data(us)&#x22;]
        EU[&#x22;fetch_sales_data(eu)&#x22;]
        AP[&#x22;fetch_sales_data(apac)&#x22;]
    end
    US --> M[&#x22;merge_datasets<br/>(chord callback)&#x22;]
    EU --> M
    AP --> M
    M --> E[&#x22;enrich_summary&#x22;]
    E --> L[&#x22;save_report&#x22;]"
/>

## starmap, map, and chunks for batching [#starmap-map-and-chunks-for-batching]

When you need to apply one task across a large iterable, `starmap`, `map`, and `chunks` avoid the overhead of dispatching a separate task message per item.

`task.starmap(iterable)` applies the task to each tuple in the iterable, **unpacking** each tuple into positional arguments. It runs as a single task that iterates locally:

```python title="starmap.py"
@app.task
def add(x: int, y: int) -> int:
    return x + y


pairs = [(1, 2), (3, 4), (5, 6), (7, 8)]
values = add.starmap(pairs).apply_async().get(timeout=10)
print(values)  # [3, 7, 11, 15]
```

`task.map(iterable)` is the single-argument variant — each item is passed as the sole argument:

```python title="map.py"
@app.task
def double(n: int) -> int:
    return n * 2


values = double.map([1, 2, 3, 4, 5]).apply_async().get(timeout=10)
print(values)  # [2, 4, 6, 8, 10]
```

`task.chunks(items, n)` splits a large iterable into batches of size `n`, dispatching **each chunk as a separate task** so the batches can be processed in parallel across workers. The result is a list of per-chunk result lists:

```python title="chunks.py"
@app.task
def add(x: int, y: int) -> int:
    return x + y


# 10 additions split into chunks of 3.
items = [(i, i + 1) for i in range(10)]
chunked = add.chunks(items, 3).apply_async().get(timeout=15)

# Flatten the per-chunk results.
flat = [v for chunk in chunked for v in chunk]
print(flat)  # [1, 3, 5, 7, 9, 11, 13, 15, 17, 19]
```

<Callout type="info">
  Use `starmap`/`map` when per-item work is cheap and you want a single task; use `chunks` when the dataset is large and you want the batches distributed across workers.
</Callout>

## Immutable signatures [#immutable-signatures]

By default a chained signature is **mutable** (`.s()`): the previous task's result is prepended as its first argument. An **immutable** signature (`.si()`) ignores the parent result entirely — its arguments are fixed. Use `.si()` when a step should run for its side effects, or with a fixed input, regardless of what came before.

```python title="immutable_signatures.py"
from celery import chain


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def multiply(x: int, y: int) -> int:
    return x * y


# Mutable: add(2,3)=5 flows into multiply(5, 10) = 50
mutable = chain(add.s(2, 3), multiply.s(10))
print(mutable.apply_async().get(timeout=10))  # 50

# Immutable: add(2,3) runs, but its result is discarded.
# multiply(100, 10) = 1000 — the 5 is ignored.
immutable = chain(add.s(2, 3), multiply.si(100, 10))
print(immutable.apply_async().get(timeout=10))  # 1000
```

You can mix the two in one chain — mutable steps pass results through, immutable steps reset the argument flow:

```python title="immutable_signatures.py"
@app.task
def log_and_return(value: int) -> int:
    print(f"  [log_and_return] received: {value}")
    return value


# add(5,5)=10 -> log_and_return(10)=10 -> multiply.si(7,3)=21 (10 discarded)
mixed = chain(
    add.s(5, 5),
    log_and_return.s(),
    multiply.si(7, 3),
)
print(mixed.apply_async().get(timeout=10))  # 21
```

## Result backend requirement [#result-backend-requirement]

Any canvas pattern that **collects or aggregates results** depends on the result backend:

* `group` — `GroupResult.get()` reads each member's stored result.
* `chord` — the `chord_unlock` task polls the backend to detect group completion before firing the callback.
* Chain results — `result.get()` on the final step reads the stored return value.

Configure the KubeMQ queue-peek backend by pointing `result_backend` at a `kubemq://` URL:

```python title="app.py"
app.config_from_object(
    {
        "broker_url": "kubemq://localhost:50000",
        "result_backend": "kubemq://localhost:50000",
        "result_expires": 3600,
    }
)
```

Results are stored as KubeMQ Queue messages on `celery-result-{task_id}` channels and read back with a non-destructive peek, and group metadata lives on `celery-group-{group_id}` channels — so no external Redis or database is required.

<Callout type="warn">
  Without a result backend, `group` and `chord` cannot return results and a chord will never unlock. The KubeMQ backend caps result expiry at &#x2A;*24 hours (86400 seconds)**; Celery's default `result_expires` of 24 hours already matches this maximum.
</Callout>

For the full backend walkthrough and configuration options, see [Result Backend](/integrations/celery/how-to/result-backend).

## Testing canvas in eager mode [#testing-canvas-in-eager-mode]

You can unit-test canvas composition **without a broker or result backend** by enabling eager mode, where tasks run synchronously in the calling process. Set `task_always_eager=True` (and `task_eager_propagates=True` so errors surface as exceptions), then assert on results.

```python title="test_canvas.py"
from celery import Celery, chain, chord, group

import kubemq_celery  # noqa: F401

app = Celery("test_canvas")
app.conf.update(
    task_always_eager=True,
    task_eager_propagates=True,
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
)


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def double(x: int) -> int:
    return x * 2


@app.task
def sum_list(values: list[int]) -> int:
    return sum(values)


def test_chain_basic():
    workflow = chain(add.s(2, 3), double.s())
    result = workflow.apply()
    assert result.result == 10


def test_group_basic():
    workflow = group(add.s(1, 1), add.s(2, 2), add.s(3, 3))
    values = workflow.apply().get()
    assert sorted(values) == [2, 4, 6]


def test_chord_basic():
    workflow = chord(
        group(add.s(1, 1), add.s(2, 2), add.s(3, 3)),
        sum_list.s(),
    )
    result = workflow.apply()
    assert result.result == 12


def test_immutable_signatures():
    # add.si(10, 20) ignores the result of add(1, 2).
    workflow = chain(add.s(1, 2), add.si(10, 20))
    result = workflow.apply()
    assert result.result == 30
```

Run the tests with `pytest`:

```bash
pytest test_canvas.py -v
```

<Callout type="info">
  Eager mode runs tasks in-process and bypasses the broker entirely, so it validates canvas composition and result-passing logic — not transport behavior. Note that in eager mode chords run synchronously without the `chord_unlock` polling path; cover the polling-fallback timing against a live KubeMQ broker in an integration test.
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Scheduling & Delayed Delivery" href="/integrations/celery/how-to/scheduling" description="Use countdown, eta, and Celery Beat — backed by KubeMQ's native delay_in_seconds." />

  <Card title="Result Backend" href="/integrations/celery/how-to/result-backend" description="Configure the queue-peek result backend that group and chord workflows depend on." />

  <Card title="Error Handling" href="/integrations/celery/how-to/error-handling" description="Dead-letter queues, retries, and chord failure handling on KubeMQ." />
</Cards>