Airflow 3 in Python: TaskFlow, Task Groups, and Dynamic Task Mapping for Modern Data Pipelines (2026)

Updated: September 10, 2026

Airflow 3 in Python is Apache Airflow's first major release since 2.0, built around a Task Execution API, a new airflow.sdk Task SDK for authoring DAGs, native DAG versioning, and asset-driven scheduling that replaces the older Datasets concept. I run a team that has ported roughly 900 DAGs from Airflow 2.10 to 3.3 over the last year, and honestly, the short answer is this: TaskFlow, task groups, and dynamic task mapping are still how you write DAGs. But where they live, how they get executed, and how you promote versions have all changed. This is the reference I wish I had on day one.

• Airflow 3.0 went GA in April 2025; 3.3.1 shipped in November 2025 and is the release most teams should target in 2026.
• Import @dag, @task, and @task_group from airflow.sdk, not from airflow.decorators. The old paths still work but are frozen and will be removed.
• Dynamic task mapping uses .expand() and .expand_kwargs(). Combine it with a reducer task for map-and-reduce style fan-outs.
• Datasets are now Assets, defined with the @asset decorator, and drive asset-aware scheduling and lineage in the new React UI.
• Native DAG versioning pins each run to the serialized DAG that produced it. The rerun_with_latest_version flag controls whether backfills use current or historical code.
• The Task SDK decouples task execution from the scheduler, so workers can run in containers, edge nodes, or via the Edge Executor without a full Airflow install.

What is new in Apache Airflow 3?

Apache Airflow 3.0 was the first major release since 2.0 in December 2020, and it's the biggest change to the internals since I started using Airflow at The New York Times. The Airflow 3 GA announcement lists four headline changes I actually feel every day: a service-oriented Task Execution API (AIP-72), the new airflow.sdk Task SDK, native DAG versioning (AIP-65/66), and Assets, which are the successor to Datasets (AIP-74/75). The React-based UI is the visible tip of that iceberg. Airflow 3.1 landed in July 2025 with the Edge Executor going generally available, and the current 3.3.1 release (November 2025) adds first-class asset partitioning, a Language Task SDK for Java and Go, and the @skip_if/@run_if decorators.

The reason to care is operational. In 2.x, the scheduler had to embed the worker runtime, so upgrades were coupled and you couldn't easily run tasks outside the cluster. In 3.x, the scheduler talks to workers over the Task Execution API, and the worker only needs the Task SDK, which is a much smaller wheel. That's why the Edge Executor works: a laptop or a K8s pod anywhere with HTTPS to the API server can be a worker. It's also why the Task SDK is worth learning even if you're happy with your DAGs. Importing from airflow.sdk is the only forward-compatible surface, and everything else is a shim.

The Task SDK and TaskFlow API in Airflow 3

In Airflow 2.x I imported @dag, @task, and @task_group from airflow.decorators. In 3.x that still works, but the officially supported import path is airflow.sdk, and I now flag any PR that reaches back into airflow.models or airflow.operators. The Task SDK ships as a separate wheel (apache-airflow-task-sdk) precisely so a worker container doesn't need the whole Airflow install. Here is the smallest useful modern DAG. I really do run this to sanity-check a fresh 3.3 install:

from __future__ import annotations
from datetime import datetime, timedelta

from airflow.sdk import dag, task


@dag(
    dag_id="daily_playback_reconcile",
    schedule="0 3 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["reconcile", "spotify-events"],
)
def daily_playback_reconcile():

    @task
    def list_partitions(logical_date: datetime) -> list[str]:
        # A real implementation would list S3 keys; keeping it inline for clarity.
        d = logical_date.strftime("%Y-%m-%d")
        return [f"s3://events/playback/date={d}/hour={h:02d}/" for h in range(24)]

    @task(pool="reconcile_pool", max_active_tis_per_dag=8)
    def reconcile(prefix: str) -> dict[str, int]:
        # Reconciliation logic returns a summary dict per hour.
        return {"prefix": prefix, "matched": 0, "missing": 0}

    @task
    def summarize(results: list[dict[str, int]]) -> None:
        total_missing = sum(r["missing"] for r in results)
        print(f"Total missing rows across 24 hours: {total_missing}")

    prefixes = list_partitions()
    results = reconcile.expand(prefix=prefixes)
    summarize(results)


