Od autora

W dzisiejszych czasach generujemy coraz więcej danych, a co za tym idzie - coraz więcej informacji. Dane mogą mieć różne formaty: od prostych plików tekstowych po filmy i obrazy. Każdy z tych obiektów może zawierać wiele przydatnych informacji, które następnie mogą być wykorzystane do podejmowania właściwych decyzji. Poza formatem danych należy również pamiętać o sposobie przechowywania i przenoszenia danych. W tym celu wykorzystujemy wspomniane pliki przechowywane na dysku twardym, w chmurze lub nawet jako załączniki do wiadomości w poczcie elektronicznej. Wiele informacji przechowujemy też w specjalnych środowiskach (np. bazach danych), które również dzielą się zależnie od ich cech oraz przeznaczenia (np.: relacyjne bazy danych, bazy dokumentowe, bazy grafowe).

Aby móc wykorzystać pełen potencjał danych, musimy często wykonywać na nich odpowiednie procesy analityczno-agregujące, by uzyskać jak najwięcej informacji, które następnie pozwolą nam na wyciągnięcie wniosków i podjęcie decyzji. Tutaj może się pojawić problem: w jaki sposób pogodzić rozmaite formaty danych, odpowiednio je ze sobą połączyć, wykonać mapowanie i konwertowanie itd.? Istnieje wiele formatów, ale ostatecznie zależy nam na jednym, wspólnym formacie danych pozwalającym na sprawne i czytelne dla człowieka analizy. Dodatkowo narzędzi do wykonywania analizy i procesowania danych jest na rynku sporo, przykładowo: Apache Spark, Microsoft Power BI, platforma Databricks czy biblioteki Pandas i NumPy. To generuje kolejny problem: jak pogodzić nie tylko różne formaty danych, lecz także różne środowiska i narzędzia? Tutaj z pomocą przychodzą tzw. orkiestratory zadań, do których należy Apache Airflow Orkiestratory służą do zarządzania zadaniami w ramach danego problemu.

Jako osoba techniczna starałem się pisać tę książkę w sposób raczej "dokumentowy". Skupiałem się głównie na elementach praktycznych - znajdziesz tutaj przede wszystkim opis poszczególnych modułów narzędzia Apache Airflow, przeprowadzę Cię przez proces instalacji i przygotowania środowiska pracy oraz wyróżnię poszczególne elementy Airflow, abyś zakończył lekturę nie tylko z wiedzą na temat orkiestratora, ale również z dobrymi praktykami. Do książki zostały dołączone materiały, które zawierają więcej przykładów niż te, które znajdziesz na poniższych stronach, dlatego warto o nich pamiętać - dzięki temu możesz uniknąć niepotrzebnego przepisywania z grafik.

Nie przedłużając - życzę przyjemnej i owocnej nauki.

Wprowadzenie

Podstawowa architektura

Do podstawowych elementów Apache Airflow (rysunek 0.1) należą:

Scheduler - planuje kolejność wykonywania zadań. Otrzymuje również informacje o ich statusach. Worker - na nim są wykonywane zadania otrzymane od schedulera. W klastrze Apache Airflow może być więcej niż jeden worker. Innymi słowy: jest to przestrzeń/środowisko dla zadań. Executor - często jest elementem schedulera, ponieważ to scheduler zleca wykonanie zadań na elemencie worker. Webserver - interfejs graficzny pozwalający na zarządzanie Apache Airflow. To on stanowi podstawową komunikację z użytkownikiem. DAG Directory - miejsce, w którym przechowujemy DAG-i (Directed Acyclic Graph). W ramach niego planujemy zadania oraz kolejność ich wykonywania. Co najważniejsze, Apache Airflow traktuje DAG jak obiekt, więc każde uruchomienie DAG-a będzie powodowało utworzenie instancji. Więcej szczegółów w kolejnych rozdziałach. Metadata Database - baza danych (PostgreSQL, MySQL lub sqlite) do przechowywania metadanych: zmienne, połączenia, stany zadań itd.

Rysunek 0.1. Podstawowa architektura Apache Airflow

