Apache Airflow Explicat: Un Ghid Prietenos Pentru Începători

Introducere

Companiile moderne bazate pe date se bazează toate pe fluxuri de date care preiau, transformă, îmbogățesc și mută informațiile dintr-o locație în alta. Aceste fluxuri de date pot implica mai mulți pași (extracția datelor brute, curățarea lor, antrenarea modelelor de învățare automată, construirea de tablouri de bord, etc.). În plus, ele trebuie să ruleze într-o ordine foarte specifică. Instrumentele de orchestrare a fluxurilor de lucru, cum ar fi Apache Airflow, ajută la asigurarea faptului că toți acești pași rulează la momentul potrivit și în ordinea corectă. De asemenea, ele fac ca fluxurile de date să fie ușor de monitorizat și de gestionat.

Airflow a fost creat la Airbnb în 2014 și a devenit una dintre cele mai populare platforme open‑source de orchestrare a fluxurilor de lucru. Este utilizat de mii de organizații pentru a automatiza și monitoriza pipeline-uri batch. În acest articol, explicăm Apache Airflow de la bază. Subiectele pe care le vom acoperi includ:

• Ce este Airflow, ce face și de ce îl folosesc echipele de date.
• Conceptele și componentele de bază ale Airflow și modul în care se îmbină.
• Cum sunt definite conductele ca grafuri aciclice orientate (DAG-uri) și o prezentare detaliată a unui DAG simplu în acțiune.
• Cazuri de utilizare comune, cum ar fi ETL/ELT, MLOps și altele.

Sperăm ca aceasta să fie o introducere practică, bazată pe exemple, în Apache Airflow pentru ingineri, studenți și echipe care încep să lucreze cu orchestrarea fluxurilor de lucru.

Principalele concluzii

• Apache Airflow este o platformă de orchestrare a fluxurilor de lucru bazată pe Python care gestionează, programează și monitorizează pipeline-uri de date complexe folosind DAG-uri (Grafice Aciete Direcționate).
• DAG-urile definesc sarcini și dependențele acestora, permițând Airflow să execute fiecare pas în ordinea corectă, cu repetări automate, jurnalizare și observabilitate încorporate.
• Arhitectura Airflow (Scheduler, Executor, Worker, baza de date Metadata, interfața web) permite execuția scalabilă și distribuită și monitorizarea în timp real a fluxurilor de lucru.
• Airflow se descurcă excelent în fluxurile de lucru batch, cum ar fi joburile ETL/ELT, pipeline-urile de antrenament ML, verificările calității datelor și analiza programată — dar nu este ideal pentru sarcini în timp real sau cu frecvență ultra-înaltă.
• Alternative precum Prefect, Dagster, Luigi, Argo Workflows și Mage AI pot fi mai potrivite în funcție de nevoi precum fluxuri de lucru bazate pe evenimente, execuție nativă pe Kubernetes sau interfețe moderne prietenoase pentru dezvoltatori.

Ce este Apache Airflow?

Apache Airflow este un sistem open-source pentru crearea, programarea și monitorizarea fluxurilor de lucru. Un flux de lucru în Airflow este un DAG (graf aciclic orientat), care este o serie de sarcini cu relații clar definite (dependențe) și fără cicluri (o sarcină nu se poate repeta în amonte). Filosofia de bază a Airflow este „configurare ca și cod”, așa că în loc să folosești o interfață vizuală cu drag-and-drop, creezi scripturi Python pentru a reprezenta fluxurile de lucru. Această abordare oferă o flexibilitate semnificativă pentru ingineri: poți să te conectezi la aproape orice tehnologie prin operatorii Airflow și bibliotecile Python. De asemenea, poți aplica cele mai bune practici din ingineria software (controlul versiunii, testare etc.) la definițiile pipeline-ului tău.

De ce contează orchestrarea fluxului de lucru?

