Apache Airflow vs Prefect: Welke Orchestrator Kies Je?

Introductie: De Strijd om Jouw Pipeline

Als data engineer in 2026 kun je er niet omheen: workflow orchestratie is het zenuwstelsel van elke moderne data stack. Je hebt duizenden taken die op het juiste moment, in de juiste volgorde en met de juiste afhankelijkheden moeten draaien. Een fout in de orchestratielaag betekent stale dashboards, mislukte ML-modellen of — erger nog — verkeerde business beslissingen op basis van verouderde data.

Twee namen domineren het gesprek: Apache Airflow en Prefect. Airflow is de gevestigde orde, geboren bij Airbnb in 2014 en inmiddels het meest gebruikte orchestratieplatform ter wereld. Prefect is de uitdager: opgericht in 2018 met de expliciete missie om alles beter te doen dan Airflow. Beide tools hebben hun fanatieke aanhangers, en beide hebben legitieme redenen om te bestaan.

In 2026 zijn beide tools volwassen. Airflow 3.0 (uitgebracht in 2025) heeft de architectuur ingrijpend verbeterd. Prefect 3.x heeft zijn serverloze ambities volledig waargemaakt. De keuze is dus niet meer "gebruik Airflow want Prefect is onvolwassen" — het is een serieuze architectuurbeslissing met langdurige gevolgen. In deze blog leggen we alles naast elkaar zodat jij een weloverwogen keuze kunt maken.

Hoe Werken Ze Allebei?

Apache Airflow: DAGs als Eerste Principe

Airflow is gebouwd rond het concept van de Directed Acyclic Graph (DAG). Een DAG is een Python-bestand dat de structuur van je pipeline definieert: welke taken er zijn, in welke volgorde ze draaien en wat hun onderlinge afhankelijkheden zijn. De daadwerkelijke logica leeft in Operators (PythonOperator, BashOperator, SparkSubmitOperator, etc.).

Prefect: Flows als Gewone Python

Prefect kiest een fundamenteel andere aanpak. In plaats van dat je je pipeline expliciet als een graph definieert, decoreer je gewone Python-functies met @flow en @task. Prefect inspecteert tijdens runtime de uitvoering en bouwt de dependency-graph automatisch op. Dit maakt Prefect-code significant leesbaarder en makkelijker te debuggen.

Praktische Codevoorbeelden

Use Case: Dagelijkse ETL Pipeline (Airflow)

We bouwen een pipeline die data uit een REST API haalt, transformeert en naar een PostgreSQL-database schrijft.

# airflow/dags/dagelijkse_api_etl.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta
import requests
import pandas as pd

# Standaard argumenten voor alle taken
default_args = {
    "owner": "data-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-alerts@bedrijf.nl"],
}

def extract_api_data(**context):
    """Haalt data op van de externe API."""
    execution_date = context["ds"]  # YYYY-MM-DD formaat
    response = requests.get(
        f"https://api.voorbeeld.nl/orders",
        params={"date": execution_date},
        headers={"Authorization": "Bearer {{ var.value.api_token }}"},
        timeout=30
    )
    response.raise_for_status()
    
    # Sla data op via XCom voor de volgende taak
    context["ti"].xcom_push(key="raw_orders", value=response.json())
    print(f"✅ {len(response.json())} orders opgehaald voor {execution_date}")

def transform_data(**context):
    """Valideert en transformeert de ruwe orderdata."""
    raw_data = context["ti"].xcom_pull(
        task_ids="extract_data", 
        key="raw_orders"
    )
    
    df = pd.DataFrame(raw_data)
    
    # Transformaties
    df["order_date"] = pd.to_datetime(df["order_date"])
    df["total_amount"] = df["quantity"] * df["unit_price"]
    df = df[df["status"] != "CANCELLED"]  # Filter geannuleerde orders
    df = df.dropna(subset=["customer_id"])
    
    print(f"📊 {len(df)} geldige orders na transformatie")
    context["ti"].xcom_push(key="clean_orders", value=df.to_dict("records"))

def load_to_postgres(**context):
    """Laadt getransformeerde data naar PostgreSQL."""
    clean_data = context["ti"].xcom_pull(
        task_ids="transform_data",
        key="clean_orders"
    )
    
    hook = PostgresHook(postgres_conn_id="postgres_datawarehouse")
    
    # Upsert logica: INSERT ON CONFLICT UPDATE
    insert_sql = """
        INSERT INTO orders_staging (order_id, customer_id, order_date, total_amount, status)
        VALUES (%s, %s, %s, %s, %s)
        ON CONFLICT (order_id) DO UPDATE
        SET total_amount = EXCLUDED.total_amount,
            status = EXCLUDED.status,
            updated_at = NOW();
    """
    
    rows = [
        (r["order_id"], r["customer_id"], r["order_date"], 
         r["total_amount"], r["status"])
        for r in clean_data
    ]
    
    hook.run(insert_sql, parameters=rows)
    print(f"💾 {len(rows)} orders geladen naar datawarehouse")

# DAG definitie
with DAG(
    dag_id="dagelijkse_api_etl",
    default_args=default_args,
    description="Dagelijkse ETL van orders API naar datawarehouse",
    schedule_interval="0 6 * * *",  # Elke dag om 06:00
    start_date=datetime(2026, 1, 1),
    catchup=False,  # Geen backfill voor gemiste runs
    tags=["etl", "orders", "productie"],
) as dag:

    extract_task = PythonOperator(
        task_id="extract_data",
        python_callable=extract_api_data,
    )

    transform_task = PythonOperator(
        task_id="transform_data",
        python_callable=transform_data,
    )

    load_task = PythonOperator(
        task_id="load_to_postgres",
        python_callable=load_to_postgres,
    )

    # Afhankelijkheidsketen definiëren
    extract_task >> transform_task >> load_task

