Apache Airflow w 2026: orkiestracja potoków danych, DAG-i i pytania rekrutacyjne

Apache Airflow 3.2 to najbardziej znacząca ewolucja platformy orkiestracji open-source od czasu gruntownej przebudowy architektonicznej w wersji 3.0. Przy ponad 10 milionach pobrań miesięcznie na PyPI i wdrożeniach w firmach od Airbnb po Spotify, Airflow utrzymuje pozycję dominującego narzędzia do budowania, harmonogramowania i monitorowania potoków danych. Niniejszy artykuł przedstawia tworzenie DAG-ów z wykorzystaniem nowego Task SDK, wzorce orkiestracji potoków danych oraz pytania, które regularnie pojawiają się na rozmowach kwalifikacyjnych na stanowiska data engineering w 2026 roku.

Tworzenie DAG-ów z Airflow Task SDK

Airflow 3.0 wprowadził Task SDK — samodzielny pakiet oddzielający definicje DAG-ów od wewnętrznych mechanizmów Airflow. Głównym założeniem jest tworzenie przenośnych, stabilnych wersjonowo DAG-ów, które przetrwają aktualizacje Airflow bez konieczności modyfikacji kodu. Wszystkie kluczowe obiekty — DAG, dag, task, BaseOperator, Connection, Variable — znajdują się teraz w module airflow.sdk.

Stare ścieżki importów (airflow.decorators.task, airflow.models.dag.DAG) nadal działają w wersji 3.2, ale zostały oznaczone jako przestarzałe i zostaną usunięte w przyszłych wydaniach.

# etl_daily_revenue.py
import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["finance", "etl"],
)
def etl_daily_revenue():
    @task()
    def extract_transactions() -> list[dict]:
        # Pulls raw transactions from the payments API
        import requests
        response = requests.get("https://api.internal/payments/daily")
        return response.json()["transactions"]

    @task(multiple_outputs=True)
    def transform(transactions: list[dict]) -> dict:
        # Aggregates by currency and computes totals
        totals = {}
        for tx in transactions:
            currency = tx["currency"]
            totals[currency] = totals.get(currency, 0) + tx["amount"]
        return {"totals": totals, "count": len(transactions)}

    @task()
    def load(totals: dict, count: int):
        # Inserts aggregated row into the warehouse
        from warehouse import insert_revenue_summary
        insert_revenue_summary(totals=totals, transaction_count=count)

    raw = extract_transactions()
    result = transform(raw)
    load(result["totals"], result["count"])

etl_daily_revenue()

Dekorator @dag zastępuje tradycyjny menedżer kontekstu with DAG(...). Z kolei @task przekształca zwykłe funkcje Pythona w zadania Airflow, a serializacja XCom odbywa się automatycznie. Wywołanie transform(raw) deklaruje zależność między zadaniami — Airflow buduje krawędzie grafu DAG na podstawie grafu wywołań funkcji.

Dynamiczne mapowanie zadań dla zmiennych obciążeń

Statyczne DAG-i przestają się sprawdzać, gdy liczba elementów do przetworzenia zmienia się między kolejnymi uruchomieniami. Dynamiczne mapowanie zadań, stabilne od Airflow 2.4 i udoskonalone w serii 3.x, rozwiązuje ten problem poprzez rozszerzanie zadań w czasie wykonania za pomocą metody .expand().

# etl_multi_region.py
import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@hourly",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["regional", "etl"],
)
def etl_multi_region():
    @task()
    def get_regions() -> list[str]:
        # Returns the active regions from config
        return ["us-east-1", "eu-west-1", "ap-southeast-1"]

    @task()
    def process_region(region: str) -> dict:
        # Processes events for a single region
        from data_processor import aggregate_events
        return aggregate_events(region)

    @task()
    def merge_results(results: list[dict]):
        # Combines all regional outputs into a single report
        from reporter import publish_global_report
        publish_global_report(results)

    regions = get_regions()
    # .expand() creates one mapped task instance per region
    processed = process_region.expand(region=regions)
    merge_results(processed)

etl_multi_region()

W momencie wykonania Airflow tworzy trzy równoległe instancje process_region — po jednej na każdy region. Jeśli następnego dnia pojawi się czwarty region, kod DAG-a nie wymaga żadnych zmian. Zadanie merge_results oczekuje na zakończenie wszystkich zmapowanych instancji przed rozpoczęciem własnego wykonania.

Partycje assetów: harmonogramowanie oparte na danych w Airflow 3.2

Przed wersją 3.2 harmonogramowanie oparte na danych działało na poziomie całego assetu: gdy DAG producenta aktualizował asset, wszystkie DAG-i konsumentów uruchamiały się niezależnie od tego, który wycinek danych uległ zmianie. Partycje assetów rozwiązują ten problem, wprowadzając granularność na poziomie poszczególnych partycji.