Punctul cheie este că, adesea, ai mai mulți pași într-un flux de lucru de date care trebuie executați în secvență. Un exemplu simplu este pipeline-ul ETL (Extract/Transform/Load). Pipeline-urile pot deveni rapid mai complexe pe măsură ce adaugi mai multe sarcini, dependențe, ramificații etc., și trebuie să ruleze și într-un ritm regulat. Intervenția manuală sau uneltele simple de programare devin adesea inadecvate în aceste cazuri. Airflow rezolvă acest lucru acționând ca un orchestrator central. Are un graf aciclic orientat (DAG) de sarcini (noduri) cu dependențe (muchii). Știe când să execute fiecare sarcină (asteptând ca dependențele anterioare să se finalizeze înainte de a rula), inclusiv programarea (execuție la intervale periodice), gestionarea erorilor (reîncercări automate, alerte) și înregistrarea logurilor.

Dacă considerăm întregul flux de date ca fiind o rețetă, atunci fiecare sarcină individuală este un subpas (tocare, fierbere, aranjare pe farfurie etc.). Airflow este ca managerul de bucătărie care cunoaște rețetele și supraveghează toate sub-pasurile care au loc în fiecare bucătărie, asigurându-se că totul se întâmplă în ordinea corectă și la momentul potrivit.

Fără orchestrarea fluxului de lucru, ai putea avea toți bucătarii și ingredientele, dar acestea nu sunt coordonate. Airflow oferă această coordonare, permițând producerea „felului de mâncare” dorit (execuția fluxului de lucru).

De ce folosesc echipele de date Airflow?

Airflow a devenit popular printre inginerii de date, inginerii ML și echipele DevOps deoarece răspunde multor nevoi ale pipeline-urilor de date complexe. Iată câteva motive importante pentru care echipele aleg Airflow:

• Orchestrarea fluxurilor de lucru complexe și a dependențelor: Airflow a fost conceput pentru fluxuri de lucru cu un număr mare de pași interdependenți. Airflow face ușoară declararea cerințelor pentru sarcinile anterioare și ulterioare, asigurând că sarcinile se execută în ordinea corespunzătoare. Dacă sarcina C depinde de sarcinile A și B, programatorul Airflow este conștient de aceste relații și se va asigura că A și B rulează (posibil în paralel) înainte de a rula C. Airflow poate gestiona dependențe complexe, logică de ramificare etc.
• Programare centralizată (Cron cu superputeri): Oferă un programator care vă permite să executați sarcini conform unui anumit program. Ați putea defini zeci sau sute de fluxuri de lucru care declanșează sarcini la momente diferite (pe oră, zilnic, lunar) folosind programatorul Airflow. Airflow înțelege relațiile dintre sarcini. Poate programa și declanșa inteligent sarcinile pentru a fi rulate. Programatorul central vă eliberează de nevoia de a scrie scripturi diferite pe servere diferite.

• Scalabilitate și Execuție Distribuită: Airflow poate să se scaleze odată cu fluxurile tale de lucru. Poate fi configurat pentru a executa sarcini în paralel pe multiple procese de lucru sau mașini. De asemenea, oferă mai mulți executori (discuți mai târziu în acest articol) care pot rula sarcini pe sisteme distribuite precum Celery sau Kubernetes pe un cluster de noduri. Aceasta înseamnă că poți paraleliza sarcini grele (procesarea de date mari) și poți avea mai multe pipeline-uri care rulează simultan.
• Monitorizare, înregistrare și alertare: Vine cu o interfață web bogată care oferă vizibilitate în timp real asupra fluxurilor de lucru. Poți vedea care sarcini au avut succes și care au eșuat și poți vizualiza ieșirea din jurnal pentru fiecare sarcină. De asemenea, poți declanșa manual și/sau reîncerca sarcini prin intermediul interfeței web.

Componente cheie ale fluxului de aer

Mai jos este o prezentare succintă a principalelor componente ale Apache Airflow. Folosește acest tabel ca referință rapidă pentru a înțelege ce este fiecare componentă și ce funcție îndeplinește.

Cum Funcționează Airflow (Arhitectură)

Iată o prezentare pas cu pas a modului în care Airflow procesează un flux de lucru (DAG):

