Apache Airflow Task Groups v Pythonu: Produkční patterny pro datové pipeline (2026)

Aktualizováno: 24. června 2026

Task Groups v Apache Airflow jsou logické kontejnery zavedené ve verzi 2.0, které seskupují související tasky do jednoho rozbalitelného uzlu v Graph View, aniž by měnily způsob plánování. Na rozdíl od starého SubDagOperator nevytvářejí samostatný DAG run a neztrácí se na nich žádný výkon (slouží čistě k vizuální a logické organizaci). V tomhle průvodci ukážu, jak je v roce 2026 používáme v produkčních ingestion pipeline: od TaskFlow API přes dynamic mapping až po nested groups s pooly a Assets.

• Task Group je vizuální seskupení tasků uvnitř DAGu; nemění semantiku schedulingu ani neběží na vlastním workerovi.
• Dekorátor @task_group je jediný způsob, jak dynamicky mapovat skupinu tasků přes .expand(). Kontextový manager TaskGroup(...) to neumí.
• TaskFlow API (@task) automaticky řeší XComy a závislosti mezi funkcemi, takže kód vypadá jako běžný Python.
• Nested task groups používáme pro pipeline typu extract, validate, transform, load, kde každá fáze má interní substrukturu.
• Pooly omezují souběžný počet instancí tasků a jsou nezbytné, pokud taskem voláte sdílený zdroj (Postgres, REST API, S3 throttling).
• V Airflow 3.x nahrazují Datasety jednotné Assets, které umožňují data-driven scheduling napříč DAGy bez cronu.

Co jsou Task Groups a kdy je použít

Task Group v Apache Airflow je čistě vizuálně-logická konstrukce: tasky uvnitř ní žijí na stejném DAGu, mají stejný scheduler context a používají standardní operátory. Skupinu pozná Graph View i Grid View v UI a vykreslí ji jako rozbalitelný uzel s prefixem v task_id (např. extract.read_postgres). Tím se DAG s padesáti tasky zúží na osm logických bloků a code review jednoho task groupu už není detektivka. V Airflow 2.x i 3.x jsou Task Groups doporučenou alternativou ke starému SubDagOperator, který se v 3.0 finálně označil za deprecated.

Task Groups používáme vždy, když pipeline obsahuje opakující se logické fáze nebo paralelní větve, které spolu sémanticky souvisí. Typické případy z mé praxe: ingestion z N zdrojů (jeden task group na zdroj), tří- až čtyřfázový ETL (extract, validate, transform, load), nebo fan-out na partitioned tabulku, kde se každá partice zpracovává stejnou sekvencí kroků. Naopak je nepoužíváme, pokud byste skupinu vytvořili jen kvůli jednomu jedinému tasku. To je akorát šum navíc v task_id a v logu.

TaskFlow API jako základ moderních DAGů

TaskFlow API zavedené v Airflow 2.0 a dále rozšířené v 3.x odstraňuje boilerplate kolem PythonOperator a manuálních XComů. Dekorátor @task obalí libovolnou Python funkci, návratová hodnota se automaticky uloží jako XCom a předání do dalšího tasku probíhá jako volání běžné funkce. Tohle je dnes defaultní způsob psaní DAGů a všechny příklady v článku ho používají. Detailní popis najdete v oficiálním TaskFlow tutoriálu Airflow.

from datetime import datetime
from airflow.sdk import dag, task
from airflow.utils.task_group import TaskGroup

@dag(
    dag_id="playback_ingestion",
    schedule="@hourly",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 3},
    tags=["ingestion", "playback"],
)
def playback_ingestion():

    @task
    def fetch_partition(hour: str) -> str:
        # vrací cestu k surovému parquet souboru v S3
        return f"s3://raw/playback/{hour}.parquet"

    @task
    def validate(path: str) -> str:
        # pydantic / Great Expectations kontroly
        return path

    @task
    def load_to_warehouse(path: str) -> int:
        # vrací počet vložených řádků
        return 12_345

    with TaskGroup("hourly_etl") as etl:
        raw = fetch_partition("{{ data_interval_start | ts_nodash }}")
        clean = validate(raw)
        loaded = load_to_warehouse(clean)

    etl

playback_ingestion()

Všimněte si, že závislost fetch → validate → load Airflow odvodí automaticky z toho, jak si funkce předávají návratové hodnoty. Žádné set_upstream, žádné >> uvnitř groupu. To je největší produkční výhoda TaskFlow API: refaktoring tasků nepřepisuje graf závislostí.

Task Group vs SubDAG: proč už SubDAGy nepíšeme

Pokud potkáte v legacy kódu SubDagOperator, je to známka, že DAG vznikl před Airflow 2.0. SubDAGy vytvářely samostatný DAG run pro každý uzel, což přinášelo tři reálné problémy: vlastní scheduler latency (často 30+ sekund overhead na uzel), nutnost ladit konkurenci přes pool i u parent DAGu, a UI, ve kterém se uživatel proklikával hluboko, aby viděl běžné tasky. Task Groups žádný z těchto problémů nemají. Srovnání najdete v tabulce níže.

Vlastnost	SubDAG (legacy)	Task Group (2.x / 3.x)
Samostatný DAG run	Ano	Ne
Scheduler overhead	Vysoký (vlastní run loop)	Nulový
Sdílí default_args	Ne (vlastní DAG)	Ano
Dynamic mapping	Nepodporováno	Ano přes @task_group
Vizualizace v UI	Samostatná obrazovka	Rozbalitelný uzel v Graph View
Doporučení v 2026	Deprecated, nepoužívat	Standard pro logické bloky

Produkční ETL pattern s Task Groups

Pro klasický ETL používáme třífázový pattern: tři task groupy (extract, transform, load) navázané za sebou bitshift operátory, a uvnitř každé groupy paralelní fan-out po zdrojích nebo partičních klíčích. Tahle struktura se osvědčuje, protože review jasně odpovídá obchodním fázím, retry se dá nastavit per-group přes default_args v dekorátoru, a alert "transform_group selhal" přesně řekne, ve které části pipeline máte hořet.

from airflow.sdk import dag, task
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta

SOURCES = ["postgres_orders", "stripe_payments", "segment_events"]

@dag(
    dag_id="daily_revenue_etl",
    schedule="0 3 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "execution_timeout": timedelta(minutes=30),
    },
    max_active_runs=1,
    tags=["etl", "revenue"],
)
def daily_revenue_etl():

    @task
    def extract_source(name: str) -> str:
        return f"s3://staging/{name}/{{{{ ds }}}}.parquet"

    @task
    def transform(path: str, source: str) -> str:
        return f"s3://curated/{source}/{{{{ ds }}}}.parquet"

    @task
    def load(path: str) -> None:
        ...

    @task
    def reconcile(_: list) -> None:
        # smírná kontrola: row counts vs zdroje
        ...

    with TaskGroup("extract") as extract_group:
        raw = {s: extract_source.override(task_id=f"extract_{s}")(s) for s in SOURCES}

    with TaskGroup("transform") as transform_group:
        curated = {
            s: transform.override(task_id=f"transform_{s}")(raw[s], s)
            for s in SOURCES
        }

    with TaskGroup("load") as load_group:
        loaded = [load.override(task_id=f"load_{s}")(curated[s]) for s in SOURCES]

    extract_group >> transform_group >> load_group >> reconcile(loaded)

daily_revenue_etl()

Pokud chcete pipeline rozšířit o validaci dat před načtením, podívejte se na náš průvodce Pydantic v2 pro validaci datových pipeline a na testování pipeline pomocí pytest a Great Expectations. Obojí navazuje na pattern výše a doplňuje ho o kontrolní vrstvu.

Dynamic task mapping přes @task_group

Když se počet zdrojů nebo partičních klíčů zjišťuje až za běhu (z databáze, REST API, S3 listingu), nelze je hardcodovat do DAG souboru. Pro tenhle případ existuje dynamic task mapping popsaný v oficiální dokumentaci Dynamic Task Mapping. Důležitý detail, na který se snadno narazí: dynamicky lze mapovat jen task groupu definovanou dekorátorem @task_group, ne contextovým managerem with TaskGroup(...). Tenhle bug jsem si v jednom projektu vyrobil sám a dvacet minut hledal, proč mi .expand() hází AttributeError.

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

@dag(
    dag_id="multi_tenant_ingestion",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
)
def multi_tenant_ingestion():

    @task
    def list_tenants() -> list[str]:
        # za běhu načteme aktuální seznam tenantů
        return ["acme", "globex", "initech", "umbrella"]

    @task_group(group_id="per_tenant")
    def process_tenant(tenant: str):

        @task
        def extract(t: str) -> str:
            return f"s3://raw/{t}/data.json"

        @task
        def load(path: str, t: str) -> int:
            return 42

        raw = extract(tenant)
        load(raw, tenant)

    process_tenant.expand(tenant=list_tenants())

multi_tenant_ingestion()