Bardziej szczegółowe architektury poznasz w rozdziale 10., "Symulacja środowiska HA", w którym utworzysz środowisko typu HA (High Availability).

Przygotowanie środowiska pracy

Wymagania

Na początku trzeba zaznaczyć, że pomyślna konfiguracja Apache Airflow może zostać przeprowadzona jedynie na systemach operacyjnych z rodziny Linux lub macOS. Jeżeli pracujesz w systemie Windows, rekomendowana jest instalacja Linuksa (nawet na maszynie wirtualnej) lub wykorzystanie konteneryzacji za pomocą Dockera lub Podmana. Dodatkowo należy zainstalować wspieraną wersję Pythona. Na czas tworzenia książki Apache Airflow wspiera Pythona od wersji 3.8.

Do przygotowania naszego środowiska pracy będziemy potrzebowali:

systemu macOS lub Linux (Linux może być jako maszyna wirtualna lub kontener), Pythona w wersji 3.12.x, bazy danych PostgreSQL w wersji 13 lub nowszej (ja korzystam z 16.8).

Apache Airflow można uruchomić za pomocą WSL2 w systemie Windows, ale to rozwiązanie nie jest wspierane przez twórców i bywa rzadko wykorzystywane.

Instalacja modułów i zależności

Przejdźmy do instalacji niezbędnych bibliotek. Na początku skonfigurujmy nasze środowisko pracy. W tym celu przygotujemy katalog o nazwie airflow-sandbox i utworzymy w nim środowisko wirtualne Pythona.

Teraz możemy przejść do instalacji Apache Airflow. Zgodnie z dokumentacją możemy zbudować całe narzędzie za pomocą źródła dostępnego na GitHubie lub wykorzystać gotowe pakiety z https://www.pypi.org. Rekomendowana jest druga opcja, w ramach której należy wykonać polecenie:

Apache Airflow ma tak dużo zależności i dodatkowych rozszerzeń (providers), że rekomendowana jest instalacja ze wskazaniem ograniczeń za pomocą opcji -constraints. W dokumentacji znajdziemy link do listy ograniczeń dla każdej wersji Airflow. Jeżeli zatem chcemy wykonać "kompletną i poprawną instalację" dla naszej wersji Pythona, należy wykonać polecenie:

Niestety powyższe polecenia nie instalują wszystkich potrzebnych narzędzi i instalują tylko podstawowe dodatki (providers, o których będzie mowa w kolejnych rozdziałach). Jeżeli chcemy wykorzystać bazę danych MySQL lub PostgreSQL, będziemy potrzebować dodatkowych modułów, aby móc połączyć się z bazą danych. Dla bazy PostgreSQL należy zainstalować psycopg2-binary, a dla MySQL mysqlclient. Zależnie od wyboru bazy danych należy wykonać dodatkowo poniższe polecenia.

PostgreSQL:

MySQL:

Od tej pory w naszym środowisku wirtualnym Pythona mamy dostęp do polecenia/ narzędzia airflow.

Apache Airflow nie wspiera bazy MariaDB, pomimo jej podobieństw do MySQL-a.

Do materiałów zostały dołączone moduły oraz plik tekstowy z ograniczeniami, aby móc zainstalować je lokalnie z poziomu katalogu w trybie offline.

Inicjowanie środowiska

Aby uruchomić kompletne środowisko testowe oparte na sqlite, możemy wykonać polecenie airflow standalone i poczekać na uruchomienie wszystkich elementów narzędzia oraz otrzymać hasło do użytkownika admin. Jednak skupimy się na instalacji "krok po kroku" oraz różnicach w tym podejściu.

Na początku musimy zdefiniować katalog domowy Apache Airflow - u mnie będzie to /airflow. Robimy to za pomocą zmiennej środowiskowej AIRFLOW_HOME. We wskazanym katalogu będzie tworzona cała struktura orkiestratora. Wykonujemy w terminalu polecenie:

Teraz możemy zainicjować domyślną konfigurację za pomocą airflow db migrate. W naszym katalogu pojawią się nowe elementy:

Plik airflow.cfg - konfiguracja, którą omówię w kolejnym kroku. Plik airflow.db - baza sqlite, która jest domyślna dla Apache Airflow. Nie będzie nam ona potrzebna, ponieważ przełączymy się na MySQL/PostgreSQL. Katalog logs - tutaj będą logowane operacje wykonywane przez elementy Apache Airflow.

Polecenie airflow db migrate wygeneruje komunikat podobny do tego z rysunku 0.2.

Rysunek 0.2. Wynik polecenia airflow db migrate

Przejdźmy teraz do pliku airflow.cfg. Plik jest obszerny, ponieważ domyślnie jest generowany wraz z komentarzami do zmiennych. Wiele atrybutów będzie omówionych w kolejnych rozdziałach, natomiast na ten moment warto przyjrzeć się poniższym:

dags_folder - tutaj będziemy umieszczać nasze DAG-i oraz ich elementy, np. skrypty SQL-a. Domyślnie jest to katalog dags w katalogu domowym. default_timezone - domyślna strefa czasowa wykorzystywana przez Airflow. Wpływa ona na działanie większości elementów narzędzia, głównie na harmonogram wykonywania zadań. Co najważniejsze, wszystkie daty będą tłumaczone na strefę czasową ustawioną pod tą właśnie zmienną. Przykładowo: DAG utworzony dla strefy czasowej Europe/Amsterdam zostanie ustawiony na UTC. Rekomendowane jest wykorzystywanie UTC dla projektów w różnych strefach czasowych. Dla naszego środowiska do nauki możemy ustawić czas polski, czyli zmienić wartość na Europe/Warsaw. executor - kto będzie wykonywał nasze zadania w ramach orkiestracji. Domyślnie jest to SequentialExecutor, który wykonuje wszystkie zadania w sposób sekwencyjny (jedno po drugim), dlatego jest on używany jedynie w środowiskach testowych. My będziemy korzystać na początku z LocalExecutor, ponieważ pozwoli nam on lokalnie wykonywać zadania w sposób równoległy. parallelism - odpowiada za to, ile zadań (ich instancji) w całym Apache Airflow może być uruchomionych jednocześnie. Przykładowo: jeśli mamy DAG zawierający trzy zadania i uruchomimy go cztery razy, powoduje to uruchomienie do 12 instancji zadań jednocześnie. U mnie będzie to wartość 12. max_active_tasks_per_dag - maksymalna liczba zadań aktywnych jednocześnie w DAG-u. Ustawiam wartość na 4. dags_are_paused_at_creation - powoduje, że nowo utworzony DAG nie zostanie automatycznie uruchomiony. Najlepiej jest zostawić domyślną wartość. max_active_runs_per_dag - maksymalna liczba uruchomionych instancji DAG-a (a nie zadań wewnątrz). Dla swojego środowiska ustawię wartość 8. load_examples - wartość True (domyślna) spowoduje załadowanie przykładowych DAG-ów. Rekomenduję ustawienie na False, aby mieć czyste środowisko pracy. dag_dir_list_interval - ustala w sekundach, jak często Airflow skanuje katalog z DAG-ami w poszukiwaniu nowych plików (nowych DAG-ów). Domyślnie jest to 5 minut (300), natomiast dla usprawnienia naszej nauki ustawimy 1 minutę (60). sql_alchemy_conn - definiuje połączenie do bazy danych. Domyślnie jest to sqlite w katalogu domowym Airflow. Natomiast nam zależy, aby połączyć się z MySQL-em lub PostgreSQL-em, by móc korzystać z LocalExecutor. Apache Airflow korzysta z SQLAlchemy do połączenia z bazą danych. Ja będę korzystał z bazy danych PostgreSQL.

Oto szablony dla zmiennej sql_alchemy_conn, jeżeli chcemy zamienić w konfiguracji bazę danych SQLite na PostgreSQL lub MySQL:

Przygotowanie bazy danych/uruchomienie Apache Airflow