• Crearea DAG-ului: Autorul DAG-ului scrie un script Python care definește un DAG. Acest script include aspecte precum programul (de exemplu, „rulează zilnic la ora 9”) sau setările de declanșare manuală. De asemenea, conține definițiile sarcinilor (prin intermediul operatorilor sau funcțiilor @task), precum și dependențele dintre sarcini (de exemplu, task1 >> task2 (task2 așteaptă ca task1 să se termine)). Fișierul Python este salvat în folderul DAGs al Airflow pentru a fi detectat de Airflow. Airflow va citi (analiza) acest fișier periodic pentru a încărca definiția DAG-ului.
• Programarea și analizarea DAG-urilor: Airflow Scheduler este un proces care rulează continuu în fundal. Acesta verifică dacă fiecare DAG trebuie să fie executat în funcție de programul său sau de declanșatoare. De exemplu, un DAG cu un program „zilnic” poate fi preluat de scheduler, care observă că o nouă execuție trebuie programată la miezul nopții. Airflow Scheduler verifică, de asemenea, dacă sarcinile din cadrul DAG-urilor sunt pregătite pentru a rula – ceea ce înseamnă că toate dependențele lor anterioare sunt complete și momentul programat pentru executare (data logică) a sosit.
• Executarea sarcinilor: Fiecare sarcină care trebuie executată este predată Executorului de către Programator. Executorul decide unde și cum să ruleze sarcina. Aceasta poate fi executată local sau trimisă într-o coadă de lucru. Pentru o arhitectură distribuită, sarcina va fi adăugată într-o coadă și apoi selectată de procesele Worker care execută sarcinile.
• Actualizarea stării și gestionarea erorilor: Fiecare instanță de sarcină raportează starea sa (success, failure etc.) înapoi la scheduler pe măsură ce se finalizează, iar scheduler-ul își actualizează baza de date cu metadatele legate de starea sarcinii.
  Urmărirea stării este ceea ce permite scheduler-ului să știe ce să facă în continuare – de exemplu, dacă Task A s-a finalizat cu succes, știe că poate programa acum Task B, sau dacă Task A a eșuat și mai sunt încercări disponibile, va programa o reîncercare, sau dacă a eșuat fără încercări rămase, marchează execuția DAG ca eșuată. Dacă o sarcină eșuează, Airflow are logica de a reîncerca automat sarcina dacă este configurat pentru reîncercări. Dacă eșuează în continuare după reîncercări, Airflow poate genera alerte privind eșecurile (trimitere e-mail sau mesaj pe Slack etc.). Toate acestea sunt configurabile ca parte a parametrilor DAG sau ai sarcinii (de exemplu, numărul de reîncercări, întârzierea între reîncercări, e-mail de notificare la alertă).
• Monitorizare și intervenție: Airflow Web UI oferă o interfață live a procesului de mai sus. Poți vedea rulările unui DAG (de exemplu, o nouă linie în vizualizarea Tree pentru o rulare de astăzi) și poți urmări cum se schimbă stările task-urilor pe măsură ce progresează. Vizualizarea Graph ilustrează task-urile DAG-ului și arată care au fost finalizate sau au întâmpinat erori.
• Curățare și Următoarea Execuție: Când o execuție DAG este finalizată (toate sarcinile au avut succes sau au fost sărite/eșuate), Airflow trece la următoarea execuție programată. Toate metadatele despre execuțiile anterioare sunt păstrate în baza de date, iar istoricul poate fi verificat oricând. Scheduler-ul va continua să declanșeze noi execuții conform programului sau prin declanșarea manuală. Acest proces se repetă până când procesul Airflow este oprit.

Înțelegerea DAG-urilor în Airflow

Un DAG în Airflow definește programul, precum și sarcinile și dependențele necesare pentru a executa un flux de lucru. Un DAG nu trebuie să știe ce se află în fiecare sarcină; trebuie doar să definească când și în ce ordine ar trebui să ruleze acestea.

Sarcini, operatori și senzori