Warto rozważyć scenariusz, w którym trzy upstreamowe DAG-i generują cogodzinne statystyki zawodników dla różnych lig sportowych. Downstreamowy DAG analityczny powinien uruchamiać się dopiero wtedy, gdy wszystkie trzy ligi opublikowały dane za tę samą godzinę.

# downstream_analytics.py
import pendulum
from airflow.sdk import dag, task
from airflow.timetables.assets import CronPartitionTimetable

# Define partitioned assets with hourly granularity
nba_stats = Asset("s3://datalake/nba/hourly/", partitions=CronPartitionTimetable("0 * * * *"))
epl_stats = Asset("s3://datalake/epl/hourly/", partitions=CronPartitionTimetable("0 * * * *"))
nfl_stats = Asset("s3://datalake/nfl/hourly/", partitions=CronPartitionTimetable("0 * * * *"))

@dag(
    # Triggers only when all three assets have matching partition
    schedule=(nba_stats & epl_stats & nfl_stats),
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def unified_sports_analytics():
    @task()
    def aggregate_all_leagues(**context):
        partition = context["partition"]
        # Process only the specific hourly slice across all leagues
        from analytics import build_cross_league_report
        build_cross_league_report(partition=partition)

    aggregate_all_leagues()

unified_sports_analytics()

Harmonogramowanie oparte na partycjach eliminuje nadmiarowe uruchomienia potoków. DAG downstream uruchamia się wyłącznie wtedy, gdy dokładnie ta partycja, której potrzebuje, jest dostępna we wszystkich źródłach upstream — nie wcześniej i nie na niepełnych danych.

Natywne zadania asynchroniczne dla obciążeń I/O

Airflow 3.2 dodaje natywną obsługę asynchroniczną w PythonOperator. Wcześniej uruchamianie współbieżnych operacji wejścia/wyjścia (tysiące wywołań API, masowe pobieranie plików) wymagało tworzenia własnych operaturów deferrowalnych. W wersji 3.2 wystarczy przekazać funkcję async bezpośrednio do dekoratora @task.

# async_file_download.py
import asyncio
import aiohttp
from airflow.sdk import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["async", "download"],
)
def async_file_download():
    @task()
    async def download_files():
        urls = [f"https://data.source/files/{i}.csv" for i in range(500)]
        async with aiohttp.ClientSession() as session:
            # Downloads 500 files concurrently instead of sequentially
            tasks = [fetch_and_save(session, url) for url in urls]
            results = await asyncio.gather(*tasks)
        return {"downloaded": len(results)}

    download_files()

async def fetch_and_save(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as resp:
        content = await resp.read()
        filename = url.split("/")[-1]
        with open(f"/data/downloads/{filename}", "wb") as f:
            f.write(content)
        return filename

async_file_download()

Podejście asynchroniczne umożliwia jednoczesne pobieranie 500 plików w ramach pojedynczego slotu workera, w przeciwieństwie do 500 sekwencyjnych wywołań HTTP w wersji synchronicznej. W przypadku obciążeń zdominowanych przez operacje I/O przekłada się to na przyspieszenie o rząd wielkości bez konieczności uruchamiania dodatkowych workerów.

Architektura Airflow: komponenty i przepływ wykonania

Zrozumienie architektury Airflow jest niezbędne zarówno do jego obsługi w środowisku produkcyjnym, jak i do odpowiadania na pytania rekrutacyjne. Pięć komponentów współdziała podczas każdego uruchomienia DAG-a:

• Scheduler — parsuje pliki DAG, rozwiązuje zależności i kolejkuje zadania do wykonania. W Airflow 3.x scheduler działa bezstanowo względem bazy danych metadanych.
• Executor — określa, gdzie zadania są uruchamiane. LocalExecutor obsługuje konfiguracje jednowęzłowe. CeleryExecutor dystrybuuje zadania pomiędzy węzły robocze. KubernetesExecutor uruchamia osobny pod na każde zadanie.
• Workers — wykonują właściwy kod zadań. Dzięki nowemu Task Execution API w Airflow 3.0, workery komunikują się poprzez stabilny kontrakt, co umożliwia wykonywanie zadań w kontenerach, środowiskach brzegowych lub zewnętrznych środowiskach uruchomieniowych.
• Baza danych metadanych — PostgreSQL (zalecany) lub MySQL. Przechowuje definicje DAG-ów, stany zadań, wartości XCom, dane uwierzytelniające połączeń oraz logi audytu.
• Serwer webowy — interfejs użytkownika Airflow do monitorowania uruchomień DAG-ów, przeglądania logów, ręcznego wyzwalania uruchomień i zarządzania połączeniami.

Airflow vs Prefect vs Dagster w 2026 roku

Airflow konkuruje z dwoma znaczącymi alternatywami. Właściwy wybór zależy od wielkości zespołu, istniejącej infrastruktury oraz typów budowanych potoków danych.

| Funkcjonalność | Airflow 3.2 | Prefect 3.x | Dagster 1.9 | |---|---|---|---| | Definicja DAG | Dekoratory Python (airflow.sdk) | Dekoratory Python (@flow, @task) | Dekoratory Python (@asset, @op) | | Harmonogramowanie | Cron, asset-aware, partition-aware | Cron, event-driven | Cron, sensor-based, asset-aware | | Model wykonania | Scentralizowany scheduler + rozproszone workery | Hybrydowy (serwer + pule zadań) | Scentralizowany dagster-daemon | | Zadania dynamiczne | .expand() zmapowane zadania | Natywne pętle Python | Dynamiczne partycje | | Obsługa async | Natywna w 3.2 | Natywna od 2.0 | Async I/O ops | | Izolacja wielozespołowa | Wbudowana (3.2 eksperymentalna) | Oparta na workspace (Cloud) | Branch deployments | | Wielkość społeczności | Największa (35k+ gwiazd GitHub) | Rosnąca (18k+ gwiazd) | Rosnąca (12k+ gwiazd) | | Najlepsze zastosowanie | Złożone potoki wielozespołowe z ustaloną infrastrukturą | Mniejsze zespoły stawiające na szybką iterację | Organizacje zorientowane na assety danych |

Przewaga Airflow tkwi w jego ekosystemie: ponad 80 pakietów providerów obejmujących każdą ważną usługę chmurową, bazę danych i API. Prefect wypróżnia drogę w obszarze doświadczenia deweloperskiego, oferując mniejszą ilość kodu szablonowego. Model Dagstera, skoncentrowany na assetach danych, sprawdza się najlepiej w zespołach, które myślą w kategoriach produktów danych, a nie sekwencji zadań.

Pytania rekrutacyjne: Apache Airflow dla inżynierów danych

Poniższe pytania odzwierciedlają to, czego kierownicy rekrutacji i starsi inżynierowie faktycznie oczekują podczas rozmów kwalifikacyjnych na stanowiska data engineering w firmach wykorzystujących Airflow w środowiskach produkcyjnych.

Czym jest DAG i w jaki sposób Airflow go wykorzystuje?

DAG (Directed Acyclic Graph, skierowany graf acykliczny) definiuje przepływ pracy jako zbiór zadań z zależnościami, z gwarancją braku odwołań cyklicznych. Airflow parsuje pliki Pythona z katalogu dags/, buduje graf zależności, a scheduler określa kolejność wykonania. Każde uruchomienie DAG-a tworzy obiekt DagRun powiązany z datą logiczną. Ograniczenie acykliczności zapewnia, że scheduler zawsze jest w stanie znaleźć poprawną kolejność wykonania — zadanie A uruchamia się przed zadaniem B, nigdy odwrotnie w tym samym czasie.

Jak działa XCom i kiedy należy go unikać?

XCom (cross-communication) służy do przekazywania niewielkich porcji danych między zadaniami. Zadanie wysyła wartość zwracaną do XCom, a zadania downstream pobierają ją. Task SDK obsługuje ten mechanizm automatycznie przy przekazywaniu wartości zwracanych między dekorowanymi zadaniami. XCom domyślnie przechowuje dane w bazie danych metadanych, co czyni go nieodpowiednim do dużych zbiorów danych (ponad kilka KB). Do transferu dużych wolumenów danych należy korzystać z zewnętrznego storage (S3, GCS) i przekazywać przez XCom jedynie ścieżkę referencji.

Jaka jest różnica między parametrami schedule, start_date i catchup?

Parametr schedule (przemianowany z schedule_interval w Airflow 3.x) definiuje częstotliwość uruchomień DAG-a: wyrażenie cron, obiekt timedelta, obiekt timetable lub wyzwalacz oparty na assetach. start_date ustala najwcześniejszą datę logiczną, dla której może zostać utworzone uruchomienie DAG-a. Ustawienie catchup=True (domyślne) tworzy uruchomienia DAG-a dla wszystkich pominiętych interwałów między start_date a bieżącym momentem. Z kolei catchup=False instruuje scheduler, aby pominął przeszłe interwały i harmonogramował dopiero od bieżącego czasu. Częsty wzorzec produkcyjny to catchup=False dla operacyjnych DAG-ów i catchup=True dla historycznych backfilli.

Czym różni się KubernetesExecutor od CeleryExecutor?

CeleryExecutor utrzymuje pulę długotrwałych procesów workerowych połączonych za pośrednictwem brokera wiadomości (Redis lub RabbitMQ). Zadania są kolejkowane i wykonywane na dostępnych workerach. KubernetesExecutor tworzy nowy pod Kubernetes dla każdego zadania, wykorzystując obraz Docker i wymagania zasobowe określone dla danego zadania. Celery oferuje niższą latencję (brak narzutu na uruchamianie podów) i dobrze sprawdza się dla jednorodnych obciążeń. KubernetesExecutor zapewnia silniejszą izolację, kontrolę zasobów na poziomie pojedynczego zadania i eliminuje konieczność zarządzania statyczną pulą workerów — co stanowi optymalne rozwiązanie dla heterogenicznych obciążeń, gdzie poszczególne zadania mają różne wymagania dotyczące zależności i pamięci.

Jakie strategie obsługi błędów zadań są dostępne w Airflow?

Airflow oferuje szereg mechanizmów obsługi błędów. Parametry retries i retry_delay konfigurują automatyczne ponowne próby z wykładniczym wycofaniem (backoff). on_failure_callback wyzwala niestandardową logikę (powiadomienia Slack, incydenty PagerDuty) w momencie awarii zadania. trigger_rule kontroluje, jak reagują zadania downstream — all_success (domyślnie), one_success, all_failed lub none_failed_min_one_success. Dla przejściowych problemów infrastrukturalnych parametr retry_exponential_backoff=True wydłuża czas oczekiwania między kolejnymi próbami. Mechanizm SLA (przemianowany na Deadline Alerts w 3.x) monitoruje czas wykonania i wyzwala callbacki, gdy zadania przekraczają oczekiwany czas działania.

Najlepsze praktyki produkcyjne dla wdrożeń Airflow

Niezawodne działanie Airflow na dużą skalę wymaga uwzględnienia wzorców operacyjnych, które wykraczają daleko poza samo poprawne pisanie DAG-ów.

Idempotentne zadania. Każde zadanie powinno generować ten sam wynik przy wielokrotnym wykonaniu z tymi samymi danymi wejściowymi. Zamiast zwykłego INSERT należy stosować INSERT ... ON CONFLICT lub MERGE. Dane wyjściowe powinny być partycjonowane według daty logicznej. Takie podejście gwarantuje bezpieczne ponowne próby i backfille bez duplikacji danych.

Małe, skoncentrowane DAG-i. Warto opierać się pokusie budowania monolitycznych DAG-ów z dziesiątkami zadań. Złożone potoki powinny być dzielone na wiele DAG-ów połączonych poprzez assety (wcześniej datasety). Mniejsze DAG-i są szybciej parsowane, łatwiejsze w debugowaniu i umożliwiają częściowe restarty potoków.

Zarządzanie połączeniami. Wszystkie dane uwierzytelniające powinny być przechowywane w menedżerze połączeń Airflow lub zewnętrznym backendzie sekretów (AWS Secrets Manager, HashiCorp Vault). Dane uwierzytelniające nie powinny nigdy być kodowane bezpośrednio w plikach DAG. Airflow 3.x szyfruje pola połączeń w spoczynku w bazie danych metadanych.

Monitoring i alarmowanie. Metryki Airflow należy eksportować do Prometheus lub StatsD. Kluczowe metryki do śledzenia to scheduler_heartbeat, dag_processing.total_parse_time oraz executor.queued_tasks. Airflow 3.2 dodaje śledzenie OpenTelemetry (traces) zapewniające kompleksową obserwowalnosc potoków od początku do końca.

Przygotowanie do rozmowy kwalifikacyjnej z zakresu data engineering obejmującej orkiestrację potoków danych? SharpSkill oferuje dedykowane pytania rekrutacyjne z Airflow oraz moduły ćwiczeniowe odwzorowujące rzeczywiste scenariusze ze środowisk produkcyjnych.

Podsumowanie

• Airflow 3.2 wprowadza Task SDK (airflow.sdk) jako stabilne API do tworzenia DAG-ów — migracja importów już teraz pozwoli uniknąć zmian łamiących kompatybilność w przyszłych wydaniach
• Partycje assetów umożliwiają harmonogramowanie oparte na danych z granularnością na poziomie partycji, eliminując nadmiarowe uruchomienia potoków przy częściowych aktualizacjach danych
• Natywna obsługa asynchroniczna w PythonOperator obsługuje obciążenia związane z I/O (masowe pobieranie plików, równoległe wywołania API) bez konieczności tworzenia własnych operatorów deferrowalnych
• Dynamiczne mapowanie zadań za pomocą .expand() dostosowuje się do zmiennych rozmiarów obciążeń w czasie wykonania bez modyfikacji kodu DAG-a
• Przygotowanie do rozmów kwalifikacyjnych powinno obejmować mechanikę DAG-ów, kompromisy między executorami (Celery vs Kubernetes), ograniczenia XCom oraz wzorce idempotentności
• Wdrożenia produkcyjne zyskują na stosowaniu małych, idempotentnych DAG-ów, zewnętrznym zarządzaniu sekretami oraz eksporcie metryk do platform obserwowalności