Při běhu scheduler vytvoří jednu instanci task groupu na každého tenantta a v Graph View je uvidíte jako per_tenant[0], per_tenant[1], ... Mapping respektuje max_map_length (default 1024). Pokud čekáte víc, zvedněte ho v airflow.cfg, případně pipeline rozdělte. Pro map-reduce výstupy slouží .zip() a běžný @task, který přijme list návratových hodnot z mapped groupu.

Nested Task Groups pro multi-zdrojové pipeline

Task Groups lze vnořovat libovolně hluboko, ale v praxi se mi osvědčily maximálně dvě úrovně, jinak Graph View přestane být přehledný. Typický use case je "source-of-truth" pipeline, kde každý zdroj má svůj vlastní extract+validate sub-flow a všechny zdroje se pak sbíhají do společného load groupu.

with TaskGroup("ingestion") as ingestion:
    for source in ["orders", "shipments", "returns"]:
        with TaskGroup(group_id=source) as source_tg:
            r = extract_source.override(task_id="extract")(source)
            v = validate.override(task_id="validate")(r)
            v  # validate je terminál sub-groupu

with TaskGroup("warehouse") as warehouse:
    merged = merge_streams()
    load(merged)

ingestion >> warehouse

Výhoda je, že každý zdroj má v UI vlastní rozbalitelný blok pod ingestion.orders, ingestion.shipments atd., a kliknutím na nadřazený uzel ingestion dostanete jeden agregovaný stav. To se ohromně hodí pro oncall, který nepotřebuje vidět padesát zelených checkboxů. Chce vidět "ingestion: zelená, warehouse: červená" a jít dál.

Pooly, max_active_tis_per_dag a kontrola paralelismu

Task Group sama o sobě neřídí paralelismus. Všechny tasky uvnitř běží podle globálního scheduleru a konfigurace executoru. Pokud chcete například omezit počet souběžných dotazů do Postgresu na 4, založte si v UI (Admin → Pools) pool s názvem postgres_pool a slot count 4, a v tasku ho deklarujte přes pool="postgres_pool". To je čistší než max_active_tis_per_dag, protože pool sdílí kapacitu napříč všemi DAGy, které sahají na stejný zdroj.

@task(pool="postgres_pool", pool_slots=1, retries=2)
def extract_partition(p: str) -> str:
    ...

@task(pool="s3_write_pool", pool_slots=1, queue="io_heavy")
def upload_parquet(df_path: str) -> str:
    ...

Druhá užitečná páka je max_active_tasks na úrovni DAGu (kolik tasků z tohoto DAGu může běžet souběžně) a max_active_runs (kolik souběžných DAG runs Airflow povolí). V jedné ingestion pipeline, kterou jsem ladil, jsme měli max_active_runs=1 kvůli inkrementálnímu cursor pointeru. Bez toho by dvě souběžné instance natáhly stejnou hodinu dvakrát a duplicitu pak řešíme dva dny po incidentu. Pokud chcete přesný popis poolů a fair-scheduling triků, projděte průvodce task groups od Astronomeru.

Předávání dat: XCom, externí storage a Assets

XCom je metadatový mechanismus. Návratová hodnota tasku se serializuje (JSON v defaultu, picklem pokud zapnete) a uloží do metastoru. Pro malá data (cesty, ID, status flagy, řádky < 10 KB) je to ideální. Pro DataFrame, JSON s tisícem objektů nebo binární payload XCom nepoužívejte. Zaplníte metabázi a každý další DAG parse se zpomalí, protože scheduler načítá XCom historii.

• Malé hodnoty: vraťte z @task primitivy nebo malé dict. TaskFlow API zařídí XCom transparentně.
• Větší payloady: uložte do S3/GCS/local a předávejte pouze cestu. Tým, který si řekne "předáme si tabulku přes XCom", vždycky končí incidentem.
• Cross-DAG dependence: v Airflow 3.x použijte Assets (přejmenované Datasety). Producent DAGu zapíše do assetu, konzument se "přihlásí k odběru" přes schedule=[asset] a spustí se, jakmile producent doběhne.

from airflow.sdk import Asset, dag, task

curated_revenue = Asset("s3://curated/revenue/")

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def produce_revenue():
    @task(outlets=[curated_revenue])
    def write():
        ...
    write()

@dag(schedule=[curated_revenue], start_date=datetime(2026, 1, 1), catchup=False)
def consume_revenue():
    @task
    def read():
        ...
    read()

Pro typovou bezpečnost payloadu mezi tasky doporučuji obalit XCom hodnoty Pydantic modely. Popisuji to v článku Pydantic v2 pro validaci dat v Pythonu. Validace u hranice tasku stojí mikrosekundy a ušetří hodiny ladění data corruption.

