$devvkit learn --librarie airflow-&-prefect-guide

Airflow & Prefect Guide

[workflow][data][scheduling][python]

Automation & CI/CD

Install

# Airflow:
pip install apache-airflow
airflow db init
airflow standalone

# Prefect:
pip install prefect
prefect server start
# Open: http://localhost:8080

Airflow is the industry standard for data pipeline orchestration. You define DAGs (Directed Acyclic Graphs) in Python: each task is a step, dependencies define execution order. The web UI shows DAG runs, task status, logs, and timelines (Gantt charts).

Prefect is the modern alternative with native async support, automatic retries, caching, and a cleaner API. Tasks are decorated functions (`@task`), flows are the DAG (`@flow`). Prefect Cloud provides observability without managing a server. Both handle scheduling, retries, alerting, and backfilling.

Core concepts: Operators/ Tasks are the work units, DAGs/Flows define the graph, Scheduler triggers runs, Executor runs tasks (Sequential, Local, Celery, Kubernetes). Airflow has 1000+ operators (AWS, GCP, Azure, Snowflake, dbt, etc.). GUI: Airflow UI, Prefect UI, Dagster Dagit.

Airflow Setup

Start Airflow· Initialize and start.

export AIRFLOW_HOME=~/airflow
echo "AIRFLOW_HOME=$AIRFLOW_HOME"
airflow db init                            # Create SQLite DB
airflow users create \
  --username admin --password admin \
  --firstname Admin --lastname Admin \
  --role Admin --email admin@example.com
airflow standalone                          # Start webserver + scheduler
# Open: http://localhost:8080

Airflow: Celery executor· Scale with workers.

# airflow.cfg
executor = CeleryExecutor
celery_result_backend = db+postgresql://airflow:airflow@postgres/airflow

# Start workers:
airflow celery worker

# Start flower (monitoring):
airflow celery flower

# Kubernetes executor for auto-scaling:
# executor = KubernetesExecutor

Airflow DAGs

Basic DAG· Define a simple workflow.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG(
    'my_pipeline',
    start_date=datetime(2025, 1, 1),
    schedule='@daily',                    # Cron expression
    catchup=False,
    default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)}
) as dag:

    def extract(): return {'data': [1, 2, 3]}
    def transform(data): return [x * 2 for x in data]
    def load(data): print(f'Loaded: {data}')

    extract_task = PythonOperator(task_id='extract', python_callable=extract)
    transform_task = PythonOperator(
        task_id='transform', python_callable=transform,
        op_kwargs={'data': '{{ ti.xcom_pull(task_ids="extract") }}'}
    )
    load_task = PythonOperator(task_id='load', python_callable=load)

    extract_task >> transform_task >> load_task

Airflow CLI: manage DAGs· CLI operations.

airflow dags list                           # List all DAGs
airflow dags pause my_dag                    # Pause DAG
airflow dags unpause my_dag                  # Unpause
airflow dags trigger my_dag                  # Manual trigger
airflow tasks test my_dag extract_task 2025-01-01  # Test single task
airflow dags backfill my_dag -s 2025-01-01 -e 2025-01-10  # Backfill

Airflow Operators

Airflow BashOperator· Run shell commands.

from airflow.operators.bash import BashOperator

task = BashOperator(
    task_id='run_script',
    bash_command='python /scripts/process.py --date {{ ds }}',
    env={'DATABASE_URL': 'postgres://...'},
    cwd='/app'
)

Airflow sensors· Wait for external events.

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id='wait_for_data',
    filepath='/data/input_{{ ds }}.csv',
    poke_interval=30,       # Check every 30s
    timeout=3600,           # Timeout after 1 hour
    mode='reschedule',      # Don't hold worker slot
)

extract >> wait_for_file >> process

Prefect Setup

Prefect: basic flow· Simple data pipeline.

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task(retries=2, retry_delay_seconds=30)
def fetch_data(url: str) -> dict:
    import httpx
    return httpx.get(url).json()

@task
def transform(data: dict) -> list:
    return [item['value'] * 2 for item in data['items']]

@task
def save(results: list) -> None:
    import json
    with open('output.json', 'w') as f:
        json.dump(results, f)

@flow(name='data_pipeline', task_runner=ConcurrentTaskRunner())
def pipeline(url: str = 'https://api.example.com/data'):
    data = fetch_data(url)
    transformed = transform(data)
    save(transformed)

# Run:
pipeline()
pipeline(url='https://other-api.com')

Prefect Flows

Prefect: schedule & deploy· Production deployment.

# Prefect server:
prefect server start

# Create deployment:
pipeline.serve(name='prod-pipeline', cron='0 6 * * *')

# Or via CLI:
prefect deploy pipeline.py:pipeline
  --name prod-pipeline
  --cron '0 6 * * *'
  --param url=https://api.example.com

# Start worker:
prefect worker start --pool default