Po wprowadzeniu powyższych zmian możemy załadować nową konfigurację. W tym celu na początku weryfikujemy, czy możemy połączyć się z bazą danych za pomocą polecenia airflow db check, a następnie ponownie wykonujemy airflow db migrate przy pomyślnym połączeniu z bazą. Poniżej możesz zauważyć, że teraz Apache Airflow łączy się z bazą PostgreSQL. Wyniki poleceń możemy zobaczyć na rysunku 0.3.

Rysunek 0.3. Wynik polecenia db migrate po podłączeniu do PostgreSQL-a

Możemy teraz przejść do utworzenia użytkownika. Robimy to w terminalu za pomocą polecenia:

Aby utworzyć administratora, wykonujemy polecenie w konsoli:

Jak widzimy, to polecenie zawiera kilka opcji. Aby je wykonać, musimy zdefiniować:

nazwę użytkownika (--username), początkowe hasło (--password), imię (--firstname), nazwisko (--lastname), adres e-mail (--email), rolę (--role).

Informacje dotyczące adresu e-mail czy nazwiska nie muszą być prawdziwe, natomiast najważniejsze są dane do logowania oraz rola użytkownika. Domyślnie Apache Airflow ma kilka wstępnie zdefiniowanych ról:

Admin - ma pełny dostęp do wszystkich funkcji Airflow, w tym zarządzania użytkownikami, konfiguracji systemu, edycji DAG-ów, uruchamiania i zatrzymywania zadań oraz przeglądania logów. User - standardowy użytkownik, który może przeglądać i uruchamiać DAG-i, ale nie ma dostępu do zarządzania użytkownikami ani konfiguracji systemu. Op - użytkownik z ograniczonymi uprawnieniami administracyjnymi, który może zarządzać działaniem DAG-ów, ale nie ma dostępu do konfiguracji systemu ani zarządzania użytkownikami. Viewer - użytkownik tylko do odczytu, który może przeglądać DAG-i, ich stan i logi, ale nie może ich edytować ani uruchamiać. Public - użytkownik bez autoryzacji, który ma dostęp tylko do bardzo ograniczonych informacji (lub nie ma dostępu do żadnych).

Oczywiście możemy definiować własne role, co będzie omówione w kolejnych rozdziałach.

Po wykonaniu naszego polecenia zobaczymy na konsoli ogromną ilość informacji w temacie tworzenia ról użytkownika i przypisywania do nich odpowiednich uprawnień. Tak wiele zadań występuje za każdym razem, kiedy tworzymy użytkownika lub wykonujemy jakiekolwiek polecenie związane z uprawnieniami na świeżej bazie danych dla Apache Airflow. Jeżeli wszystko pójdzie dobrze, to na końcu pojawi się informacja, że użytkownik został utworzony (rysunek 0.4).

Rysunek 0.4. Potwierdzenie utworzenia użytkownika

Powoli zbliżamy się do końca - możemy nareszcie uruchomić Apache Airflow. Będzie się on składał z dwóch elementów: webserver oraz scheduler. Oba elementy będą działać w terminalu, dopóki nie zamkniemy okien. Aby uruchomić narzędzia, wykonujemy poniższe polecenia, każde w innym terminalu:

Pozostaje nam jeszcze zweryfikować nasze pliki z logami. Jeżeli webserver i scheduler uruchomiły się prawidłowo, to w pliku znajdziemy wpisy podobne do tych na rysunkach 0.5 oraz 0.6. Dodatkowo pojawi się nowy plik: airflow-webserver.pid, zawierający ID procesu, na którym działa webserver.

Rysunek 0.5. Potwierdzenie uruchomienia webservera

Rysunek 0.6. Potwierdzenie uruchomienia schedulera

Możesz wykorzystać narzędzie nohup i znak & do uruchomienia zadań w tle. Przykładowe polecenia:

Powyższe wpisy oczywiście mogą się różnić od moich. Najważniejsze, aby w plikach pojawiła się informacja, że dla webservera uruchomiono serwer gunicorn, który działa pod adresem localhost na porcie 8080, oraz workery, które działają na porcie 8793.