daily_playback_reconcile()

Three things are worth pointing out. First, the DAG is a callable, and the last line invokes it, which is how Airflow discovers the DAG at parse time. Second, logical_date is injected automatically because I annotated the parameter, so you no longer need get_current_context() for the common cases. Third, reconcile.expand(prefix=prefixes) is dynamic task mapping in action; the scheduler creates one task instance per element of prefixes at run time, not at parse time. The Task SDK's dynamic task mapping docs spell out the full API, including the newer .expand_kwargs(), which lets you map over dictionaries so each instance gets multiple keyword arguments.

How do you use task groups in Airflow 3?

Task groups are still my preferred way to bundle related tasks into a single collapsible box in the UI, and the decorator API is unchanged in 3.x. Only the import path moves to airflow.sdk. A task group is not an execution boundary (it does not add retries, pooling, or SLAs by itself); it is purely a namespacing and rendering construct. I use them for two reasons: to make a 40-task DAG visually navigable, and to attach shared defaults (retries, pool) to a subset of tasks without polluting the DAG-level defaults.

from airflow.sdk import dag, task, task_group


@dag(schedule="@hourly", start_date=datetime(2026, 1, 1), catchup=False)
def hourly_ingest():

    @task_group(group_id="ingest_kafka", default_args={"retries": 3})
    def ingest_kafka(topic: str):

        @task
        def pull(topic: str) -> str:
            return f"/tmp/{topic}.arrow"

        @task
        def validate(path: str) -> str:
            # Pandera check, schema drift log, etc.
            return path

        @task
        def load(path: str) -> None:
            print(f"COPY INTO warehouse.{topic} FROM {path}")

        load(validate(pull(topic)))

    for t in ["plays", "skips", "seeks"]:
        ingest_kafka.override(group_id=f"ingest_{t}")(t)


hourly_ingest()

Two patterns from that snippet worth stealing. Calling ingest_kafka.override(group_id=f"ingest_{t}")(t) in a loop is how you get three cleanly labelled groups without copy-pasting the inner tasks. You can override group_id, tooltip, and default_args per invocation. And attaching default_args={"retries": 3} at the group level means only Kafka-side tasks retry aggressively; the group is the natural scope for that policy. If you need real isolation (a separate executor, resource limits, or a K8s namespace), reach for an @task.kubernetes() or the TaskGroupCommand-style patterns from Astronomer's blog rather than a plain @task_group.

What is dynamic task mapping in Airflow?

Dynamic task mapping lets the scheduler create N task instances at run time from the output of an upstream task, instead of the DAG author hard-coding N at parse time. It landed in 2.3, was hardened through 2.9, and in 3.x picks up a fix that finally makes the mapped-task statistics correct. (Previously the counters double-counted or dropped states on partial failures. I have the Slack scars.) The core methods are .expand(), which maps over positional or keyword iterables one-per-arg, and .expand_kwargs(), which maps over a list of dicts.

from airflow.sdk import dag, task
from datetime import datetime


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def score_regions():

    @task
    def list_regions() -> list[dict]:
        # Each dict becomes the kwargs for one mapped task instance.
        return [
            {"region": "us-east-1", "shard": 0},
            {"region": "us-east-1", "shard": 1},
            {"region": "eu-west-1", "shard": 0},
            {"region": "ap-south-1", "shard": 0},
        ]

    @task(max_active_tis_per_dag=4)  # cap concurrency of the mapped task
    def score(region: str, shard: int) -> dict:
        # Real work: pull events, run pandas categorical group-by, write parquet.
        return {"region": region, "shard": shard, "rows": 12_345}

    @task
    def rollup(results: list[dict]) -> None:
        # Classic map-and-reduce: the reducer receives the full list.
        total = sum(r["rows"] for r in results)
        print(f"Scored {total:,} rows across {len(results)} shards")

    rollup(score.expand_kwargs(list_regions()))


score_regions()

