Apache Airflow 3, planowanie potoków danych

Apache Airflow 3, czyli harmonogram, który wie, co od czego zależy

Zadania uruchamiane cyklicznie da się obsłużyć wpisami w harmonogramie systemowym, dopóki jest ich pięć i żadne nie zależy od drugiego. Przy pięćdziesięciu zadaniach, z których jedno musi poczekać na trzy inne, a przy błędzie trzeba wiedzieć, które dane są nieaktualne, potrzebny jest inny mechanizm.

Airflow opisuje potok jako graf zależności: zadania, kolejność i warunki. Platforma pilnuje, co się wykonało, co zawiodło i co czeka, a interfejs pokazuje to w postaci, którą da się przeczytać w trakcie awarii.

Trzecia wersja, wydana w kwietniu 2025 roku, przyniosła zmiany, które zmieniają sposób pracy bardziej niż numer sugeruje.

Pierwszy przepływ

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

@dag(schedule="0 6 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def raport_dzienny():

    @task
    def pobierz_zamowienia() -> list[dict]:
        return baza.zamowienia_z_wczoraj()

    @task
    def policz_wskazniki(zamowienia: list[dict]) -> dict:
        return {"liczba": len(zamowienia), "wartosc": sum(z["kwota"] for z in zamowienia)}

    @task
    def wyslij_raport(wskazniki: dict) -> None:
        poczta.wyslij("zarzad@firma.pl", wskazniki)

    wyslij_raport(policz_wskazniki(pobierz_zamowienia()))

raport_dzienny()

Zależności wynikają z przekazywania wyników między zadaniami, więc nie deklaruje się ich osobno. Ostatnia linia opisuje cały graf: pobierz, policz, wyślij, w tej kolejności.

Przestrzeń nazw używana w imporcie to nowość trzeciej wersji. Wcześniej importowało się z różnych miejsc wewnętrznych, co przy aktualizacjach powodowało zmiany łamiące zgodność. Teraz jest jeden zestaw stabilnych interfejsów do pisania przepływów, oddzielony od wnętrza platformy.

Nadrabianie zaległości jest w trzeciej wersji domyślnie wyłączone, bo ustawienie catchup_by_default zmieniło wartość domyślną na fałsz. W przykładzie zapisano to mimo wszystko wprost, żeby intencja była widoczna w kodzie, a nie ukryta w konfiguracji instalacji. Jeśli włączysz je świadomie albo przeniesiesz konfigurację z drugiej wersji, gdzie domyślnie było włączone, uruchomienie przepływu z datą początkową sprzed roku wywoła trzysta sześćdziesiąt pięć przebiegów naraz.

Wersjonowanie przepływów

Najważniejsza zmiana trzeciej wersji i odpowiedź na problem, który wcześniej powodował najwięcej zamieszania.

W poprzednich wersjach ponowne uruchomienie starego przebiegu wykonywało go na aktualnym kodzie. Przebieg sprzed miesiąca, wznowiony po zmianie logiki, dawał inny wynik niż pierwotnie i nikt nie wiedział, który jest prawidłowy.

Teraz platforma śledzi wersję kodu, z którą przebieg został uruchomiony, i domyślnie wznawia go na tej samej. Ustawienie pozwala wybrać zachowanie, ale domyślne jest właściwe: przebieg z marca odtwarza się tak, jak wyglądał w marcu.

To zmienia sposób pracy przy poprawianiu danych historycznych, z jednym wyjątkiem, o którym łatwo zapomnieć. Nadrabianie zaległości zachowało dawne zachowanie i domyślnie biegnie na najnowszym kodzie. Rozstrzyga o tym ustawienie rerun_with_latest_version, które przyjmuje fałsz przy czyszczeniu i ponownym uruchomieniu, a prawdę przy nadrabianiu zaległości, i które da się nadpisać na poziomie pojedynczego żądania albo pojedynczego przepływu.

Interfejs pokazuje przy każdym przebiegu, na której wersji został wykonany. Przy diagnozowaniu rozbieżności w danych to pierwsza rzecz do sprawdzenia i wcześniej trzeba jej było szukać w historii repozytorium, zestawiając daty wdrożeń z datami przebiegów.

Zasoby i wyzwalanie zdarzeniami

Druga duża zmiana dotyczy tego, co uruchamia przepływ. Harmonogram czasowy jest prosty i ma jedną wadę: zadanie o siódmej zakłada, że dane z poprzedniego kroku są gotowe, a jeśli nie są, przetwarza stare.

Model oparty o zasoby odwraca tę logikę. Przepływ deklaruje, co produkuje i co konsumuje, a platforma uruchamia go wtedy, gdy potrzebne dane się pojawiły.

from airflow.sdk import asset

@asset(schedule="@daily")
def zamowienia_surowe() -> None:
    baza.zaladuj_z_zrodla()

@asset(schedule=zamowienia_surowe)
def zamowienia_oczyszczone() -> None:
    baza.przetworz()

Drugi przepływ nie ma własnej godziny. Uruchomi się, gdy pierwszy skończy i zaktualizuje swój zasób, więc nie trzeba zgadywać odstępu ani dokładać marginesu bezpieczeństwa.

Trzecia wersja rozszerzyła to o wyzwalanie zdarzeniami spoza platformy. Wiadomość w kolejce, plik w magazynie obiektowym albo zdarzenie z systemu zewnętrznego mogą uruchomić przepływ, co usuwa najczęstsze obejście, czyli odpytywanie źródła co pięć minut.

Zadania w innych językach

Trzecia wersja dołożyła warstwę pozwalającą pisać poszczególne zadania w językach innych niż Python, przy zachowaniu opisu przepływu w Pythonie.

To odpowiedź na sytuację powtarzalną w zespołach danych: logika przetwarzania istnieje już w innym języku, a jedynym powodem przepisywania jej do Pythona było ograniczenie narzędzia. Teraz zadanie może być wykonane tam, gdzie żyje, a platforma odpowiada za kolejność, ponowienia i widoczność.

Praktyczny wniosek: przy istniejącym kodzie w innym języku warto sprawdzić tę drogę, zanim zaplanujesz migrację. Przepisywanie działającego kodu wyłącznie po to, żeby zmieścił się w narzędziu, rzadko jest dobrą inwestycją. Warstwa pośrednicząca oraz zestawy narzędzi dla Javy i Go są w wersji 3.3 oznaczone jako eksperymentalne, więc traktuj je jako drogę do sprawdzenia, a nie jako fundament produkcji.

Ponowienia i obsługa błędów

Potok danych zawodzi regularnie, bo źródła bywają niedostępne, a dane niekompletne. Sposób obsługi tych sytuacji odróżnia potok produkcyjny od skryptu.

@task(retries=3, retry_delay=timedelta(minutes=5), retry_exponential_backoff=True)
def pobierz_z_api() -> list[dict]:
    return klient.pobierz_dane()

Ponowienia z narastającym odstępem obsługują błędy przejściowe, czyli niedostępność usługi i przekroczony limit zapytań. To pokrywa większość awarii, które naprawiają się same.

Osobną sprawą jest to, co zrobić, gdy ponowienia się skończą. Domyślnie zadanie zostaje oznaczone jako nieudane, a zadania zależne czekają. Przy potoku, w którym jeden nieudany krok nie powinien blokować reszty, można ustawić inną regułę wyzwalania kolejnego zadania.

Warto też odróżnić błąd od braku danych. Zadanie, które nie znalazło nic do przetworzenia, nie jest awarią i lepiej oznaczyć je jako pominięte niż jako nieudane, bo inaczej alarmy stają się szumem.

Ostatnią rzeczą jest powiadamianie. Nieudany potok bez powiadomienia zauważa się, gdy ktoś pyta o brakujący raport. Wywołanie zwrotne przy niepowodzeniu, wysyłające wiadomość na kanał zespołu, kosztuje kilka linii i skraca czas reakcji z dni do minut. Warto dołożyć do niego link do zapisu przebiegu, żeby osoba odbierająca powiadomienie od razu wiedziała, gdzie patrzeć.

Testowanie potoków

Potok jest kodem, więc podlega tym samym zasadom co reszta, a jednak testuje się go najrzadziej.

Najprostszy i najskuteczniejszy zabieg to wydzielenie logiki przetwarzania poza definicję zadania. Funkcja przyjmująca dane i zwracająca wynik da się przetestować zwykłym testem jednostkowym, bez uruchamiania czegokolwiek.

def policz_wskazniki(zamowienia: list[dict]) -> dict:
    return {"liczba": len(zamowienia), "wartosc": sum(z["kwota"] for z in zamowienia)}

@task
def zadanie_wskazniki(zamowienia: list[dict]) -> dict:
    return policz_wskazniki(zamowienia)

Zadanie staje się wtedy cienkim opakowaniem, a cała logika żyje w miejscu, które da się sprawdzić w milisekundy. To jedna zmiana, która daje najwięcej i nie wymaga żadnych narzędzi poza tymi, których zespół już używa.

Drugi poziom to sprawdzenie samych definicji przepływów: czy wszystkie da się wczytać bez błędu, czy nie ma cykli w grafie, czy identyfikatory są unikalne. Taki test w potoku budowania wychwytuje literówki, zanim trafią na produkcję.

Trzeci to uruchomienie pojedynczego zadania na danych testowych. Przydaje się przy zadaniach dotykających systemów zewnętrznych, gdzie test jednostkowy nie wystarcza, a uruchamianie całego grafu jest zbyt kosztowne.

Warto przy okazji wiedzieć, skąd ta gimnastyka się bierze. Definicja przepływu jest tutaj osobnym grafem, więc żeby uruchomić cokolwiek w teście, potrzebujesz planisty i bazy metadanych, i stąd cały zabieg z wyciąganiem logiki na zewnątrz. Narzędzia składające przepływ z dekoratorów na zwykłych funkcjach, jak Prefect, tego problemu nie mają, bo przepływ wywołuje się w teście jak każdą inną funkcję. Cena idzie w drugą stronę: struktura grafu jest tam znana dopiero w trakcie wykonania, a nie przed startem.

Uruchamianie i utrzymanie

Platforma składa się z kilku elementów: planisty decydującego, co uruchomić, serwera interfejsu, bazy metadanych i procesów wykonawczych. To znacznie więcej niż pojedyncza usługa i warto to wiedzieć przed decyzją.

Wdrożenie u siebie oznacza utrzymanie tych elementów, kopie zapasowe bazy metadanych i pilnowanie, żeby procesy wykonawcze miały wystarczające zasoby. Przy jednym zespole to kilka godzin miesięcznie, przy większej instalacji osobna rola.

Alternatywą są usługi zarządzane, oferowane przez głównych dostawców chmurowych i przez firmy specjalizujące się w tej platformie. Płacisz wtedy za utrzymanie i zyskujesz czas, tracąc część kontroli nad wersją i konfiguracją.

Trzecia droga to rezygnacja z platformy przy prostych potrzebach. Pięć zadań bez zależności nie wymaga niczego poza harmonogramem, a wdrożenie tej platformy do takiego zestawu jest nieproporcjonalne. Warto policzyć ten koszt uczciwie, bo instalacja rośnie razem z liczbą przepływów i po roku wymaga uwagi, której nikt nie planował.

Migracja z drugiej wersji

Przejście na trzecią wersję jest wykonalne, ale nie sprowadza się do podniesienia numeru w zależnościach.

Największą zmianą są importy. Definicje przepływów sięgające bezpośrednio do wnętrza platformy trzeba przenieść na nową przestrzeń nazw. Przy kilkunastu przepływach to praca na godzinę, przy dwustu warto napisać skrypt zamieniający najczęstsze wzorce.

Drugą jest zmiana nazewnictwa. Pojęcie zbiorów danych zastąpiono zasobami, przy zachowaniu tego samego mechanizmu wejść i wyjść. Kod korzystający ze starej nazwy wymaga poprawki, choć samo działanie pozostaje zbliżone.

Trzecią jest zachowanie przy ponownym uruchamianiu. Poprzednia wersja zawsze używała aktualnego kodu, obecna domyślnie wraca do wersji pierwotnej. To zwykle poprawa, ale jeśli Twój proces poprawiania danych zakładał stare zachowanie, trzeba to ustawić świadomie.

Praktyczna kolejność: najpierw uruchom instalację testową z nową wersją i przepuść przez nią kopię przepływów, potem popraw to, co się nie wczytuje, a na końcu zaplanuj przełączenie produkcji. Wersja trzecia zachowała zgodność wsteczną dla większości definicji, więc lista poprawek zwykle jest krótsza, niż wynikałoby ze zmiany numeru głównego.

Airflow kontra alternatywy

Trzeci wiersz to inna kategoria, choć bywa mylona. Ta platforma planuje powtarzalne przetwarzanie danych o określonych porach albo po zdarzeniach. Rozwiązanie do trwałego wykonania obsługuje procesy biznesowe, które biegną raz na zamówienie i trwają tygodniami. Nakładanie się jest niewielkie.

Wybór między dwoma pierwszymi wierszami zależy od tego, czy zaczynasz od zera. Przy istniejącej instalacji i setkach przepływów przejście jest kosztowne i rzadko uzasadnione. Przy nowym projekcie warto porównać oba, bo różnice w modelu pojęciowym są znaczne.

Argumentem, który zwykle przesądza na korzyść tej platformy, jest liczba gotowych integracji. Połączenie z bazami, magazynami danych, usługami chmurowymi i systemami kolejkowymi zwykle istnieje i jest używane przez wiele zespołów, co znaczy, że przypadki brzegowe zostały już znalezione przez kogoś innego. Przy mniej popularnych narzędziach ta praca spada na Ciebie.

Drugim argumentem jest dostępność ludzi. Doświadczenie z tą platformą jest powszechne, więc nowa osoba w zespole zwykle zna podstawy. To brzmi jak argument miękki, a przy utrzymaniu instalacji przez kilka lat ma znaczenie praktyczne.

Typowe błędy

Pierwszy to nadrabianie zaległości włączone ręcznie albo odziedziczone z konfiguracji drugiej wersji. Przepływ z datą początkową sprzed roku wygeneruje setki przebiegów naraz i zablokuje wszystko inne, choć w trzeciej wersji trzeba się o to postarać, bo domyślnie jest wyłączone.

Drugi to logika przetwarzania w pliku definiującym przepływ. Ten plik jest odczytywany przez planistę wielokrotnie na minutę, więc kod wykonujący zapytania do bazy przy imporcie spowalnia całą instalację.

Trzeci to zadania bez limitu czasu. Zadanie wiszące w nieskończoność zajmuje miejsce w puli i blokuje pozostałe, nie zgłaszając niczego.

Czwarty to przekazywanie dużych danych między zadaniami. Wyniki przechodzą przez bazę metadanych, więc przekazanie tabeli zamiast ścieżki do niej zapycha bazę i spowalnia wszystko.

Piąty to jeden przepływ obejmujący cały proces. Trzydzieści zadań w jednym grafie utrudnia ponawianie fragmentu i sprawia, że błąd na końcu wymaga powtórzenia całości.

Szósty to pomijanie testów. Przepływ jest kodem, więc logikę przetwarzania warto wydzielić do funkcji z testami, zamiast sprawdzać ją wyłącznie przez uruchamianie całego grafu.

FAQ

Do czego służy Airflow?

Do planowania i wykonywania potoków przetwarzania danych z zależnościami między krokami. Platforma pilnuje kolejności, ponowień i widoczności stanu, a przepływ opisujesz kodem w Pythonie jako graf zadań.

Co nowego przyniosła trzecia wersja?

Trzy rzeczy zmieniają sposób pracy: wersjonowanie przepływów, dzięki któremu stary przebieg odtwarza się na kodzie z tamtego czasu, model oparty o zasoby z wyzwalaniem zdarzeniami zamiast samego harmonogramu, oraz stabilna przestrzeń nazw do pisania przepływów, oddzielona od wnętrza platformy.

Czy Airflow nadaje się do małych projektów?

Rzadko. Platforma wymaga bazy metadanych, planisty, serwera interfejsu i procesów wykonawczych, więc przy pięciu niezależnych zadaniach koszt utrzymania przewyższa korzyść. Sens pojawia się przy zależnościach między zadaniami i przy potrzebie widoczności, co zwykle zaczyna się od kilkunastu przepływów.

Czym różni się od Temporal?

To inne przeznaczenie, mimo podobnego brzmienia opisów. Airflow planuje powtarzalne przetwarzanie danych, uruchamiane cyklicznie albo po zdarzeniu. Temporal wykonuje procesy biznesowe uruchamiane pojedynczo, które mogą trwać tygodniami i wymagają odporności na awarie w połowie.

Czy zadania muszą być w Pythonie?

Opis przepływu tak, ale trzecia wersja pozwala wykonywać poszczególne zadania w innych językach. To przydatne, gdy logika przetwarzania już istnieje i przepisywanie jej byłoby pracą wykonaną wyłącznie po to, żeby zmieścić się w narzędziu.

Dokumentacja stoi na stronie projektu, a opis zmian w trzeciej wersji znajdziesz we wpisie o jej wydaniu.