Prefect: Modern Workflow Orchestration for Data Pipelines

Prefect is the modern Python workflow orchestration platform that treats your code as the source of truth. No DAG classes, no XML — just Python functions decorated with @flow and @task, with built-in retries, caching, scheduling, and a beautiful UI.

What is Prefect?

Prefect is an open-source workflow orchestration tool built for data engineers and data scientists who want a modern alternative to Apache Airflow. Released in 2018, Prefect 2.0 (Orion) fundamentally redesigned the architecture: workflows are native Python functions, dynamic control flow is supported out of the box, and the deployment model decouples scheduling from execution through Work Pools and Workers.

Prefect Cloud provides a hosted control plane with a rich UI showing flow run history, task states, logs, and scheduling — with a generous free tier. Workers run on your own infrastructure (local, Docker, Kubernetes, ECS) and poll Prefect Cloud for work. This hybrid model means your data never leaves your infrastructure while still getting the benefits of a managed orchestration backend.

Flows, Tasks & Retries

# pip install prefect
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta
import httpx

# Tasks — individual units of work with retry/caching
@task(
    retries=3,
    retry_delay_seconds=10,
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=1),
)
def fetch_data(url: str) -> dict:
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

@task(log_prints=True)
def transform_data(raw: dict) -> list[dict]:
    records = raw.get("data", [])
    return [
        {
            "id": r["id"],
            "value": r["metrics"]["total"] * 1.1,
            "category": r["type"].upper(),
        }
        for r in records
        if r.get("active", True)
    ]

@task
def load_to_warehouse(records: list[dict]) -> int:
    # ... write to database
    print(f"Loaded {len(records)} records")
    return len(records)

# Flow — orchestrates tasks
@flow(name="ETL Pipeline", log_prints=True)
def etl_pipeline(api_url: str = "https://api.example.com/data") -> int:
    raw = fetch_data(api_url)
    transformed = transform_data(raw)
    count = load_to_warehouse(transformed)
    return count

if __name__ == "__main__":
    result = etl_pipeline()
    print(f"Pipeline done: {result} records loaded")

Deployments & Scheduling

# Prefect Deployments — schedule flows for production
from prefect import flow
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule

@flow
def etl_pipeline(api_url: str = "https://api.example.com"):
    ...

# Create a deployment with a cron schedule
deployment = Deployment.build_from_flow(
    flow=etl_pipeline,
    name="production-etl",
    parameters={"api_url": "https://api.example.com/v2"},
    schedule=CronSchedule(cron="0 6 * * *", timezone="UTC"),  # daily at 6am
    work_pool_name="my-work-pool",
    tags=["production", "etl"],
)
deployment.apply()

# --- CLI equivalent ---
# prefect deploy --name production-etl
# prefect worker start --pool my-work-pool
# prefect deployment run etl-pipeline/production-etl

Error Handling & Subflows

# Error handling, retries, and subflows
from prefect import flow, task
from prefect.utilities.annotations import quote
import asyncio

@task(retries=5, retry_delay_seconds=exponential(60))
async def call_flaky_api(endpoint: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(endpoint, timeout=10)
        resp.raise_for_status()
        return resp.json()

@flow
def process_region(region: str) -> dict:
    """Subflow — runs independently, has its own run in the UI."""
    data = call_flaky_api(f"https://api.example.com/{region}")
    return {"region": region, "count": len(data)}

@flow(name="Global ETL", on_failure=[send_alert])
def global_etl():
    # Run subflows for each region — appear as nested runs in UI
    results = [process_region(r) for r in ["us", "eu", "ap"]]
    total = sum(r["count"] for r in results)
    print(f"Total records: {total}")

def send_alert(flow, flow_run, state):
    """Called automatically on flow failure."""
    print(f"Flow {flow.name} failed: {state.message}")

Prefect vs Airflow vs Dagster