Airflow DAGs
TaskFlow API
from airflow.decorators import dag, task
from datetime import datetime
@dag(schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False)
def etl():
@task()
def extract() -> dict: return fetch_data()
@task()
def transform(data: dict) -> dict: return clean(data)
@task()
def load(data: dict): write_to_db(data)
load(transform(extract()))
etl()
Classic DAG
with DAG('pipeline', schedule='0 6 * * *', default_args={'retries': 2}) as dag:
t1 = PythonOperator(task_id='extract', python_callable=extract)
t2 = BashOperator(task_id='transform', bash_command='python transform.py')
t1 >> t2
Sensors: FileSensor, HttpSensor — wait for conditions
XCom: ti.xcom_push(key='count', value=42) / ti.xcom_pull(task_ids='extract')
CLI: airflow dags trigger / airflow tasks test
ComeOnOliver