Three production rules I now enforce in code review. One: put a real cap on mapped concurrency (max_active_tis_per_dag, a pool, or both). A mapped task with 40k instances will happily saturate your executor and page you at 3 AM. Two: never map over an unbounded iterable without a guard. I add a len(xs) < 10_000 assertion in the upstream task and let it fail loudly. Three: reducers that receive the list of results must be prepared for the "some failed" case. In 3.x you can set trigger_rule="all_done" on the reducer and inspect which upstream instances actually returned data.

Assets and event-driven scheduling

Assets are the biggest conceptual change in 3.x. In 2.x, a Dataset was a URI that a producer DAG "updated" and a consumer DAG could depend on to trigger. In 3.x, that idea is elevated to an Asset: a first-class object with a name, a URI, a producer function, and lineage, defined with the @asset decorator (AIP-74 and AIP-75). It is closer to how Dagster models data than to legacy Airflow, and it is the recommended scheduling mechanism for anything data-triggered.

from airflow.sdk import asset, dag, task
from datetime import datetime


@asset(schedule="@hourly", uri="s3://warehouse/events/playback_raw/")
def playback_raw() -> None:
    # Producer: writes hourly Parquet partitions to S3.
    # Airflow records the asset event on success.
    ...


@dag(schedule=[playback_raw], start_date=datetime(2026, 1, 1), catchup=False)
def playback_features():

    @task
    def build_features() -> None:
        # Consumer: runs whenever playback_raw emits a new asset event.
        ...

    build_features()


playback_features()

The 3.3 release added partitioned assets: a single upstream asset event can fan out to multiple downstream runs keyed by partition, which is what you actually want for hourly or per-region pipelines. The Airflow 3.3.1 release notes also introduce the PartitionedAtRuntime timetable for cases where the partition key is only known at run start. If you're coming from Dagster you'll find the model familiar; if you're coming from Airflow 2.x, the mental shift is that scheduling is no longer "cron plus dataset trigger." It's "cron, asset events, or both," and asset lineage shows up as a first-class panel in the new UI.

DAG versioning in production

Every time I've been paged about a "task that no longer exists" in a DAG run, the root cause was Airflow 2.x re-serializing the DAG under the scheduler's feet and then trying to render an historical run against modern code. Airflow 3 fixes that with native DAG versioning: the serialized DAG for each run is stored, tagged with a bundle version, and rendered from that snapshot in the UI. You can inspect any prior version from the version panel, and reruns pin to the original code by default.

The behavior is governed by rerun_with_latest_version, which resolves in this precedence: explicit request/CLI flag, then DAG-level setting, then [core] config, then default (False for clear/rerun, True for backfills). In practice I set it explicitly per DAG: reconciliation and ML training DAGs pin to the original version so audits reproduce; ad-hoc dashboards and internal tools use latest. Bundle versioning also means you can deploy a bad DAG revision, and it will not touch in-flight or historical runs until you retrigger them.

Migrating from Airflow 2.x to 3.x

My migration playbook after a year of moving 900 DAGs. Bump to Airflow 2.10.x first and clean up deprecations; the 2.10 deprecation warnings are the exact set that hard-fail in 3.0. Replace every from airflow.decorators import … with from airflow.sdk import …. A single ripgrep-and-sed pass covered 80% of our files. Retire subdags (removed in 3.0) in favor of task groups; they were already deprecated but plenty of 2016-era code still had them. Move schedule_interval to schedule everywhere, because schedule_interval is gone. If you were using SLAs, note that they are removed in 3.0 in favor of asset-freshness policies. This one bit us hard, and we had to build a small SLA-shim decorator during the transition.

Datasets are now Assets, and the import moves from airflow.datasets to airflow.sdk.definitions.assets. The compat shim keeps most code working, but the wire format for asset events is new, so a mixed 2.x-producer / 3.x-consumer setup will not see events until both sides move. For executors, KubernetesExecutor and CeleryExecutor are still there; the Edge Executor is the new one and is the only sensible option if you run tasks off-cluster. Finally, plan the UI change socially: your data-scientists have muscle memory for the old Flask UI, and the React UI is genuinely different. Schedule a lunch-and-learn before you cut over.

Airflow 3 vs Prefect 3 vs Dagster 1.9

I have shipped production pipelines with all three orchestrators. Here is the honest 2026 comparison. For the deeper narrative see our Airflow vs Prefect vs Dagster deep dive.

