## TL;DR
Decorate the task with `@task` and call `.expand(arg=[...])` to fan out one task per list element at runtime; collect results with `.expand()` on a downstream task or reduce manually with XCom. Mapping replaces code-generated static tasks with data-driven parallelism.

```text
airflow dynamic task mapping expand reduce
```

## Use this when
- The number of parallel tasks depends on runtime data
- You currently generate tasks in a Python loop
- You need to aggregate mapped task outputs

## Not for this skill when
- You are configuring deferrable operators
- You need SLA miss alerts
- Sensors time out

## Steps

1. The basic expand pattern:

```python
from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def mapped_pipeline():
    @task
    def list_tables():
        return ["orders", "users", "events"]
    @task
    def process(table: str):
        print(f"processing {table}")
    process.expand(table=list_tables())

mapped_pipeline()
```
Expected output: three `process` task instances at runtime, one per table. The list comes from an upstream task, so the fan-out is data-driven, not hardcoded.

2. Expand over multiple arguments with `expand_kwargs`:

```python
@task
def load(table: str, date: str):
    ...
load.expand_kwargs([{"table": t, "date": d} for t in tables for d in dates])
```
Expected output: one task instance per dict. Use `expand()` for a single list argument, `expand_kwargs()` for combinations.

3. Reduce: aggregate mapped outputs in a downstream task:

```python
@task
def process(table: str) -> int:
    return row_count(table)
@task
def summarize(counts: list[int]):
    print(f"total rows: {sum(counts)}")
counts = process.expand(table=list_tables())
summarize(counts)
```
Expected output: `summarize` receives the list of all mapped return values. Passing a mapped XCom into a regular task automatically reduces it.

4. Control parallelism so a 10,000-element expand does not crush the system:

```python
@task(max_active_tis_per_dag=16)
def process(table: str):
    ...
```
Expected output: at most 16 mapped instances running at once per DAG run. Without a cap, a large expand can starve every other DAG of workers.

5. Know the limits of mapping vs static tasks:

```text
Mapped tasks: great for data-driven fan-out; the graph shows one
node that expands at runtime. Debugging one bad element means
finding it among N instances.
Static loop tasks: visible individually in the graph; better when
the set is fixed and small.
```
Expected output: the right choice per use case. Do not map over a fixed list of 3 known tables just because you can; a static loop is clearer.

## Variant phrasings

### airflow dynamic tasks variable number
`.expand()` over an upstream task's output list (step 1). The count is decided at runtime.

### airflow mapped task xcom aggregate
Pass the mapped result into a downstream task; Airflow collects the list (step 3).

### airflow expand vs static tasks
Expand for runtime-determined parallelism, static loops for fixed small sets (step 5).

## Why it happens
Classic Airflow builds the DAG at parse time, so "N tasks" meant N task objects in code. Dynamic task mapping defers the fan-out to runtime: one mapped task definition expands into N task instances based on actual data, which is what data pipelines usually need.

## Edge cases
- The expand argument must be a list (or XCom resolving to one); a generator or lazy object fails at runtime.
- Mapped tasks need Airflow 2.3+; `max_active_tis_per_dag` needs 2.6+.
- Very large expands (100k+) strain the scheduler and the metadata DB; chunk the input list.
- Mapped task instance logs are per map index; the UI groups them, use the index filter to find the failing element.

## Provenance

Resolved from the public thread: https://vectle.com/posts/pst_c91K0PkO3Bg4LOdvMakgbA