Nareszcie jesteśmy gotowi, aby zalogować się do Apache Airflow. Przechodzimy na https://localhost:8080 w naszej przeglądarce i widzimy ekran powitalny (rysunek 0.7). Na środku mamy formularz pozwalający na zalogowanie się do orkiestratora, natomiast w prawym górnym rogu informacje o czasie oraz możliwość ustawienia strefy czasowej do wyświetlenia. Domyślnie będziemy mieli do wyboru czas UTC i naszą strefę czasową oraz czas ustawiony w airflow.cfg (obok strefy czasowej serwera zobaczymy niebieski prostokąt z napisem Server), a także przycisk przenoszący do formularza logowania.

Rysunek 0.7. Formularz logowania do Apache Airflow

Podajemy dane logowania naszego użytkownika (którego niedawno utworzyliśmy) i widzimy "ekran powitalny" Apache Airflow (rysunek 0.8). Nareszcie mamy gotowe środowisko pracy!

Rysunek 0.8. Ekran po zalogowaniu

Pokrótce omówię poszczególne elementy strony głównej - szersze opisy pojawią się w kolejnych rozdziałach. Na górnej belce (ten sam poziom co logo orkiestratora) mamy sekcje:

DAGs - ekran główny, na którym aktualnie się znajdujemy. Cluster Activity - statystyka oraz podstawowy monitoring naszego klastra. Datasets - lista i aktywność zbiorów danych oraz elementów, które z nich korzystają. Security - kwestie administracyjne, jak tworzenie ról użytkownika czy samych użytkowników. Browse - sekcja pozwalająca przeglądać historie wykonywania zadań bądź DAG-ów. Admin - ma dostęp do dodatkowych elementów dla naszych zadań, np. do połączenia do baz danych, zmiennych globalnych. Docs - odnośniki do oficjalnej dokumentacji Airflow. Przyciski umożliwiające ustawienie jasnego/ciemnego motywu, wyświetlenia czasu w wybranej strefie czasowej, nazwa naszego użytkownika.

Wróćmy teraz do głównej zakładki. Będziemy mieli tutaj wyświetloną listę naszych DAG-ów i ogólną statystykę (liczba zatrzymanych DAG-ów, wyszukiwarka po etykietach/nazwie i odświeżenie stanu listy). Dla każdego DAG-a będziemy widzieć informacje (od lewej):

Czy jest zatrzymany. Nazwa wraz z etykietami (DAG). Właściciel (Owner) - domyślnie airflow. Historia uruchomień (Runs) - są to wszystkie uruchomienia danego DAG-a. Harmonogram (schedule). Ostatni czas wykonania (Last Run). Kolejne planowane wykonanie (Next Run). Aktualny stan zadań (Recent Tasks) - jest to stan wszystkich aktualnie zaplanowanych zadań. Jeżeli DAG zostanie zatrzymany, to widzimy tylko ostatni status. Uruchomienie/przeładowanie/usunięcie DAG-a (Actions). Odnośniki (Links) - przejście do szczegółowego panelu wybranego DAG-a.

Na sam koniec możesz zajrzeć na profil użytkownika (rysunek 0.9). Będą tam informacje podane podczas jego tworzenia oraz opcja resetowania hasła i aktualizacji danych. Co ciekawe, nie mamy możliwości zmiany adresu e-mail.

Rysunek 0.9. Profil użytkownika

Rozdział 1. DAG i zadania

Zanim przejdziemy do tworzenia pierwszych elementów do zarządzania naszymi procesami, chciałbym wspomnieć o materiałach dołączonych do tej książki. Każdy przykład został również umieszczony w materiałach dodatkowych jako plik, którego nazwa jest podana obok kodu.

Pierwszy DAG

Podstawowym elementem orkiestracji zadań jest DAG ( Direct Acyclic Graph ). Zgodnie z nazwą nasze zadania wewnątrz DAG-a muszą być skierowane w jednym kierunku (nie możemy cofać się lub wykonywać zadania w pętli) i mieć określony cel (być deterministyczne). Każdy punkt w naszym grafie to pojedyncze zadanie, określane za pomocą obiektów Airflow. Obiekty powiązane z zadaniami dzielimy na:

operatory ( operators ) - wykonują określone zadanie (np. przeliczanie wartości, kopiowanie plików); sensory ( sensors) - weryfikują aktualny stan danego elementu (np. jest możliwe połączenie z bazą danych lub plik istnieje); uchwyty ( hooks ) - rozszerzenia dla operatorów i sensorów. Za ich pomocą możemy dodać nowe funkcje do powyższych elementów, np. połączenie do bazy danych czy wykonywanie kopii zapasowych.

Wszystkie DAG-i będziemy tworzyć jako pliki Pythona ( *.py ) w katalogu dags . W tym katalogu będziemy również umieszczać elementy naszych zadań, np. skrypty lub moduły z funkcjami Pythona. Apache Airflow pozwala również na grupowanie DAG-ów poprzez tworzenie podkatalogów (webserver będzie skanował całą zawartość katalogu dags ). Aby zwiększyć czytelność i zachować porządek, każdy rozdział będzie miał dedykowany katalog według formatu: r <numer rozdziału> , zaczynając od 2 (rysunek 1.1).

Rysunek 1.1. Struktura katalogów

Przejdźmy zatem do utworzenia pierwszego DAG-a o nazwie hello_dag . Tworzymy w katalogu r2 plik o nazwie hello_dag.py . DAG możemy tworzyć na trzy sposoby:

Za pomocą operatora with - najpopularniejszy sposób, głównie wykorzystywany. Za pomocą podstawowych konstruktorów - najrzadziej wykorzystywany, ale jego elementy są używane w punktach 1 i 3. Za pomocą dekoratorów - najnowszy sposób, wykorzystywany w podejściu TaskFlow, rekomendowany przez twórców orkiestratora.

Nie wszystkie operatory mają wersję dla dekoratorów (np. operatory dla SQL-a jej nie mają). Z tego powodu TaskFlow korzysta częściowo z podejścia, które w pełni wykorzystujemy z operatorem with . Jeżeli nie przedstawiono wersji TaskFlow, to najprawdopodobniej nie ma dekoratora.

W książce skupiam się na pierwszym sposobie i na jego podstawie omawiam działanie poszczególnych funkcji. Dodatkowo obok jest prezentowana wersja TaskFlow, która w praktyce korzysta również z pozostałych opcji. Oczywiście każde podejście ma swoje plusy i minusy.

Przejdźmy teraz do importowania niezbędnych elementów:

Zaimportowaliśmy dwa obiekty: DAG, który pozwoli nam utworzyć graf, oraz Empty Operator . Operator w praktyce nie wykonuje żadnej czynności, pozwalając nam utworzyć puste zadanie. Jest to niezbędne, ponieważ każdy DAG musi zawierać co najmniej jedno zadanie!

Każdy z elementów ma wiele argumentów, które będziemy poznawać sukcesywnie z każdym rozdziałem. Aby utworzyć najprostszy DAG, powinniśmy skonfigurować:

dag_id - ID DAG-a. Najczęściej ustawiamy takie samo jak nazwa pliku, ponieważ rekomendowane jest utworzenie oddzielnego pliku dla każdego DAG-a (1 plik = 1 DAG). ID musi być unikatowe w całym środowisku Apache Airflow. start_time - data startu DAG-a. Możemy ustawić datę z przeszłości lub po prostu None . schedule - harmonogram wykonywania dla naszego DAG-a. Jeżeli ustawimy None , to nasz DAG będzie uruchamiany "na żądanie". tags - lista etykiet. Nie jest wymagana do utworzenia DAG-ów, ale usprawni nam to zarządzanie nimi.

Etykiety będę wykorzystywał głównie do oznaczania rozdziałów oraz sposobów implementacji DAG-a. Dodatkowo wersje pod TaskFlow zawierają dopisek _tf w nazwie. Etykieta with oznacza klasyczną implementację, a taskflow prezentuje wersje z wykorzystaniem dekoratorów.

W przypadku zadania potrzebujemy jednego elementu:

task_id - ID zadania. Unikatowe na poziomie DAG-a.

Wersja klasyczna - hello_dag.py