Testování DAGů s Task Groups

Testujte tři vrstvy zvlášť: (1) parsování DAGu jednoduchým importem v pytest (žádné syntax chyby, žádné cyklické závislosti), (2) jednotkové testy callable funkcí uvnitř @task (volat je přímo bez kontextu, protože TaskFlow funkce jsou stále běžné Python funkce) a (3) integrační testy s Airflow standalone, který spustí DAG v lokálním LocalExecutor a ověří, že tasky dosáhnou stavu success.

# tests/test_revenue_dag.py
from airflow.models import DagBag

def test_dag_parses_without_errors():
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    assert dag_bag.import_errors == {}
    assert "daily_revenue_etl" in dag_bag.dag_ids

def test_task_group_structure():
    dag = DagBag(dag_folder="dags/", include_examples=False).get_dag("daily_revenue_etl")
    group_ids = {tg.group_id for tg in dag.task_group.children.values()
                 if hasattr(tg, "group_id") and tg.group_id}
    assert {"extract", "transform", "load"}.issubset(group_ids)

Časté chyby a jak jim předejít

1. Kolize task_id v různých groupách. Když máte @task definovaný uvnitř funkce a zavoláte ho v dvou groupách, Airflow se snaží oba registrovat jako stejný task_id. Řešení: .override(task_id="...") na každém volání.
2. Předávání velkých DataFrame přes XCom. Symptom: scheduler se zpomalí, metabáze nabobtná. Řešení: psát na S3 a předávat cestu.
3. Použití TaskGroup(...) contextu místo @task_group dekorátoru pro dynamic mapping. .expand() je dostupné jen na dekorátoru, context manager to nepodporuje.
4. Vnoření groupů více než 2× hluboko. Graph View se stává nečitelným a debug logy mají task_id jako etl.ingestion.orders.validate.row_count_check. To nikdo nečte.
5. Chybějící retries v default_args. V produkci síťové chyby (S3 throttling, DB connection reset) jsou normální. retries=2 + retry_delay=timedelta(minutes=5) je rozumný default.
6. Nezohlednění idempotence. Task v Airflow MUSÍ být idempotentní. Pokud běží podruhé (retry, backfill), nesmí způsobit duplicitu. INSERT bez UPSERT je klasický footgun.
7. Předpoklad, že Task Group sama řídí paralelismus. Nepravda. Použijte pool, max_active_tis_per_dag nebo max_active_tasks.

Pro hlubší rozšíření pipeline o transformační vrstvu se podívejte na náš průvodce ETL pipeline s Pandas a SQLAlchemy, který krásně doplňuje produkční patterny popsané výše.

Často kladené otázky

Jaký je rozdíl mezi Task Group a SubDAG v Airflow?

Task Group je čistě vizuální seskupení tasků uvnitř stejného DAGu bez vlastního DAG runu, zatímco SubDAG vytvářel samostatný DAG run s vlastní scheduler smyčkou. SubDAG je v Airflow 3.x oficiálně deprecated kvůli režii a problémům s konkurencí, takže pro veškerou novou logiku používejte Task Groups.

Lze v Airflow Task Group dynamicky mapovat?

Ano, ale jen pokud Task Group definujete přes dekorátor @task_group (ne přes context manager with TaskGroup(...)). Volání .expand(arg=task_returning_list()) vytvoří jednu instanci task groupu na každý prvek seznamu vrácený z předchozího tasku za běhu.

Jak předávat data mezi tasky v Task Group?

Pro malé hodnoty (cesty, ID, status flagy) použijte návratovou hodnotu z @task funkce. TaskFlow API ji automaticky uloží jako XCom. Pro větší payloady (DataFrame, velké JSON) ukládejte data do S3/GCS a přes XCom předávejte pouze cestu; XCom by neměl obsahovat víc než desítky kilobajtů.

Můžu vnořovat Task Groups?

Ano, vnořování je plně podporované a v UI se nested groups vykreslují jako rozbalitelné podstromy. V praxi doporučuji nejdéle dvě úrovně zanoření, hlouběji se Graph View stává nepřehledným a task_id v lozích se prodlouží natolik, že je nikdo nečte.

Jak omezit počet souběžných tasků uvnitř Task Group?

Task Group samotná paralelismus neřídí. Pro omezení použijte Airflow Pool (v UI Admin → Pools) a v tasku ho deklarujte přes @task(pool="moje_pool"), případně nastavte max_active_tasks na úrovni celého DAGu. Pool je preferovaný, protože sdílí kapacitu napříč všemi DAGy, které sahají na stejný sdílený zdroj.