🥇Apache Airflow: uproszczenie ETL

Cześć, nazywam się Dmitrij Logwinenko—Inżynier danych w dziale analityki grupy firm „Wiezot”.

Opowiem wam o wspaniałym narzędziu do tworzenia procesów ETL—Apache Airflow. Ale Airflow jest tak wszechstronny, że warto mu się przyjrzeć nawet, jeśli nie zajmujecie się przetwarzaniem danych, a raczej potrzebujecie okresowo uruchomić jakieś procesy i monitorować ich wykonanie.

Tak, nie tylko będę opowiadał, ale również pokazywał: program zawiera wiele kodu, zrzutów ekranu i rekomendacji.

Co zazwyczaj widzisz, gdy guglę Airflow / Wikimedia Commons

Spis treści

Wprowadzenie
Część główna, praktyczna (i trochę teoretyczna)
Część końcowa, informacyjno-referencyjna
- Pułapki, które zebraliśmy dla was
- Środki do jeszcze większej automatyzacji
Linki

Wprowadzenie

Apache Airflow—jest jak Django:

napisany w Pythonie,
ma świetny panel administracyjny,
nieograniczone możliwości rozbudowy,

— tylko lepszy, stworzony do zupełnie innych celów, a mianowicie (jak napisano wcześniej):

uruchamianie i monitorowanie zadań na nieograniczonej liczbie maszyn (na ile pozwoli ci Celery/Kubernetes i twoje sumienie)
z dynamicznym generowaniem workflow z bardzo prostego do napisania i zrozumienia kodu Python
i możliwością łączenia ze sobą dowolnych baz danych i API za pomocą zarówno gotowych komponentów, jak i stworzonych przez siebie wtyczek (co jest niezwykle proste).

Używamy Apache Airflow w taki sposób:

zbieramy dane z różnych źródeł (wiele instancji SQL Server i PostgreSQL, różne API z metrykami aplikacji, nawet 1C) do DWH i ODS (w naszym przypadku to Vertica i Clickhouse).
jak zaawansowany cron, który uruchamia procesy konsolidacji danych w ODS oraz monitoruje ich konserwację.

Do niedawna nasze potrzeby były zaspokojone przez jeden mały serwer z 32 rdzeniami i 50 GB pamięci. W Airflow działa:

ponad 200 DAG-ów (właściwie workflow, w które załadowaliśmy zadania),
w każdym średnio po 70 zadań,
to dobra rzecz uruchamiana (też średnio) raz na godzinę.

A o tym, jak się rozwijaliśmy, napiszę poniżej, a teraz zdefiniujmy über-zadanie, które będziemy rozwiązywać:

Mamy trzy instancje SQL Server, każda z nich z 50 bazami danych — instancjami jednego projektu, co oznacza, że ich struktura jest podobna (prawie wszędzie, muahaha), co sprawia, że w każdej z nich znajduje się tabela Orders (można bez problemu wpasować taką tabelę w każdą firmę). Zbieramy dane, dodając pola pomocnicze (serwer źródłowy, baza źródłowa, identyfikator zadania ETL) i naiwności rzucimy je do, powiedzmy, Vertica.

Zaczynamy!

Część główna, praktyczna (i trochę teoretyczna)

Po co nam (i wam) to?

Kiedy drzewa były duże, a ja byłem prostym SQL-pracownikiem w jednej z rosyjskich sieci detalicznych, zajmowaliśmy się procesami ETL, czyli strumieniami danych, za pomocą dwóch dostępnych nam narzędzi:

Informatica Power Center — to niezwykle zaawansowany system, bardzo wydajny, z własnym sprzętem i własnym wersjonowaniem. Używałem może 1% jego możliwości. Dlaczego? Po pierwsze, ten interfejs jakoś z lat 00 psychicznie na nas naciskał. Po drugie, to narzędzie jest przystosowane do niezwykle skomplikowanych procesów, szałowego ponownego wykorzystania komponentów i innych ważnych funkcji dla przedsiębiorstw. O tym, że kosztuje jak skrzydło Airbus A380 rocznie, milczymy.
Uwaga, zrzut ekranu może sprawić ból osobom poniżej 30. roku życia

SQL Server Integration Server — to towarzyszący nam w naszych wewnętrznych strumieniach projektowych. A tak naprawdę: SQL Server już używamy, więc nie używać jego narzędzi ETL byłoby jakoś nieodpowiedzialnie. Wszystko w nim jest dobrze: interfejs piękny, a raporty wykonania... Ale nie za to kochamy produkty programowe, o nie, nie za to. Możemy wersjonować jego dtsx (który jest XML-em z mieszającymi się podczas zapisywania węzłami), ale po co? A zrobienie pakietu zadań, który przeniesie setkę tabel z jednego serwera na drugi? Tak, nawet jeśli masz dwadzieścia, palec wskazujący odpadnie ci od klikania w mysz. Ale zdecydowanie wygląda bardziej stylowo:

Zdecydowanie szukaliśmy wyjść. Sprawa nawet prawie doszła do samodzielnego generatora pakietów SSIS…

… a potem znalazła mnie nowa praca. A tam dotarł do mnie Apache Airflow.

Kiedy dowiedziałem się, że opisy procesów ETL to po prostu kod Pythona, prawie nie tańczyłem z radości. W ten sposób strumienie danych poddano wersjonowaniu i diffom, a zrzucenie tabel z jednolitą strukturą z setki baz danych do jednego celu stało się zadaniem w kodzie Pythona na półtorej do dwóch ekranów 13 cali.

Budujemy klaster

Nie róbmy z tego przedszkola i nie mówmy o zupełnie oczywistych rzeczach, takich jak instalacja Airflow, wybranej przez Was bazy danych, Celery i innych rzeczy opisanych w dokumentacji.

Abyśmy mogli od razu przystąpić do eksperymentów, naszkicowałem docker-compose.yml w którym:

Zbudujemy właściwie Airflow: Scheduler, Webserver. Tam również będzie działać Flower do monitorowania zadań Celery (ponieważ już go wpakowali w apache/airflow:1.10.10-python3.7, a my nie mamy nic przeciwko);
PostgreSQL, w którym Airflow będzie zapisywał swoje informacje operacyjne (dane planisty, statystyki wykonania itd.), a Celery – oznaczał zakończone zadania;
Redis, który będzie pełnił rolę brokera zadań dla Celery;
Celery worker, który zajmie się bezpośrednim wykonywaniem zadań.
Do folderu . /dags będziemy wrzucać nasze pliki z opisem DAG-ów. Będą one pobierane na bieżąco, więc nie ma potrzeby zerowania całego stosu po każdym drobnym błędzie.

W niektórych miejscach kod w przykładach jest podany niekompletnie (aby nie zaśmiecać tekstu), a gdzie indziej modyfikuje się w trakcie. Pełne, działające przykłady kodu można znaleźć w repozytorium. https://github.com/dm-logv/airflow-tutorial.

docker-compose.yml

wersja: '3.4'

x-airflow-config: &airflow-config
  AIRFLOW__CORE__DAGS_FOLDER: /dags
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=
  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow

  AIRFLOW__CORE__PARALLELISM: 128
  AIRFLOW__CORE__DAG_CONCURRENCY: 16
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4
  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'

  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'
  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'

  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflow

x-airflow-base: &airflow-base
  image: apache/airflow:1.10.10-python3.7
  entrypoint: /bin/bash
  restart: always
  volumes:
    - ./dags:/dags
    - ./requirements.txt:/requirements.txt

usługi:
  # Redis jako broker Celery
  broker:
    image: redis:6.0.5-alpine

  # DB dla metadanych Airflow
  airflow-db:
    image: postgres:10.13-alpine

    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow

    volumes:
      - ./db:/var/lib/postgresql/data

  # Główna kontener z Airflow Webserver, Scheduler, Celery Flower
  airflow:
    <<: *airflow-base

    environment:
      <
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint initdb &&
          (/entrypoint webserver &) &&
          (/entrypoint flower &) &&
           /entrypoint scheduler"

    ports:
      # Celery Flower
      - 5555:5555
      # Airflow Webserver
      - 8080:8080

  # Pracownik Celery, będzie skalowany za pomocą `--scale=n`
  worker:
    <<: *airflow-base

    environment:
      <
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

Uwagi:

W konfiguracji kompozycji w dużej mierze bazowałem na znanym obrazie puckel/docker-airflow — koniecznie sprawdź. Może w życiu nie będziesz potrzebował niczego więcej.
Wszystkie ustawienia Airflow są dostępne nie tylko przez airflow.cfg, ale także przez zmienne środowiskowe (chwała programistom), z czego świadomie skorzystałem.
Oczywiście, nie jest to gotowe do produkcji: celowo nie ustawiłem heartbeatów dla kontenerów, nie przejmowałem się zabezpieczeniami. Ale zrobiłem minimum potrzebne do naszych eksperymentów.
Zwróć uwagę, że:
- Folder z DAG-ami musi być dostępny zarówno dla planisty, jak i dla pracowników.
- To samo dotyczy wszystkich zewnętrznych bibliotek — wszystkie muszą być zainstalowane na maszynach z planistą i pracownikami.

A teraz po prostu:

$ docker-compose up --scale worker=3

Po tym, jak wszystko się uruchomi, można spojrzeć na interfejsy webowe:

Airflow: http://127.0.0.1:8080/admin/
Flower: http://127.0.0.1:5555/dashboard

Podstawowe pojęcia

Jeśli nic nie zrozumiałeś z tych wszystkich „dagów”, oto krótki słowniczek:

Scheduler — najważniejszy człowiek w Airflow, który kontroluje, aby pracowały roboty, a nie człowiek: monitoruje harmonogram, aktualizuje dagi, uruchamia zadania.
Ogólnie rzecz biorąc, w starych wersjach miał problemy z pamięcią (nie, nie amnezja, a wycieki) i w konfiguracjach pozostał nawet parametr legacyny run_duration — interwał jego ponownego uruchomienia. Ale teraz wszystko jest w porządku.
DAG (znany również jako „dag”) — „graf acykliczny skierowany”, ale taka definicja niewiele komu powie, a w zasadzie to kontener dla współdziałających ze sobą zadań (patrz poniżej) lub odpowiednik Package w SSIS i Workflow w Informatica.
Oprócz dagów mogą być też subdagy, ale prawdopodobnie do nich nie dojdziemy.
DAG Run — zainicjowany dag, któremu przypisany jest swój execution_date. Dagruny jednego daga mogą działać równolegle (jeśli oczywiście uczynisz swoje zadania idempotentnymi).
Operator — to fragmenty kodu odpowiedzialne za realizację konkretnego działania. Istnieją trzy typy operatorów:
- akcja, jak na przykład nasz ulubiony PythonOperator, który potrafi wykonać dowolny (prawidłowy) kod Python;
- transfer, które przenoszą dane z miejsca na miejsce, powiedzmy, MsSqlToHiveTransfer;
- sensor pozwoli reagować lub spowolnić dalsze wykonanie dana do wystąpienia jakiegoś zdarzenia. HttpSensor może zainicjować wskazany punkt końcowy, i kiedy uzyska odpowiedni wynik, uruchomić transfer GoogleCloudStorageToS3Operator. Ciekawski umysł zapyta: „po co? Przecież można robić powtórzenia bezpośrednio w operatorze!” A potem, aby nie zaśmiecać puli zadań zawieszonymi operatorami. Sensor uruchamia się, sprawdza i umiera do następnej próby.
Task — zadeklarowane operatory, niezależnie od rodzaju i przypisane do daga, zdobywają status zadania.
Task instance — kiedy generał-planista zdecydował, że zadania należy wysłać do wykonawców-robotników (bezpośrednio na miejscu, jeśli używamy LocalExecutor lub na zdalnym węźle w przypadku CeleryExecutor), przypisuje im kontekst (tj. zestaw zmiennych — parametrów wykonania), rozwija wzorce komend lub zapytań i umieszcza je w puli.

Generujemy zadania

Na początku określimy ogólny schemat naszego daga, a potem będziemy coraz bardziej zagłębiać się w szczegóły, ponieważ stosujemy pewne nietrywialne rozwiązania.

Zatem w najprostszym przypadku taki dag będzie wyglądał tak:

od datetime import timedelta, datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator

from commons.datasources import sql_server_ds