Sarcinile sunt unitățile de bază de lucru în Airflow. În Airflow, sarcinile sunt instanțe ale operatorilor, care sunt șabloane pentru acțiuni predefinite. Airflow vine cu o varietate de operatori de bază. BashOperator îți permite să rulezi comenzi în shell, iar PythonOperator îți permite să apelezi funcții Python. Airflow oferă, de asemenea, un decorator @task, care îți permite să transformi o funcție Python obișnuită într-o sarcină.

Pe lângă operatori, Airflow oferă și senzori. Senzorii sunt sarcini care așteaptă ca o condiție sau un eveniment să fie îndeplinit înainte de a avea succes. Aceștia pot fi folosiți pentru lucruri precum așteptarea apariției unui fișier sau popularea unui tabel.

Dependențele sarcinilor

Pentru a declara dependențe între sarcini, utilizați operatorii >> și << (recomandați) sau metode precum set_upstream și set_downstream. De asemenea, este posibil să lanțuiți sarcinile, să creați dependențe cross‑downstream și chiar să construiți dinamic liste de sarcini. Un DAG fără nicio dependență ar fi doar un set independent de sarcini.

Cele mai bune practici pentru proiectarea DAG

Tabelul de mai jos prezintă câteva linii directoare de avut în vedere atunci când proiectați DAG-uri.

Exemplu: Un DAG Airflow simplu

Nimic nu consolidează conceptele mai bine decât un exemplu. Să parcurgem un cod simplu de DAG în Airflow și să explicăm părțile sale. Acest exemplu va crea un mic flux de lucru cu două sarcini în succesiune:

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

# Define the DAG and its schedule
with DAG(
    dag_id="example_dag",
    description="A simple example DAG",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",  # runs daily
    catchup=False
) as dag:
    # Task 1: Print current date
    task1 = BashOperator(
        task_id="print_date",
        bash_command="date"
    )
    # Task 2: Echo a message
    task2 = BashOperator(
        task_id="echo_hello",
        bash_command="echo 'Hello, Airflow!'"
    )
    # Define dependency: task1 must run before task2
    task1 >> task2

Să descompunem ceea ce se întâmplă aici:

• importați clasele necesare DAG (pentru a defini DAG-ul) și BashOperator (pentru a rula comenzi shell/bash în sarcini). De asemenea, importăm datetime pentru a specifica data de început a DAG-ului.
• Creăm un obiect DAG folosind un manager de context (with DAG(…) as dag:). Furnizați DAG-ului un ID unic „example_dag” (trebuie să fie unic într-un mediu Airflow), o descriere (opțional, pentru a ajuta oamenii să înțeleagă), un start_date și un schedule_interval. Aici, setăm un schedule_interval preset (@daily) încorporat și start_date=datetime(2025,1,1). De asemenea, am setat catchup=False pentru a preveni completarea retroactivă a oricăror execuții lipsă între start_date și ziua de astăzi (dorim să ruleze doar de acum înainte, zilnic). Acest bloc întreg creează un DAG despre care Airflow va fi informat să ruleze zilnic la începutul zilei.
• Apoi definim două sarcini în contextul DAG:
• task1 este definit folosind BashOperator pentru a executa o comandă bash (date). task_id este „print_date” și va fi folosit pentru a identifica această sarcină în interfața UI și în jurnale. Când rulează, această sarcină va afișa pur și simplu data/ora curentă (în jurnalul sarcinii).
• task2 de asemenea folosește un BashOperator cu task_id=“echo_hello”, și command=echo ‘Hello, Airflow!’. Acesta va afișa un mesaj frumos. Aceste sarcini sunt momentan doar definite, încă nu sunt conectate.
• Ultima linie (task1 >> task2) specifică o dependență (flux): task1 se revarsă în task2. Acest lucru înseamnă că task1 va rula și se va finaliza cu succes înainte ca task2 să poată fi luat în considerare pentru programare. Grafic, în graful DAG, va exista o săgeată care indică de la task1 la task2.

Câteva lucruri de reținut în acest exemplu:

• Am folosit doar BashOperator pentru simplitate, dar am fi putut folosi un DAG cu tipuri mixte de operatori. Am putea avea un PythonOperator pentru a apela o funcție Python, sau un SimpleHttpOperator pentru a contacta un API, și așa mai departe.
• Structura with DAG(…) as dag: este un idiom comun pentru crearea unei instanțe DAG. Toate sarcinile care sunt create în interiorul ei sunt automat asociate cu acel DAG.
• catchup=False este un parametru utilizat în multe DAG-uri cu o start_date în trecut, atunci când nu vrei ca acesta să ruleze pentru toate datele anterioare. Dacă am lăsa catchup=True (implicit) și astăzi ar fi 10 ianuarie 2025, Airflow ar încerca să ruleze DAG-ul pentru 1, 2, 3, … până pe 10 ianuarie, recuperând acele zile ratate din urmă. L-am dezactivat aici pentru simplitate.
• Acest exemplu folosește calea de import relativă pentru BashOperator. În Airflow 2, mulți dintre operatori au fost mutați sub airflow.providers. Acest exemplu este scris cu presupunerea că acei provideri sunt instalați (BashOperator se află în nucleul Airflow).

Acest exemplu simplu poate fi extins cu ușurință – ai putea adăuga sarcini suplimentare (poate să faci un apel API, apoi să încarci rezultatele într-o bază de date, apoi să trimiți o notificare). De asemenea, poți adăuga o logică mai complexă (cum ar fi ramificarea – Airflow suportă ramificarea de tip if/else folosind, de exemplu, BranchOperator).

Pe măsură ce DAG-urile tale devin mai mari, interfața UI și jurnalizarea din Airflow devin de neprețuit pentru a înțelege ce se întâmplă. În acest exemplu, jurnalul pentru print_date ar conține data sistemului care a fost tipărită, iar jurnalul pentru echo_hello ar conține „Hello, Airflow!”. Dacă print_date ar eșua (poate că nu ar putea găsi comanda date, într-un scenariu ipotetic), atunci echo_hello nu ar fi niciodată executat (datorită dependenței), iar rularea DAG-ului ar fi marcată ca eșuată. Ai putea apoi să inspectezi jurnalul, să remediezi problema și să rulezi din nou acea sarcină sau întregul DAG.

Cazuri de utilizare comune pentru Airflow

Mai jos este un rezumat structurat al cazurilor comune de utilizare a Apache Airflow. Tabelul evidențiază ce implică fiecare caz de utilizare și cum adaugă Airflow valoare în acel context.

Când să NU folosești Airflow

Deși Airflow este puternic, nu este instrumentul potrivit pentru orice situație. Înțelegerea limitărilor sale te ajută să alegi abordarea corectă de orchestrare:

• Sarcini în timp real sau de tip streaming: Airflow este orientat pe loturi, cu un început și un sfârșit bine definit. Nu este potrivit pentru sarcini de lungă durată, declanșate de evenimente sau de tip streaming. Dacă colectați fluxuri continue de evenimente (clicuri ale utilizatorilor, senzori IoT etc.) și trebuie să le procesați cu latență foarte scăzută în timp real, Airflow nu este cea mai bună alegere.
• Sarcini cu frecvență mare sau multe sarcini scurte: Acesta este un alt caz în care Airflow poate să nu funcționeze sau cel puțin nu eficient. Sarcinile Airflow au un anumit overhead (în urmărirea în baza de date etc.), așa că dacă trebuie să declanșezi ceva la câteva secunde sau ai mii de sarcini mici, Airflow s-ar putea să nu se scaleze bine.
• Fluxuri de lucru pur bazate pe evenimente: Acesta este similar cu cazul de utilizare streaming, dar pentru ceva care ar trebui să fie declanșat de un eveniment. „Ori de câte ori se întâmplă un eveniment X, fă Y” este un caz de utilizare perfect valid pentru Airflow, dar dacă singurul lucru care ar trebui să declanșeze acest lucru este un eveniment (de exemplu, „ori de câte ori ajunge un fișier într-un bucket S3, fă X”), atunci Airflow nu este cea mai ușoară soluție aici. Airflow utilizează în general senzori (de exemplu, S3KeySensor) pentru a „simți” prezența fișierelor/evenimentelor. Senzorii au dezavantaje deoarece se bazează pe interogare periodică (polling). Problemele cu senzorii includ declanșarea pe fișierele existente și faptul că nu sunt neapărat responsivi la evenimente (de exemplu, fac polling la intervale).

Pe scurt, Airflow este potrivit pentru fluxurile de lucru care sunt periodice, orientate pe loturi sau complexe de definit și gestionat. S-ar putea să nu fie cea mai bună alegere dacă cazul tău de utilizare necesită procesare în timp real sau continuă, sarcini extrem de frecvente sau fluxuri de lucru foarte simple care nu necesită suprasarcina Airflow. În acest caz, s-ar putea să vrei să explorezi alte opțiuni sau să simplifici soluția ta.

Alternative la Airflow

Spațiul de orchestrare a fluxurilor de lucru a explodat în ultimii ani, și, deși Airflow este un lider, există câteva unelte alternative care merită cunoscute:

Airflow nu este singura opțiune disponibilă, iar cea mai bună alegere depinde de cazul tău de utilizare. Luigi, Prefect și Dagster sunt frecvent menționate ca principalele alte opțiuni open-source din același domeniu (orchestratori de fluxuri de lucru bazate pe Python). Dacă UI-ul învechit al Airflow sau alte limitări ți se par incomode, ar trebui să iei în considerare evaluarea acestora. Prefect încearcă să fie un Airflow „mai bun” sau „mai simplu”, Dagster încearcă să fie un orchestrator mai „structurat” cu accent pe activele de date, iar Luigi este un precursor mai simplu al Airflow. Pe de altă parte, dacă cazul tău de utilizare diferă semnificativ (streaming în timp real sau complet nativ cloud, de exemplu), este posibil să nu folosești deloc Airflow și să iei direct în considerare platforme de streaming sau servicii de orchestrare gestionate.

Gânduri Finale

Apache Airflow este o platformă open-source pentru orchestrarea fluxurilor complexe de lucru computaționale și a pipeline-urilor de procesare a datelor. Inițial dezvoltat la Airbnb, a câștigat o popularitate semnificativă în comunitățile de inginerie a datelor și în machine learning. Componentele Airflow includ un programator (scheduler), un executor, lucrători (workers), o bază de date pentru metadate și o interfață web. Acest lucru permite echipelor să ruleze și să întrețină ușor pipeline-uri ETL, MLOps și de infrastructură gata de producție. Abstracția DAG (Directed Acyclic Graph) a Airflow permite păstrarea fluxurilor de lucru explicite, testabile și ușor de întreținut în timp. Ecosistemul platformei, format din operatori și furnizori creați de comunitate, simplifică de asemenea orchestrarea fluxurilor de lucru care depind de orice tehnologie sau instrument.

Totuși, nu este o soluție universală. Funcționează cel mai bine cu fluxuri de lucru pe loturi sau micro-loturi, iar utilizatorii trebuie să fie confortabili cu Python. Pentru fluxuri de lucru în timp real sau cu frecvență ridicată, există alte soluții mai potrivite decât Airflow. Pentru oamenii de știință de date sau echipele care preferă să scrie fluxuri de lucru declarative, ar putea merita să ia în considerare instrumente alternative Airflow.

Airflow este încă activ dezvoltat, iar noi funcționalități sunt adăugate în timp, cum ar fi programarea bazată pe evenimente, seturile de date și resursele. Conceptele și cele mai bune practici prezentate în acest ghid ar trebui să te ajute, indiferent de experiența ta, să înțelegi cum funcționează orchestrarea fluxurilor de lucru și când ar trebui folosit Airflow (sau nu). Sperăm că acest ghid va servi drept un punct de plecare pentru a explora mai departe acest univers complex.

Referințe

• Apache Airflow
• Ce este Airflow?
• Apache Airflow: Cazuri de utilizare, arhitectură și 6 sfaturi pentru succes
• Introducere pentru începători în Airflow cu Docker — Pipeline-uri de date în cloud (I)
• 8 alternative pentru Airflow - Cazuri de utilizare
• Prezentare generală a arhitecturii
• Programator

Mulțumim că ați învățat împreună cu comunitatea Linux-Console.net.