Com utilitzar Airflow: Una guia pràctica i integral per construir pipelines de dades fiables
Si transfereixes dades o coordines treballs de ML, probablement has sentit el mateix refrany: "Simplement posa-ho a Airflow." La veritat és que Apache Airflow destaca quan necessites visibilitat, fiabilitat i control sobre fluxos de treball complexos. En aquesta guia pràctica, et guiarem pas a pas sobre com utilitzar Airflow, des dels conceptes bàsics fins als patrons preparats per a la producció, perquè puguis lliurar pipelines en els quals confies.
Ho mantindrem accionable: obtindràs un model mental per a DAGs i tasques, exemples pràctics amb l'API TaskFlow, opcions de desplegament, estratègies de prova i bones pràctiques. Al final, passaràs de "Puc executar el tutorial" a "Puc executar això en producció".
Nota: Per a immersions i referències més profundes, la documentació oficial és excel·lent i s'actualitza regularment.
Què és realment Apache Airflow?
Airflow és un orquestrador, no un processador de dades. Programa, ordena i supervisa el treball que executes en altres llocs (bases de dades, magatzems, treballs de Spark, API, contenidors). Defineixes els fluxos de treball com a DAGs (Directed Acyclic Graphs), que són només fitxers de Python que codifiquen les tasques i les seves dependències. Airflow executa aquestes tasques segons la teva programació, paràmetres i entorn.
- DAG: La definició del flux de treball (gràfic de tasques amb dependències).
- Task: Una unitat de treball (funció de Python, execució de SQL, ordre de Bash, activador de treball extern, etc.).
- Operator: Una plantilla per a un tipus de tasca (p. ex.,
PythonOperator, BashOperator, KubernetesPodOperator).
- Scheduler: Decideix què executar i quan.
- Executor: Executa les tasques (localment, amb Celery, Kubernetes, etc.).
- UI: El teu centre de control per a execucions, registres, reintents i llinatge.
Comença amb els tutorials oficials un cop hagis instal·lat Airflow; et donen una visió general ràpida.
Instal·lació i execució d'Airflow de la manera correcta
Airflow és flexible. Tria el camí que coincideixi amb la teva etapa:
- Desenvolupament local (inici ràpid):
- Utilitza el Docker Compose d'inici ràpid proporcionat pel projecte. Aixeca el servidor web, el planificador, la base de dades i molt més amb valors per defecte raonables.
- Ideal per aprendre i iterar en DAGs.
- Equip petit o posada en escena:
- Celery Executor o Kubernetes Executor amb un Postgres gestionat.
- Emmagatzema els registres a S3/GCS i empaqueta les dependències amb la teva imatge o
requirements.txt.
- Kubernetes Executor per a elasticitat o Celery Executor amb treballadors d'escalat automàtic.
- Secrets externs (Vault), observabilitat robusta (registres + mètriques) i desplegaments blau/verd per a actualitzacions.
Consell: Mantén el teu codi base d'Airflow controlat per versions, en contenidors i provat abans de la promoció. La pàgina de "Millors pràctiques" descriu els patrons preparats per a la producció.
Conceptes bàsics que utilitzaràs diàriament
DAGs: El teu flux de treball com a codi
Un DAG és un fitxer de Python que defineix:
- Metadades del DAG: id, programació, data d'inici, etiquetes.
- Arguments per defecte: reintents, propietaris, SLA.
- Tasques i les seves dependències.
Pensa en un DAG com el "què" i el "quan", i en les tasques com el "com".
Tasques i operadors
Els operadors són prefabricats per a tasques comunes. Exemples:
- PythonOperator / TaskFlow
@task per a codi Python
- BashOperator per a ordres de shell
- SimpleHttpOperator per a API
- KubernetesPodOperator per a treballs en contenidors
- Proveïdors de SQL (p. ex., Snowflake, BigQuery, Postgres) per a treballs de magatzem
API TaskFlow: La manera moderna i Pythonica
L'API TaskFlow et permet escriure tasques com a funcions de Python amb @task, retornar valors que passen a través de XCom i compondre'ls de manera neta. Redueix el codi repetitiu i millora la llegibilitat, molt recomanable.
El teu primer DAG d'Airflow (edició TaskFlow)
A continuació, es mostra un exemple mínim d'estil ETL per il·lustrar idees clau: programació, TaskFlow, dependències i pas de dades 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=.
Programació, Catchup i Backfills
schedule: Cron o valors predefinits (@daily, @hourly).
start_date + catchup: Si catchup=True, Airflow omplirà les execucions des de la data d'inici. Per als pipelines d'estil de transmissió, estableix catchup=False.
- Backfills manuals: Utilitza la IU o la CLI per tornar a executar els intervals històrics.
Regla pràctica: activa catchup per a treballs per lots deterministes; desactiva per a pipelines en temps real o amb limitació de velocitat d'API.
Passar dades entre tasques (XCom) de manera segura
- Objectes petits: els valors de retorn amb TaskFlow estan bé.
- Càrregues útils grans: emmagatzema a l'emmagatzematge d'objectes (S3/GCS) amb una clau a XCom.
- Evita les dades sensibles a XCom; utilitza backends de secrets (p. ex., Vault) i variables d'entorn.
Mapeig dinàmic de tasques i càrregues de treball de fan-out
Airflow pot generar tasques de manera dinàmica en temps d'execució en funció de les entrades, ideal per a conjunts de dades particionats o treballs multiinquilí.
- Mantén els DAG deterministes i idempotents.
- Separa l'orquestració (Airflow) del càlcul (Spark, dbt, magatzems).
- Utilitza l'API TaskFlow per a la claredat i la higiene de XCom.
- Parametriza els DAG; utilitza les variables amb criteri.
- Supervisa, alerta i documenta els teus pipelines.
Com treballar amb magatzems de dades i ML
- Magatzems de dades: Utilitza operadors de proveïdor (p. ex., SnowflakeOperator, BigQueryInsertJobOperator) per a treballs de SQL. Emmagatzema SQL en fitxers o mòduls versionats.
- dbt: Activa dbt mitjançant Bash/KubernetesPodOperator o operadors dbt dedicats als proveïdors.
- ML: Orquestra la generació de funcions, l'entrenament i la inferència per lots com a tasques separades; emmagatzema els artefactes a la memòria cau i registra les mètriques.
Programació avançada: Conjunts de dades i dependències entre DAG
- Els conjunts de dades permeten que un DAG produeixi un conjunt de dades lògic que activa un altre DAG quan s'actualitza, més net que els activadors ad-hoc.
- Per als patrons heretats, ExternalTaskSensor funciona, però els conjunts de dades són més declaratius.
Seguretat i compliment
- Utilitza el control d'accés basat en rols (RBAC) a la IU.
- Aïlla els entorns per equip o límit de confiança.
- Mantén les pistes d'auditoria mitjançant registres i l'historial de canvis de connexió.
Actualitzacions i control de versions
- Prova les actualitzacions en la posada en escena amb càrregues de treball similars a la producció.
- Fixa i actualitza els proveïdors deliberadament.
- Llegeix les notes de la versió per als canvis i les obsolescències específiques de l'executor.
Una llista de comprovació ràpida per al teu primer DAG de producció
- Propietat clara (etiqueta
owner) i alertes configurades.
retries establert amb backoffs raonables.
- Tasques idempotents i dependències explícites.
- Càrregues útils XCom petites; dades grans a l'emmagatzematge.
- Registres enviats a l'emmagatzematge durador; mètriques exportades.
- Pla de desplegament (canari o blau/verd) i passos de rollback.
Exemple: Un DAG de càrrega de magatzem realista
Aquest patró extreu fitxers diaris, els valida i els carrega en una taula de magatzem, amb un mapeig dinàmic per partició i sensors diferibles.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.sensors.filesystem import FileSensor
- Revisa les millors pràctiques abans de promocionar a producció.
- Explora els documents del proveïdor per als teus sistemes (magatzems, núvols, eines de ML).
Per cert: accelera la creació amb un company d'IA
Val la pena assenyalar: si redactes molts DAG, un assistent d'IA que entengui el codi pot accelerar el codi repetitiu, generar stubs de TaskFlow i fins i tot suggerir correccions de dependència. Si vols un ajudant lleuger al costat del teu editor i navegador, Sider.AI pot ser útil per a reescriptures i explicacions ràpides de codi durant el desenvolupament. Principals conclusions
- Utilitza Airflow per orquestrar, no per calcular.
- Prefereix l'API TaskFlow per a DAG nets i provables.
- Mantén les dades fora de XCom; passa referències en canvi.
- Utilitza sensors/operadors diferibles per estalviar espais.
- Contenidoritza, prova i promociona a través d'entorns.
- Confia en els tutorials oficials i les millors pràctiques com la teva estrella polar.
FAQ
Q1:Quina és la manera més fàcil d'aprendre a utilitzar Airflow?
Comença amb el tutorial oficial per entendre els DAG, les tasques, la programació i la IU. A continuació, construeix un petit pipeline basat en TaskFlow i itera amb la guia de millors pràctiques per a la preparació per a la producció.
Q2:Hauria d'utilitzar l'API TaskFlow o els operadors clàssics a Airflow?
Utilitza l'API TaskFlow per a la majoria dels pipelines Pythonic perquè és més net i gestiona els retorns de XCom de manera natural. Els operadors clàssics encara són excel·lents per a tasques que no són de Python com Bash, SQL o treballs de contenidors.
Q3:Com puc passar dades grans entre les tasques d'Airflow?
Evita posar càrregues útils grans a XCom. Emmagatzema les dades a S3/GCS o a una base de dades i passa només referències o URI a través de XCom per mantenir les tasques ràpides i fiables.
Q4:Quin executor hauria de triar per a Airflow en producció?
Per a l'elasticitat i l'aïllament, Kubernetes Executor és un valor per defecte fort. Per a configuracions més senzilles, Celery Executor funciona bé, només assegura l'escalat automàtic, el registre robust i els secrets externalitzats.
Q5:Com gestiono les dependències entre múltiples DAG d'Airflow?
Utilitza conjunts de dades per als activadors declaratius entre DAG quan un pipeline produeix dades per a un altre. Alternativament, ExternalTaskSensor pot coordinar les execucions, però els conjunts de dades són més nets per a l'orquestració basada en dades.