dag = DAG('orders',
          schedule_interval=timedelta(hours=6),
          start_date=datetime(2020, 7, 8, 0))

def workflow(**context):
    print(context)

for conn_id, schema in sql_server_ds:
    PythonOperator(
        task_id=schema,
        python_callable=workflow,
        provide_context=True,
        dag=dag)

Zacznijmy zrozumieć:

Najpierw zaimportujemy potrzebne biblioteki i coś jeszcze;
sql_server_ds — to List[namedtuple[str, str]] z nazwami połączeń z Airflow Connections oraz bazami danych, z których będziemy pobierać naszą tabelę;
dag to deklaracja naszego dagu, która musi znajdować się w globals(), w przeciwnym razie Airflow go nie znajdzie. Musimy również poinformować dag:
- jak się nazywa orders to będzie miało swoje odzwierciedlenie w interfejsie webowym,
- że zacznie działać od północy ósmego lipca,
- a jego uruchomienie powinno nastąpić mniej więcej co 6 godzin (dla odważnych, zamiast timedelta() można użyć cron-wyrażenie, 0 0 0/6 ? * * *dla mniej odważnych @daily);
workflow() będzie wykonywać główną pracę, ale nie teraz. Teraz po prostu wypiszemy nasz kontekst w dzienniku.
A teraz prosta magia tworzenia zadań:
- przechodzimy przez nasze źródła;
- inicjalizujemy PythonOperator, które będzie wykonywać naszą puste zadanie. Nie zapomnij wskazać unikalnej (w ramach dagu) nazwy zadania i powiązać go z dagiem. Flaga workflow()provide_context w swoim obrocie dostarczy dodatkowe argumenty do funkcji, które starannie zbierzemy za pomocą **context Na razie tyle. Co uzyskaliśmy:.

nowy dag w interfejsie webowym,

ponad półtorej setki zadań, które będą wykonywane równolegle (jeśli na to pozwalają ustawienia Airflow, Celery i moc serwerów).
No, prawie to uzyskaliśmy.

Kto ustawia zależności?

Aby uprościć to wszystko, wprowadziłem do

przetwarzania docker-compose.yml na wszystkich węzłach. requirements.txt Teraz to się dzieje:

Szare kwadraty to instancje zadań, które zostały przetworzone przez planistę.

Trochę czekamy, a zadania są przejmowane przez pracowników:

Zielone, oczywiście, to te, które zostały pomyślnie wykonane. Czerwone — nieco mniej pomyślnie.

Swoją drogą, na naszym produkcie nie ma żadnego folderu,

, synchronizującego się między maszynami — wszystkie dagi znajdują się w . /dagsna naszym Gitlab, a Gitlab CI wdraża aktualizacje na maszyny przy scalaniu w sshpass Gdy pracownicy młócą nasze puste zadania, przypomnijmy sobie o innym narzędziu, które może nam coś pokazać — Flower. master.

Trochę o Flower

Pierwsza strona z podsumowującymi informacjami o węzłach-pracownikach:

Najbardziej szczegółowa strona z zadaniami, które zostały wysłane do pracy:

Najbardziej szczegółowa strona z zadaniami, które rozpoczęły pracę:

Najbardziej nudna strona z raportem naszego brokera:

Najbardziej kolorowa strona - z wykresami stanu zadań i ich czasem realizacji:

Doładujemy to, co nie zostało załadowane

Więc wszystkie zadania są wykonane, można zabrać rannych.

A rannych okazało się całkiem sporo - z różnych powodów. W przypadku prawidłowego użycia Airflow te kwadraty wskazują, że dane zdecydowanie nie dotarły.

Musimy sprawdzić logi i ponownie uruchomić nieudane instancje zadań.

Klikając na dowolny kwadrat, zobaczymy dostępne nam działania:

Można wziąć i wyczyścić to, co się nie udało. To znaczy, zapominamy, że coś się zawaliło, a ta sama instancja zadania zostanie przekazana planerowi.

Oczywiście robienie tego myszką dla wszystkich czerwonych kwadratów nie jest zbyt humanitarne - nie tego oczekujemy od Airflow. Naturalnie mamy broń masowego rażenia: Przeglądaj/Instancje zadań

Wybierzmy wszystko naraz i wyczyśćmy, klikając odpowiednią opcję:

Po wyczyszczeniu nasze zadania wyglądają tak (już nie mogą się doczekać, kiedy harmonogram je zaplanuje):

Połączenia, haki i inne zmienne

Czas spojrzeć na następny DAG, update_reports.py:

from collections import namedtuple
from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG
from airflow.contrib.operators.vertica_operator import VerticaOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.trigger_rule import TriggerRule

from commons.operators import TelegramBotSendMessage

dag = DAG('update_reports',
          start_date=datetime(2020, 6, 7, 6),
          schedule_interval=timedelta(days=1),
          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})

Report = namedtuple('Report', 'source target')
reports = [Report(f'{table}_view', table) for table in [
    'reports.city_orders',
    'reports.client_calls',
    'reports.client_rates',
    'reports.daily_orders',
    'reports.order_duration']]

email = EmailOperator(
    task_id='email_success', dag=dag,
    to='{{ var.value.all_the_kings_men }}',
    subject='Raporty DWH zaktualizowane',
    html_content=dedent("""Szanowni Państwo, raporty zaktualizowane"""),
    trigger_rule=TriggerRule.ALL_SUCCESS)

tg = TelegramBotSendMessage(
    task_id='telegram_fail', dag=dag,
    tg_bot_conn_id='tg_main',
    chat_id='{{ var.value.failures_chat }}',
    message=dedent("""
         Natalko, obudź się, my {{ dag.dag_id }} upadliśmy
        """),
    trigger_rule=TriggerRule.ONE_FAILED)

for source, target in reports:
    queries = [f"TRUNCATE TABLE {target}",
               f"INSERT INTO {target} SELECT * FROM {source}"]

    report_update = VerticaOperator(
        task_id=target.replace('reports.', ''),
        sql=queries, vertica_conn_id='dwh',
        task_concurrency=1, dag=dag)

    report_update >> [email, tg]

Czy wszyscy kiedykolwiek robili aktualizację raportów? To znowu ona: jest lista źródeł, skąd pozyskać dane; jest lista, gdzie je umieścić; nie zapominajmy o powiadomieniu, kiedy wszystko się zdarzyło lub zepsuło (no to nie dotyczy nas, prawda?).

Przejdźmy ponownie przez plik i spójrzmy na nowe, niezrozumiałe elementy:

from commons.operators import TelegramBotSendMessage — nic nie stoi na przeszkodzie, żeby stworzyć własne operatory, co zrobiliśmy, tworząc małe opakowanie do wysyłania wiadomości w Rozblokowanym. (O tym operatorze jeszcze poniżej);
default_args={} — DAG może przekazywać te same argumenty wszystkim swoim operatorom;
to='{{ var.value.all_the_kings_men }}' — pole to będzie generowane dynamicznie za pomocą Jinja i zmiennej z listą e-maili, którą starannie umieściłem w Admin/Variables;
trigger_rule=TriggerRule.ALL_SUCCESS — warunek uruchomienia operatora. W naszym przypadku wiadomość poleci do szefów tylko jeśli wszystkie zależności zostaną wykonane pomyślnie;
tg_bot_conn_id='tg_main' — argumenty conn_id przyjmują identyfikatory połączeń, które tworzymy w Admin/Connections;
trigger_rule=TriggerRule.ONE_FAILED — wiadomości w Telegramie polecą tylko przy wystąpieniu nieudanych zadań;
task_concurrency=1 — zabrania jednoczesnego uruchamiania kilku instancji tego samego zadania. W przeciwnym razie dostaniemy jednoczesne uruchomienie kilku VerticaOperator (zwracających uwagę na jedną tabelę);
report_update >> [email, tg] — wszystkie VerticaOperator zbiorą się, aby wysłać e-mail i wiadomość, tak:

Jednak ponieważ operatory powiadamiające mają różne warunki uruchomienia, będzie działać tylko jeden. W widoku drzewa wszystko wygląda nieco mniej czytelnie:

Powiem kilka słów o makrosach i ich przyjaciołach — zmiennych.

Makrosy to placeholdery Jinja, które mogą wstawiać różne przydatne informacje do argumentów operatorów. Na przykład tak:

SELECT
    id,
    payment_dtm,
    payment_type,
    client_id
FROM orders.payments
WHERE
    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} rozwinie się w zawartość zmiennej kontekstowej execution_date w formacie YYYY-MM-DD: 2020-07-14. Najlepsze jest to, że zmienne kontekstowe są przypisane do konkretnej instancji zadania (kwadratu w widoku drzewa), a przy ponownym uruchomieniu placeholdery ujawniają te same wartości.

Przypisane wartości można zobaczyć za pomocą przycisku Rendered na każdej instancji zadania. Oto tak przy zadaniu wysyłki e-maila:

A tak przy zadaniu wysyłki wiadomości:

Pełna lista wbudowanych makrosów dla ostatniej dostępnej wersji jest dostępna tutaj: Macros Reference

Co więcej, za pomocą wtyczek możemy deklarować własne makrosy, ale to już zupełnie inna historia.

Oprócz predefiniowanych elementów możemy wstawiać wartości naszych zmiennych (już z tego korzystałem w kodzie powyżej). Stwórzmy w Admin/Variables kilku elementach:

Gotowe, można korzystać:

TelegramBotSendMessage(chat_id='{{ var.value.failures_chat }}')

Wartość może być skalara lub może zawierać JSON. W przypadku JSON-a:

bot_config

{
    "bot": {
        "token": 881hskdfASDA16641,
        "name": "Verter"
    },
    "service": "TG"
}

po prostu używamy ścieżki do odpowiedniego klucza: {{ var.json.bot_config.bot.token }}.

Powiem dosłownie jedno słowo i pokażę jeden zrzut ekranu o połączeniach. To wszystko jest elementarne: na stronie Admin/Connections tworzymy połączenie, umieszczamy tam nasze login/y hasła oraz bardziej specyficzne parametry. Oto jak to wygląda:

Hasła można szyfrować (bardziej dokładnie niż w domyślnym wariancie), a można też nie podawać typu połączenia (tak jak zrobiłem dla tg_main) — rzecz w tym, że lista typów jest zakodowana w modelach Airflow i rozszerzeniem nie da się zmienić bez ingerencji w źródła (jeśli coś przeoczyłem — proszę o poprawkę), ale uzyskanie poświadczeń po nazwie nie stanowi dla nas problemu.

Można też stworzyć kilka połączeń o tej samej nazwie: w takim przypadku metoda BaseHook.get_connection(), która wydobywa połączenia po nazwie, będzie zwracać losowe z kilku odpowiedników (byłoby logiczniej zrobić Round Robin, ale zostawimy to na sumieniu programistów Airflow).

Zmienne i Połączenia to niewątpliwie wspaniałe narzędzia, ale ważne jest, aby nie stracić równowagi: które części waszych strumieni przechowujecie bezpośrednio w kodzie, a które — oddajecie do przechowywania Airflow. Z jednej strony łatwo zmienić wartość, na przykład skrzynki mailingowej, może być wygodne przez UI. Z drugiej strony — to w końcu powrót do klikania myszką, od czego chcieliśmy (ja) się uwolnić.

Praca z połączeniami to jedno z zadań hooków. W ogóle hooki Airflow to punkty połączenia z zewnętrznymi usługami i bibliotekami. Na przykład, JiraHook otworzy dla nas klienta do interakcji z Jira (można przenosić zadania tam i z powrotem), a za pomocą SambaHook można przesłać lokalny plik na punkt smb.I zbliżamy się do tego, aby zobaczyć, jak zrealizowany jest

Rozbijamy niestandardowy operator

TelegramBotSendMessage commons/operators.py

Kod z rzeczywistym operatorem: z własnym operatorem:

z typing import Union

z airflow.operators import BaseOperator

z commons.hooks import TelegramBotHook, TelegramBot

class TelegramBotSendMessage(BaseOperator):
    """Wysyła wiadomość do chat_id przy użyciu TelegramBotHook

    Przykład:
        >>> TelegramBotSendMessage(
        ...     task_id='telegram_fail', dag=dag,
        ...     tg_bot_conn_id='tg_bot_default',
        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',
        ...     message='{{ dag.dag_id }} failed :(',
        ...     trigger_rule=TriggerRule.ONE_FAILED)
    """
    template_fields = ['chat_id', 'message']

    def __init__(self,
                 chat_id: Union[int, str],
                 message: str,
                 tg_bot_conn_id: str = 'tg_bot_default',
                 *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._hook = TelegramBotHook(tg_bot_conn_id)
        self.client: TelegramBot = self._hook.client
        self.chat_id = chat_id
        self.message = message

    def execute(self, context):
        print(f'Wyślij "{self.message}" do czatu {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

Tutaj, jak w reszcie Airflow, wszystko jest bardzo proste:

Dziedziczymy po BaseOperator, który wdraża wiele specyficznych dla Airflow rzeczy (zobacz w wolnym czasie)
Ogłosiliśmy pola template_fields, w których Jinja będzie szukać makr do przetworzenia.
Zorganizowaliśmy odpowiednie argumenty dla __init__(), ustaliliśmy domyślne wartości tam, gdzie to konieczne.
Nie zapomnieliśmy również o inicjalizacji przodka.
Otworzyliśmy odpowiedni hook TelegramBotHook, uzyskaliśmy od niego obiekt klienta.
Nadpisaliśmy (przedefiniowaliśmy) metodę BaseOperator.execute(), którą Airflow będzie wywoływać, gdy nadejdzie czas uruchomienia operatora — w niej realizujemy główne działanie, nie zapominając o logowaniu. (Logujemy się, nawiasem mówiąc, bezpośrednio w stdout i stderr — Airflow wszystko przechwyci, pięknie opakuje, rozłoży, gdzie trzeba.)

Zobaczmy, co mamy w commons/hooks.py. Pierwsza część pliku, z samym hookiem:

z typing import Union

z airflow.hooks.base_hook import BaseHook
z requests_toolbelt.sessions import BaseUrlSession

class TelegramBotHook(BaseHook):
    """Hook API Telegram Bot

    Uwaga: dodaj połączenie z pustym typem połączenia i nie zapomnij
    uzupełnić Extra:

        {"bot_token": "YOuRAwEsomeBOtToKen"}
    """
    def __init__(self,
                 tg_bot_conn_id='tg_bot_default'):
        super().__init__(tg_bot_conn_id)

        self.tg_bot_conn_id = tg_bot_conn_id
        self.tg_bot_token = None
        self.client = None
        self.get_conn()

    def get_conn(self):
        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson
        self.tg_bot_token = extra['bot_token']
        self.client = TelegramBot(self.tg_bot_token)
        return self.client

Nawet nie wiem, co tutaj można wyjaśnić, po prostu zaznaczę ważne punkty:

Dziedziczymy, myślimy o argumentach — w większości przypadków będzie to jeden: conn_id;
Nadpisujemy standardowe metody: ja ograniczyłem się do get_conn(), w którym uzyskuję parametry połączenia po nazwie i wyłącznie wydobywam sekcję extra (to pole JSON), w które umieściłem token bota Telegrama zgodnie z moją instrukcją: {"bot_token": "YOuRAwEsomeBOtToKen"}.
Tworzę instancję naszego TelegramBot, przekazując mu już konkretnego tokena.

To wszystko. Klienta można uzyskać z hooka za pomocą TelegramBotHook().client lub TelegramBotHook().get_conn().

I druga część pliku, w którym zrobiłem mikroopakowanie dla Telegram REST API, żeby nie przeciążać tego samego python-telegram-bot dla jednej metody sendMessage.

class TelegramBot:
    """Wrapper API bota Telegrama

    Przykłady:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Cześć, kochanie')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Cześć, kochanie', chat_id=-1762374628374)
    """
    API_ENDPOINT = 'https://api.telegram.org/bot{}/'

    def __init__(self, tg_bot_token: str, chat_id: Union[int, str] = None):
        self._base_url = TelegramBot.API_ENDPOINT.format(tg_bot_token)
        self.session = BaseUrlSession(self._base_url)
        self.chat_id = chat_id

    def send_message(self, message: str, chat_id: Union[int, str] = None):
        method = 'sendMessage'

        payload = {'chat_id': chat_id or self.chat_id,
                   'text': message,
                   'parse_mode': 'MarkdownV2'}

        response = self.session.post(method, data=payload).json()
        if not response.get('ok'):
            raise TelegramBotException(response)

class TelegramBotException(Exception):
    def __init__(self, *args, **kwargs):
        super().__init__((args, kwargs))

Poprawna droga to złożyć to wszystko: commons/operators.py, TelegramBotHook, TelegramBot — wtyczka, umieścić w publicznym repozytorium i wydać jako Open Source.

Podczas studiowania tego, nasze aktualizacje raportów miały czas, aby skutecznie się załamać i wysłać mi komunikat o błędzie na kanał. Idę sprawdzić, co znowu poszło nie tak…

Coś się zepsuło w naszym dagu! Czyż nie na to czekaliśmy? Zgadza się!

Czy będziesz lać?

Czujecie, że coś pominąłem? Wydaje się, że obiecałem przelać dane z SQL Server do Vertica, a tu nagle zjechałem z tematu, łajdak!

To zło było zamierzone, musiałem po prostu wyjaśnić wam pewną terminologię. Teraz możemy iść dalej.

Nasz plan był taki:

Zrobić dag
Wygenerować zadania
Zobaczyć, jak wszystko wyszło ładnie
Przydzielić sesjom numery wgrywań
Pobrać dane z SQL Server
Umieścić dane w Vertica
Zebrać statystyki

Żeby uruchomić to wszystko, zrobiłem małe uzupełnienie do naszego docker-compose.yml:

docker-compose.db.yml

wersja: '3.4'

x-mssql-base: &mssql-base
  obraz: mcr.microsoft.com/mssql/server:2017-CU21-ubuntu-16.04
  restart: zawsze
  środowisko:
    ACCEPT_EULA: Y
    MSSQL_PID: Express
    SA_PASSWORD: SayThanksToSatiaAt2020
    MSSQL_MEMORY_LIMIT_MB: 1024

usługi:
  dwh:
    obraz: jbfavre/vertica:9.2.0-7_ubuntu-16.04

  mssql_0:
    <<: *mssql-base

  mssql_1:
    <<: *mssql-base

  mssql_2:
    <<: *mssql-base

  mssql_init:
    obraz: mio101/py3-sql-db-client-base
    polecenie: python3 ./mssql_init.py
    zależy_od:
      - mssql_0
      - mssql_1
      - mssql_2
    środowisko:
      SA_PASSWORD: SayThanksToSatiaAt2020
    wolumeny:
      - ./mssql_init.py:/mssql_init.py
      - ./dags/commons/datasources.py:/commons/datasources.py

Tam podnosimy:

Vertica jako host dwh z domyślnymi ustawieniami,
trzy instancje SQL Server,
napełniamy bazy ostatnimi danymi (nie zaglądajcie do mssql_init.py!)

Uruchamiamy całe dobro przy pomocy trochę bardziej skomplikowanego polecenia niż ostatnio:

$ docker-compose -f docker-compose.yml -f docker-compose.db.yml up --scale worker=3

Co nasz cudowny randomizer wygenerował, można sprawdzić, korzystając z punktu Data Profiling/Ad Hoc Query:

Głównie, nie pokazywać tego analitykom

Nie będę się szczegółowo zatrzymywać na sesjach ETL wszystko jest trywialne: tworzymy bazę, w niej tabelkę, obwrapowujemy wszystko menedżerem kontekstu, a teraz robimy tak:

z Session(task_name) jako session:
    print('Ładowanie', session.id, 'rozpoczęte')

    # Ładowanie workflow
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

z sys import stderr

class Session:
    """Sesja ETL

    Przykład:
        with Session(task_name) as session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Dobra robota'
    """

    def __init__(self, connection, task_name):
        self.connection = connection
        self.connection.autocommit = True

        self._task_name = task_name
        self._id = None

        self.loaded_rows = None
        self.successful = None
        self.comment = None

    def __enter__(self):
        return self.open()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if any(exc_type, exc_val, exc_tb):
            self.successful = False
            self.comment = f'{exc_type}: {exc_val}n{exc_tb}'
            print(exc_type, exc_val, exc_tb, file=stderr)
        self.close()

    def __repr__(self):
        return (f'')

    @property
    def task_name(self):
        return self._task_name

    @property
    def id(self):
        return self._id

    def _execute(self, query, *args):
        with self.connection.cursor() as cursor:
            cursor.execute(query, args)
            return cursor.fetchone()[0]

    def _create(self):
        query = """
            CREATE TABLE IF NOT EXISTS sessions (
                id          SERIAL       NOT NULL PRIMARY KEY,
                task_name   VARCHAR(200) NOT NULL,

                started     TIMESTAMPTZ  NOT NULL DEFAULT current_timestamp,
                finished    TIMESTAMPTZ           DEFAULT current_timestamp,
                successful  BOOL,

                loaded_rows INT,
                comment     VARCHAR(500)
            );
            """
        self._execute(query)

    def open(self):
        query = """
            INSERT INTO sessions (task_name, finished)
            VALUES (%s, NULL)
            RETURNING id;
            """
        self._id = self._execute(query, self.task_name)
        print(self, 'otwarta')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Sesja nie jest otwarta')
        query = """
            UPDATE sessions
            SET
                finished    = DEFAULT,
                successful  = %s,
                loaded_rows = %s,
                comment     = %s
            WHERE
                id = %s
            RETURNING id;
            """
        self._execute(query, self.successful, self.loaded_rows,
                      self.comment, self.id)
        print(self, 'zamknięta',
              ', udana: ', self.successful,
              ', Załadowano: ', self.loaded_rows,
              ', komentarz:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

Nadszedł czas na pobranie naszych danych z naszych półtora setek tabel. Zrobimy to za pomocą bardzo prostych linii:

source_conn = MsSqlHook(mssql_conn_id=src_conn_id, schema=src_schema).get_conn()

query = f"""
    SELECT 
        id, start_time, end_time, type, data
    FROM dbo.Orders
    WHERE
        CONVERT(DATE, start_time) = '{dt}'
    """

df = pd.read_sql_query(query, source_conn)

Za pomocą hooka uzyskamy z Airflow pymssql-połączenie
W zapytaniu wstawimy ograniczenie w postaci daty — wzorzec wrzuci w funkcję szablonizer.
Przekazujemy nasze zapytanie pandas, które wydobędzie dla nas DataFrame — przyda nam się w przyszłości.

Używam podmiany {dt} zamiast parametru zapytania %s nie dlatego, że jestem złośliwym Buriatyną, ale dlatego, że pandas nie może sobie poradzić z pymssql i podsuwa ostatniemu params: Lista, chociaż ten bardzo chce znajduje się w podzapytaniu. To jest coś, czego osobiście bardzo mi brakowało w innych bazach danych. Jeśli chcę coś porównać z podzapytaniem, to w innych bazach danych porównuję to tylko z wartością skalarną, a dla kilku kolumn muszę pisać.
Zauważ również, że programista pymssql postanowił już go nie wspierać, więc nadszedł czas, aby się przenieść na pyodbc.

Zobaczmy, jak Airflow zdefiniował argumenty naszych funkcji:

Jeśli danych nie ma, nie ma sensu kontynuować. Ale równie dziwnie byłoby uznać, że zrzut był udany. Ale to też nie jest błąd. A-a-a, co robić?! Oto co:

if df.empty:
    raise AirflowSkipException('Brak wierszy do załadowania')

AirflowSkipException powie Airflow, że nie ma błędu, a zadanie omijamy. W interfejsie nie będzie zielonego ani czerwonego kwadratu, a kolor różowy.

Dorzućmy naszym danym kilka kolumn:

df['etl_source'] = src_schema
df['etl_id'] = session.id
df['hash_id'] = hash_pandas_object(df[['etl_source', 'id']])

A mianowicie:

Baza danych, z której pobraliśmy zamówienia,
Identyfikator naszej sesji ładowania (będzie różny dla każdego zadania),
Hash z źródła i identyfikatora zamówienia — aby w końcowej bazie (gdzie wszystko będzie zebrane w jednej tabeli) mieliśmy unikalny identyfikator zamówienia.

Został przedostatni krok: zalać wszystko do Vertica. A, jak nietypowo, jednym z najefektywniejszych sposobów, aby to zrobić, jest użycie CSV!

# Export data to CSV buffer
buffer = StringIO()
df.to_csv(buffer,
          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,
          header=False, float_format='%.8f', doublequote=False, escapechar='\')
buffer.seek(0)

# Push CSV
target_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()

copy_stmt = f"""
    COPY {target_table}({df.columns.to_list()}) 
    FROM STDIN 
    DELIMITER '|' 
    ENCLOSED '"' 
    ABORT ON ERROR 
    NULL 'NUL'
    """

cursor = target_conn.cursor()
cursor.copy(copy_stmt, buffer)

Robimy specjalny odbiornik StringIO.
pandas uprzejmie złoży w nim nasz DataFrame w postaci CSV-wierszy.
Otworzę połączenie do naszej ulubionej bazy Vertica przy użyciu hooka.
A teraz za pomocą copy() wyślemy nasze dane bezpośrednio do Vertica!

Z kierowcy pobierzemy, ile wierszy zostało załadowanych, i powiemy menedżerowi sesji, że wszystko w porządku:

session.loaded_rows = cursor.rowcount
session.successful = True

I to wszystko.

Na prodzie tworzymy docelową tabelę ręcznie. Tutaj pozwoliłem sobie na małą automatyzację:

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'
create_table_query = f"""
    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (
         id         INT,
         start_time TIMESTAMP,
         end_time   TIMESTAMP,
         type       INT,
         data       VARCHAR(32),
         etl_source VARCHAR(200),
         etl_id     INT,
         hash_id    INT PRIMARY KEY
     );"""

create_table = VerticaOperator(
    task_id='create_target',
    sql=[create_schema_query,
         create_table_query],
    vertica_conn_id=target_conn_id,
    task_concurrency=1,
    dag=dag)

Za pomocą VerticaOperator() tworzę schemat bazy danych i tabelę (jeśli ich jeszcze nie ma, naturalnie). Ważne jest, aby poprawnie ustawić zależności:

for conn_id, schema in sql_server_ds:
    load = PythonOperator(
        task_id=schema,
        python_callable=workflow,
        op_kwargs={
            'src_conn_id': conn_id,
            'src_schema': schema,
            'dt': '{{ ds }}',
            'target_conn_id': target_conn_id,
            'target_table': f'{target_schema}.{target_table}'},
        dag=dag)

    create_table >> load

Podsumowując

— No właśnie, — powiedział myszak, — prawda, że teraz
Czy upewniłeś się, że w lesie jestem najstraszniejszym zwierzęciem?

Julia Donaldson, „Grufallo”

Myślę, że gdybyśmy z moimi kolegami zorganizowali zawody: kto szybciej stworzy i uruchomi proces ETL od zera: oni ze swoimi SSIS i myszką, a ja z Airflow… A potem porównalibyśmy wygodę wsparcia… Uff, sądzę, że zgodzisz się, że wyprzedziłbym ich na wszystkich frontach!

Jeśli jednak podejdziemy do tego trochę poważniej, to Apache Airflow — dzięki opisowi procesów w formie kodu — ułatwił mi pracę znacznie bardziej wygodną i przyjemną.

Jego nieograniczona rozbudowa: zarówno w zakresie pluginów, jak i predyspozycji do skalowalności — daje ci możliwość stosowania Airflow praktycznie w każdej dziedzinie: zarówno w pełnym cyklu zbierania, przygotowywania i przetwarzania danych, jak i przy uruchamianiu rakiet (na Marsa, oczywiście).

Część końcowa, informacyjno-referencyjna

Pułapki, które zebraliśmy dla was

start_date. Tak, to już lokalny mem. Przez główny argument DAG start_date przechodzą wszyscy. Krótko mówiąc, jeśli podasz w start_date aktualną datę, a w schedule_interval — jeden dzień, to DAG uruchomi się jutro nie wcześniej.
```
start_date = datetime(2020, 7, 7, 0, 1, 2)
```
I więcej żadnych problemów.
Z nim powiązana jest jeszcze jeden błąd wykonania: Task is missing the start_date parameter, co najczęściej oznacza, że zapomniałeś powiązać go z operatorem DAG.
Wszystko na jednej maszynie. Tak, zarówno bazy (samego Airflow i naszej infrastruktury), jak i serwer www, oraz harmonogram i pracownicy. I to nawet działało. Ale z czasem liczba zadań w usługach rosła, i gdy PostgreSQL zaczynał odpowiadać na indeks z 20 ms zamiast 5 ms, zabraliśmy go i przenieśliśmy.
LocalExecutor. Tak, korzystamy z niego do tej pory, i już podeszliśmy do krawędzi przepaści. LocalExecutor’a wciąż nam wystarczało, ale teraz nadszedł czas, aby rozszerzyć się o przynajmniej jednego pracownika, i trzeba będzie się postarać, aby przejść na CeleryExecutor. A ponieważ można z nim pracować także na jednej maszynie, to nic nie powstrzymuje od używania Celery nawet na serwerze, który „naturalnie, nigdy nie trafi do produkcji, przysięgam!”
Nie używanie wbudowanych środków:
- Connections do przechowywania danych logowania usług,
- SLA Misses do reakcji na zadania, które nie zostały wykonane na czas,
- XCom do wymiany metadanych (powiedziałem metadanych!) między zadaniami DAG.danymi!
Nadużywanie poczty. Co tu powiedzieć? Ustawiono powiadomienia dla wszystkich powtórzeń upadłych zadań. Teraz mam w moim roboczym Gmailu >90k wiadomości od Airflow, a webowy interfejs poczty odmawia przyjmowania i usuwania więcej niż 100 jednocześnie.

Więcej pułapek: Pułapki Apache Airflow

Środki do jeszcze większej automatyzacji

Aby jeszcze bardziej pracować głową, a nie rękami, Airflow przygotowało dla nas to:

REST API — wciąż ma status Experimental, co nie przeszkadza mu działać. Umożliwia nie tylko uzyskiwanie informacji o DAG-ach i zadaniach, ale również zatrzymywanie/uruchamianie DAG-ów, tworzenie DAG Run lub pul.
CLI — przez wiersz poleceń dostępne są liczne narzędzia, które są nie tylko niewygodne w użyciu przez WebUI, ale wręcz nie istnieją. Na przykład:
- backfill potrzebny do ponownego uruchamiania instancji zadań.
  Na przykład, przychodzą analitycy i mówią: „A u Was, towarzyszu, jest bałagan w danych od 1 do 13 stycznia! Naprawiaj, naprawiaj, naprawiaj!”. A Ty tak hop:
```
airflow backfill -s '2020-01-01' -e '2020-01-13' orders
```
- Utrzymanie bazy: initdb, resetdb, upgradedb, checkdb.
- run, która pozwala uruchomić jedną instancję zadania, ignorując wszystkie zależności. Co więcej, można ją uruchomić przez LocalExecutor, nawet gdy masz klaster Celery.
- Z grubsza to samo robi test, tylko w bazie nic nie zapisuje.
- connections umożliwia masowe tworzenie połączeń z powłoki.

Python API — dość hardcore'owy sposób interakcji, który jest przeznaczony dla wtyczek, a nie do działających ręcznie. Ale kto nam zabroni pójść do /home/airflow/dags, uruchomić ipython i zacząć szaleć? Można na przykład wyeksportować wszystkie połączenia takim kodem:

from airflow import settings
from airflow.models import Connection

fields = 'conn_id conn_type host port schema login password extra'.split()

session = settings.Session()
for conn in session.query(Connection).order_by(Connection.conn_id):
  d = {field: getattr(conn, field) for field in fields}
  print(conn.conn_id, '=', d)

Połączenie z bazą danych metadanych Airflow. Nie polecam w niej pisać, ale wyciąganie stanów zadań dla różnych specyficznych metryk można znacznie szybciej i łatwiej zrobić niż przez jakikolwiek z API.

Powiedzmy, że daleko nie wszystkie nasze zadania są idempotentne i czasami mogą się nie powieść, co jest w porządku. Ale kilka awarii to już podejrzane i warto to sprawdzić.

Uważaj, SQL!

Z LAST_executions AS (
SELECT
    task_id,
    dag_id,
    execution_date,
    state,
        row_number()
        OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC) AS rn
FROM public.task_instance
WHERE
    execution_date > now() - INTERVAL '2' DAY
),
failed AS (
    SELECT
        task_id,
        dag_id,
        execution_date,
        state,
        CASE WHEN rn = row_number() OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC)
                 THEN TRUE END AS last_fail_seq
    FROM last_executions
    WHERE
        state IN ('failed', 'up_for_retry')
)
SELECT
    task_id,
    dag_id,
    count(last_fail_seq)                       AS unsuccessful,
    count(CASE WHEN last_fail_seq
        AND state = 'failed' THEN 1 END)       AS failed,
    count(CASE WHEN last_fail_seq
        AND state = 'up_for_retry' THEN 1 END) AS up_for_retry
FROM failed
GROUP BY
    task_id,
    dag_id
HAVING
    count(last_fail_seq) > 0

Linki

I oczywiście pierwsze dziesięć linków z wyników Google zawiera zawartość mojego folderu Airflow z zakładkami.

Dokumentacja Apache Airflow — oczywiście trzeba zacząć od oficjalnej dokumentacji, ale kto czyta instrukcje?
Najlepsze praktyki — przynajmniej przeczytaj rekomendacje od twórców.
Interfejs użytkownika Airflow — na samym początku: interfejs użytkownika w obrazkach
Zrozumienie kluczowych koncepcji Apache Airflow — dobrze opisane podstawowe pojęcia, jeśli (nie daj Boże!) nie zrozumiałeś czegoś ode mnie.
Blog Tianlonga - Przewodnik po budowie serwera/klastrowej Airflow — krótki przewodnik po konfiguracji klastra Airflow.
Uruchamianie Apache Airflow w Lyft — prawie tak samo interesujący artykuł, tylko z większą ilością formalizmu i mniejszą ilością przykładów.
Jak Apache Airflow rozdziela zadania na pracowników Celery — o pracy w parze z Celery.
Najlepsze praktyki pisania DAG-ów w Apache Airflow — o idempotentności zadań, ładowaniu według ID zamiast daty, transformacjach, strukturze plików i innych interesujących rzeczach.
Zarządzanie zależnościami w Apache Airflow — zależności zadań i zasada wyzwalania, o których wspomniałem tylko w przelotie.
Airflow: Kiedy Twój DAG jest daleko za harmonogramem — jak pokonywać niektóre problemy z harmonogramem i priorytetować zadania oraz ładować utracone dane.
Przydatne zapytania SQL dla Apache Airflow — przydatne zapytania SQL do metadanych Airflow.
Rozpocznij rozwijanie przepływów pracy z Apache Airflow — jest przydatna sekcja o tworzeniu własnego sensora.
Budowanie infrastruktury Data Science Fetchr na AWS z Presto i Airflow — interesująca krótka notatka o budowie infrastruktury na AWS dla Data Science.
7 powszechnych błędów do sprawdzenia podczas debugowania DAG-ów Airflow — powszechne błędy (gdy ktoś nie czyta instrukcji).
Przechowywanie i dostęp do hasła za pomocą Apache Airflow — uśmiechnij się, jak ludzie obchodzą problem przechowywania haseł, chociaż można po prostu użyć Connections.
Zen Pythona i Apache Airflow — niejawne przeprowadzenie DAG, przekazywanie kontekstu w funkcjach, ponownie o zależnościach, a także o pomijaniu uruchomień zadań.
Airflow: Mniej znane wskazówki, triki i najlepsze praktyki — o wykorzystaniu argumentów domyślnych i parametry w szablonach oraz o Zmiennych i Połączeniach.
Profilowanie planisty Airflow — opowieść o tym, jak planista jest przygotowywany do Airflow 2.0.
Apache Airflow z 3 pracownikami Celery w docker-compose — trochę przestarzały artykuł o wdrożeniu naszego klastra w docker-compose.
4 Zadania szablonowe z użyciem kontekstu Airflow — dynamiczne zadania za pomocą szablonów i przekazywania kontekstu.
Powiadomienia o błędach w Airflow — standardowe i niestandardowe powiadomienia e-mail i Slack.
Warsztaty Airflow: skomplikowane DAGi bez obejść — Rozgałęzienia zadań, makra i XCom.