Dezelfde Pipeline in Prefect

Nu dezelfde logica in Prefect — merk op hoe de code voelt als gewone Python:

# prefect/flows/dagelijkse_api_etl.py
from prefect import flow, task, get_run_logger
from prefect.blocks.system import Secret
from prefect_sqlalchemy import SqlAlchemyConnector
import requests
import pandas as pd
from datetime import date

@task(
    retries=3,
    retry_delay_seconds=300,
    name="extract-api-data",
    description="Haalt orders op van de externe API"
)
def extract_api_data(target_date: date) -> list[dict]:
    """Haalt data op van de externe API."""
    logger = get_run_logger()
    
    api_token = Secret.load("api-token").get()
    
    response = requests.get(
        "https://api.voorbeeld.nl/orders",
        params={"date": target_date.isoformat()},
        headers={"Authorization": f"Bearer {api_token}"},
        timeout=30
    )
    response.raise_for_status()
    
    orders = response.json()
    logger.info(f"✅ {len(orders)} orders opgehaald voor {target_date}")
    return orders

@task(
    name="transform-data",
    description="Valideert en transformeert orderdata"
)
def transform_data(raw_orders: list[dict]) -> list[dict]:
    """Valideert en transformeert de ruwe orderdata."""
    logger = get_run_logger()
    
    df = pd.DataFrame(raw_orders)
    df["order_date"] = pd.to_datetime(df["order_date"])
    df["total_amount"] = df["quantity"] * df["unit_price"]
    df = df[df["status"] != "CANCELLED"]
    df = df.dropna(subset=["customer_id"])
    
    logger.info(f"📊 {len(df)} geldige orders na transformatie")
    return df.to_dict("records")

@task(
    name="load-to-postgres",
    description="Laadt getransformeerde data naar PostgreSQL"
)
def load_to_postgres(clean_orders: list[dict]) -> int:
    """Laadt getransformeerde data naar PostgreSQL."""
    logger = get_run_logger()
    
    connector = SqlAlchemyConnector.load("postgres-datawarehouse")
    
    with connector.get_connection(begin=False) as conn:
        for order in clean_orders:
            conn.execute("""
                INSERT INTO orders_staging 
                    (order_id, customer_id, order_date, total_amount, status)
                VALUES (:order_id, :customer_id, :order_date, :total_amount, :status)
                ON CONFLICT (order_id) DO UPDATE
                SET total_amount = EXCLUDED.total_amount,
                    status = EXCLUDED.status,
                    updated_at = NOW()
            """, order)
    
    logger.info(f"💾 {len(clean_orders)} orders geladen naar datawarehouse")
    return len(clean_orders)

# De flow: dit voelt als gewone Python!
@flow(
    name="dagelijkse-api-etl",
    description="Dagelijkse ETL van orders API naar datawarehouse",
    log_prints=True
)
def dagelijkse_etl_flow(target_date: date = date.today()):
    # Stap 1: Extract
    raw_orders = extract_api_data(target_date)
    
    # Stap 2: Transform (automatisch afhankelijk van extract)
    clean_orders = transform_data(raw_orders)
    
    # Stap 3: Load (automatisch afhankelijk van transform)
    loaded_count = load_to_postgres(clean_orders)
    
    return f"Succesvol {loaded_count} orders verwerkt"

if __name__ == "__main__":
    # Lokaal uitvoeren voor testen — geen scheduler nodig!
    dagelijkse_etl_flow()

Directe Vergelijking: Airflow vs Prefect

Vergelijking met Andere Orchestrators

Airflow en Prefect zijn niet de enige spelers. In 2026 zijn er serieuze alternatieven die voor specifieke use cases beter kunnen passen:

Best Practices & Production Tips

Apache Airflow in Productie

# airflow/dags/production_best_practice.py
# ✅ GOED: Imports buiten de module-scope, idempotente logica

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

# ✅ Imports die bij task-uitvoering plaatsvinden, NIET bij DAG-parsing
def transform_data(**context):
    import pandas as pd  # Import hier, niet globaal!
    from mypackage import transformer
    
    ds = context["ds"]  # Gebruik execution_date als partitie
    
    # ✅ Idempotent: schrijf naar partitie gebaseerd op datum
    output_path = f"s3://my-bucket/orders/date={ds}/output.parquet"
    
    df = transformer.run(date=ds)
    df.to_parquet(output_path)
    
    # ✅ Kleine XCom: alleen het pad, niet de data
    return output_path  # Automatisch als XCom opgeslagen

Prefect in Productie

# prefect/flows/production_best_practice.py
from prefect import flow, task
from prefect.tasks import task_input_hash
from prefect.artifacts import create_table_artifact
from datetime import timedelta

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=2),
    # ✅ Cache: sla resultaat op voor 24 uur bij zelfde input
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=24),
    persist_result=True  # ✅ Opslaan in geconfigureerde result storage
)
def expensive_transformation(date: str, config: dict) -> list[dict]:
    """Zware transformatie die gecached wordt bij identieke input."""
    import pandas as pd
    # ... zware berekeningen ...
    return results

@flow(
    name="productie-flow",
    result_storage="s3-bucket/prefect-results",  # Expliciete result storage
)
def productie_pipeline(target_date: str):
    # ✅ Artifacts voor rijke UI observability
    results = expensive_transformation(target_date, config={"env": "prod"})
    
    create_table_artifact(
        key="resultaat-samenvatting",
        table=results[:10],  # Preview van eerste 10 rijen
        description=f"Pipeline resultaten voor {target_date}"
    )
    
    return results

Wanneer Kies Je Wat?

Na al deze vergelijkingen de praktische beslisboom voor Nederlandse data engineers: