Kako uporabljati Airflow: Praktičen vodnik od začetka do konca za izgradnjo zanesljivih podatkovnih cevovodov
Če premikate podatke ali orkestrirate opravila strojnega učenja (ML), ste verjetno že slišali isti refren: »Samo daj to v Airflow.« Resnica je, da Apache Airflow blesti, ko potrebujete preglednost, zanesljivost in nadzor nad kompleksnimi poteki dela. V tem praktičnem vodniku vas bomo korak za korakom vodili skozi uporabo Airflow – od temeljnih konceptov do vzorcev, pripravljenih za produkcijo – da boste lahko ustvarili cevovode, ki jim zaupate.
Ohranili bomo praktičnost: dobili boste mentalni model za DAG-e in opravila, praktične primere z API-jem TaskFlow, možnosti uvajanja, strategije testiranja in najboljše prakse. Na koncu boste prešli od »Lahko izvedem vadnico« do »To lahko izvajam v produkciji.«
Opomba: Za podrobnejše raziskave in reference je uradna dokumentacija odlična in se redno posodablja.
Kaj Apache Airflow v resnici je?
Airflow je orkestrator – ne obdelovalec podatkov. Načrtuje, ureja in nadzoruje delo, ki ga izvajate drugje (baze podatkov, skladišča, Spark opravila, API-ji, vsebnikov). Poteke dela definirate kot DAG-e (usmerjeni aciklični grafi), ki so samo datoteke Python, ki kodirajo opravila in njihove odvisnosti. Airflow nato izvede ta opravila v skladu z vašim urnikom, parametri in okoljem.
- DAG: Definicija poteka dela (graf opravil z odvisnostmi).
- Opravilo: Enota dela (funkcija Python, izvedba SQL, ukaz Bash, sprožilec zunanjega opravila itd.).
- Operator: Predloga za vrsto opravila (npr.
PythonOperator, BashOperator, KubernetesPodOperator).
- Razporejevalnik (Scheduler): Odloča, kaj se bo izvajalo in kdaj.
- Izvajalec (Executor): Izvaja opravila (lokalno, s Celery, Kubernetes itd.).
- UI: Vaš nadzorni center za izvajanja, dnevnike, ponovne poskuse in rodovnik.
Začnite z uradnimi vadnicami, ko namestite Airflow; hitro vam dajo celotno sliko.
Namestitev in zagon Airflow na pravi način
Airflow je prilagodljiv. Izberite pot, ki ustreza vaši fazi:
- Lokalni razvoj (hiter začetek):
- Uporabite Docker Compose za hiter začetek, ki ga zagotavlja projekt. Zažene spletni strežnik, razporejevalnik, bazo podatkov in drugo s smiselnimi privzetimi nastavitvami.
- Odlično za učenje in ponavljanje DAG-ov.
- Majhna ekipa ali testiranje:
- Celery Executor ali Kubernetes Executor z upravljanim Postgresom.
- Shranjujte dnevnike v S3/GCS in zapakirajte odvisnosti s svojo sliko ali
requirements.txt.
- Kubernetes Executor za elastičnost ali Celery Executor z avtomatskim skaliranjem delavcev.
- Zunanje skrivnosti (Vault), robustna opazovalnost (dnevniki + metrike) in modro/zelene uvedbe za nadgradnje.
Nasvet: Hranite svojo Airflow izvorno kodo pod nadzorom različic, v vsebnikih in testirano pred promocijo. Stran »Najboljše prakse« opisuje vzorce, pripravljene za produkcijo.
Temeljni koncepti, ki jih boste uporabljali vsak dan
DAG-i: Vaš potek dela kot koda
DAG je datoteka Python, ki definira:
- Metapodatki DAG: id, urnik, datum začetka, oznake.
- Privzeti argumenti: ponovni poskusi, lastniki, SLA-ji.
- Opravila in njihove odvisnosti.
Mislite na DAG kot na »kaj« in »kdaj«, opravila pa kot na »kako«.
Opravila in operatorji
Operatorji so montažne enote za pogosta opravila. Primeri:
- PythonOperator / TaskFlow
@task za kodo Python
- BashOperator za ukaze lupine
- SimpleHttpOperator za API-je
- KubernetesPodOperator za vsebniška opravila
- Ponudniki SQL (npr. Snowflake, BigQuery, Postgres) za delo v skladišču
TaskFlow API: Sodoben, Pythonski način
API TaskFlow vam omogoča pisanje opravil kot funkcij Python z @task, vračanje vrednosti, ki se posredujejo prek XCom, in njihovo čisto sestavljanje. Zmanjšuje odvečno kodo in izboljšuje berljivost – zelo priporočljivo.
Vaš prvi Airflow DAG (izdaja TaskFlow)
Spodaj je minimalen primer v slogu ETL za ponazoritev ključnih idej: načrtovanje, TaskFlow, odvisnosti in posredovanje podatkov XCom.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
with DAG(
dag_id="weather_etl_example",
start_date=datetime(2024, 1, 1),
schedule="0 * * * *", # hourly
catchup=False,
tags=.
Načrtovanje, dohitevanje (Catchup) in zapolnitve (Backfills)
schedule: Cron ali prednastavitve (@daily, @hourly).
start_date + catchup: Če je catchup=True, bo Airflow zapolnil izvajanja od datuma začetka. Za cevovode v slogu pretakanja nastavite catchup=False.
- Ročne zapolnitve: Uporabite UI ali CLI za ponovno izvajanje zgodovinskih intervalov.
Praktično pravilo: omogočite dohitevanje za deterministična paketna opravila; onemogočite za cevovode v realnem času ali API z omejeno hitrostjo.
Varno posredovanje podatkov med opravili (XCom)
- Majhni predmeti: vračanje vrednosti s TaskFlow je v redu.
- Velika izhodna sporočila: shranite v shrambo predmetov (S3/GCS) s ključem v XCom.
- Izogibajte se občutljivim podatkom v XCom; uporabite zaledne sisteme za skrivnosti (npr. Vault) in spremenljivke okolja.
Dinamično preslikavo opravil in razvejanih delovnih obremenitev
Airflow lahko dinamično ustvarja opravila med izvajanjem na podlagi vhodnih podatkov – idealno za particionirane nabor podatkov ali večnajemniška opravila.
- Ohranite DAG-e deterministične in idempotentne.
- Ločite orkestracijo (Airflow) od računanja (Spark, dbt, skladišča).
- Uporabite API TaskFlow za jasnost in higieno XCom.
- Parametrizirajte DAG-e; uporabljajte spremenljivke preudarno.
- Spremljajte, opozarjajte in dokumentirajte svoje cevovode.
Kako delati s podatkovnimi skladišči in ML
- Podatkovna skladišča: Uporabite operatorske programe ponudnika (npr. SnowflakeOperator, BigQueryInsertJobOperator) za opravila SQL. Shranite SQL v datotekah ali modulih z različicami.
- dbt: Sprožite dbt prek Bash/KubernetesPodOperator ali namenskih dbt operatorjev pri ponudnikih.
- ML: Orkestrirajte generiranje funkcij, usposabljanje in paketno sklepanje kot ločena opravila; predpomnite artefakte v shrambi in beležite metrike.
Napredno načrtovanje: Nabori podatkov in odvisnosti med DAG-i
- Nabori podatkov omogočajo, da en DAG ustvari logični nabor podatkov, ki sproži drug DAG, ko je posodobljen – čisteje kot ad-hoc sprožilci.
- Za stare vzorce ExternalTaskSensor deluje, vendar so nabori podatkov bolj deklarativni.
Varnost in skladnost
- Uporabite nadzor dostopa na podlagi vlog (RBAC) v UI.
- Izolirajte okolja na ekipo ali mejo zaupanja.
- Hranite revizijske sledi prek dnevnikov in zgodovine sprememb povezave.
Nadgradnje in različice
- Testirajte nadgradnje v testnem okolju s produkcijskimi delovnimi obremenitvami.
- Pripnite in nadgradite ponudnike premišljeno.
- Preberite opombe ob izdaji za spremembe in opustitve, specifične za izvajalca.
Hiter kontrolni seznam za vaš prvi produkcijski DAG
- Jasno lastništvo (oznaka
owner) in konfigurirana opozorila.
retries nastavljen z razumnimi odmiki.
- Idempotentna opravila in eksplicitne odvisnosti.
- Majhne obremenitve XCom; veliki podatki v shrambi.
- Dnevniki, poslani v trajno shrambo; izvožene metrike.
- Načrt uvajanja (kanarski ali modro/zeleni) in koraki za povrnitev.
Primer: Realističen DAG za nalaganje skladišča
Ta vzorec izvleče dnevne datoteke, jih preveri in naloži v tabelo skladišča, z dinamičnim preslikavanjem na particijo in preložljivimi senzorji.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.sensors.filesystem import FileSensor
- Pred promocijo v produkcijo preglejte najboljše prakse.
- Raziščite dokumente ponudnika za vaše sisteme (skladišča, oblaki, orodja ML).
Mimogrede: Pospešite avtorstvo z AI pomočnikom
Omeniti velja: če pripravljate veliko DAG-ov, lahko pomočnik AI, ki razume kodo, pospeši odvečno kodo, ustvari ogrodja TaskFlow in celo predlaga popravke odvisnosti. Če želite lahkega pomočnika poleg urejevalnika in brskalnika, je Sider.AI lahko priročen za hitro prepisovanje kode in razlage med razvojem. Ključni zaključki
- Uporabite Airflow za orkestracijo, ne za računanje.
- Za čiste, preizkusljive DAG-e raje uporabite API TaskFlow.
- Hranite podatke zunaj XCom; namesto tega posredujte reference.
- Uporabite preložljive senzorje/operatorje za shranjevanje rež.
- Vsebnike, testirajte in promovirajte skozi okolja.
- Zanašajte se na uradne vadnice in najboljše prakse kot svojo severnico.
Pogosta vprašanja
Q1:Kateri je najlažji način za učenje uporabe Airflow?
Začnite z uradno vadnico, da razumete DAG-e, opravila, načrtovanje in UI. Nato zgradite majhen cevovod, ki temelji na TaskFlow, in ponavljajte z vodnikom po najboljših praksah za pripravljenost na produkcijo.
Q2:Ali naj uporabim API TaskFlow ali klasične operatorje v Airflow?
Uporabite API TaskFlow za večino Pythonskih cevovodov, ker je čistejši in naravno obravnava vračila XCom. Klasični operatorji so še vedno odlični za ne-Pythonska opravila, kot so Bash, SQL ali opravila v vsebnikih.
Q3:Kako posredujem velike količine podatkov med opravili Airflow?
Izogibajte se vstavljanju velikih obremenitev v XCom. Shranite podatke v S3/GCS ali bazo podatkov in posredujte samo reference ali URI-je prek XCom, da bodo opravila hitra in zanesljiva.
Q4:Katerega izvajalca naj izberem za Airflow v produkciji?
Za elastičnost in izolacijo je Kubernetes Executor močna privzeta nastavitev. Za enostavnejše nastavitve Celery Executor dobro deluje – samo zagotovite samodejno skaliranje, robustno beleženje in eksternalizirane skrivnosti.
Q5:Kako obravnavam odvisnosti med več DAG-i Airflow?
Uporabite nabore podatkov za deklarativne sprožilce med DAG-i, ko en cevovod ustvari podatke za drugega. Lahko pa ExternalTaskSensor koordinira izvajanja, vendar so nabori podatkov čistejši za orkestracijo, ki temelji na podatkih.