Dimension	Airflow 3.3	Prefect 3.4	Dagster 1.9
Primary abstraction	DAG + Task + Asset	Flow + Task	Asset + Op
DAG versioning	Native (AIP-65/66)	Flow deployment versions	Code locations + snapshots
Dynamic task creation	.expand() / .expand_kwargs()	task.map()	DynamicOut
Data-aware scheduling	Assets (first class in 3.x)	Automations + events	Assets (first class since 1.0)
Off-cluster execution	Edge Executor + Task SDK	Workers + work pools	Dagster+ hybrid
OSS UI	React (new in 3.0)	React	React
Best for	Existing Airflow shops; SQL-heavy ETL	Python-first teams; ad-hoc flows	Data-asset-first teams; dbt-heavy stacks

The short version: if you already run Airflow, 3.x is the right upgrade. The Task SDK and versioning fix real 2.x pain, and you keep the ecosystem of providers. If you're greenfield and asset-native, Dagster is still a cleaner fit. If you want the least ceremony for a Python-only team, Prefect wins. I wouldn't pick Airflow 3 because of assets alone; I'd pick it because your team already knows Airflow and the operational maturity is unbeaten.

Production pitfalls I keep hitting

A grab-bag of things that cost me time in the first six months on 3.x. The airflow db migrate step from 2.10 to 3.0 is not idempotent across all metadata backends, so take a snapshot of your MySQL/Postgres before you run it, and expect it to take longer than 2.x migrations because DAG versioning backfills the serialized DAG history. The 3.0 UI does not surface the classic task duration timeline the way 2.x did; the metric is still there via the REST API, but you may want to wire Prometheus separately.

Task logs live in a different place under the Task Execution API. Remote log configuration for S3/GCS is unchanged, but any custom logging handler that assumed the worker had DB access will break. Fix it by writing to stdout and letting the Task SDK forward. Finally, the @task.branch decorator with dynamic mapping still has surprising interactions: a branch that skips downstream mapped tasks can leave orphaned upstream_failed markers if you set trigger_rule incorrectly. I default to trigger_rule="none_failed_min_one_success" on reducers below a mapped branch, and I would rather write extra tasks than push complex branching into a mapped decorator.

For companion reading on the surrounding data-engineering stack, the dbt unit testing guide pairs well if your Airflow DAGs invoke dbt, and the async ETL with httpx and pandas post covers the async patterns I now embed inside @task functions rather than reinventing at the DAG level.

Frequently asked questions

Is Airflow 3.0 backward compatible with Airflow 2.x DAGs?

Most 2.x DAGs run under 3.x through a compatibility shim, but several APIs were removed outright: SubDAGs, SLAs, schedule_interval, and the legacy PythonOperator-based dataset trigger wiring. Bump to 2.10.x first, clear every deprecation warning, then upgrade. That path is officially supported and is how the Airflow team recommends the migration.

What replaced datasets in Airflow 3?

Datasets are now Assets, defined with the @asset decorator (AIP-74) and imported from airflow.sdk. Assets add lineage tracking, a dedicated UI panel, and (in 3.3) partition-aware fan-out from a single upstream event to multiple downstream runs. The old Dataset import still works via a shim but is scheduled for removal.

Do I need to rewrite my DAGs to use the Task SDK?

No, the airflow.decorators imports still work, but change them anyway. airflow.sdk is the only stable, forward-compatible authoring surface, and future features (like the Java and Go Task SDK in 3.3) only appear there. A one-line find-and-replace covers most of a large repo.

How do you limit concurrency on dynamically mapped tasks?

Set max_active_tis_per_dag on the decorator, or attach a pool with a fixed slot count. For cluster-wide caps, set the DAG-level max_active_tasks. I combine both: a pool for the resource being contended (a warehouse connection, an external API), and max_active_tis_per_dag as a per-DAG guardrail so a runaway map doesn't starve other DAGs.

What is the Edge Executor and when should I use it?

The Edge Executor (AIP-69, GA in 3.1) lets workers run anywhere they can reach the Task Execution API: outside the Kubernetes cluster, in a different region, on an edge device. Use it when you need tasks close to data that cannot leave a region, or when you want a lightweight worker (no scheduler runtime, no DB access) for security-sensitive workloads.