How To Use Python Prefect for Data Pipeline Orchestration

Intermediate

You’ve got a data pipeline that works perfectly — until it doesn’t. The API returns a 503 at 3am, your CSV has a malformed row on line 4,712, and the entire pipeline crashes silently. By morning you have stale data, confused dashboards, and no clue where things went wrong. Stitching together retries, logging, and scheduling with raw Python is possible, but it quickly becomes a second job on top of your actual job.

Prefect is a workflow orchestration framework for Python that handles all of this for you. You decorate your existing functions with @task and @flow, and suddenly you get automatic retries, structured logging, result caching, scheduling, and a real-time UI — without rewriting your business logic. Prefect 3.x (the current stable release) is pip-installable and works entirely locally; you don’t need a managed cloud service unless you want one.

In this article we will cover installing Prefect, decorating functions into tasks and flows, adding retries and caching, inspecting runs in the Prefect UI, scheduling automated runs, and building a full ETL pipeline as a real-life project. By the end you will be able to turn any ad-hoc Python script into a monitored, fault-tolerant pipeline in under an hour.

Prefect in Python: Quick Example

The fastest way to see Prefect in action is to build a tiny pipeline with two tasks chained inside a flow. The example below fetches a post from a public API and prints its title — with automatic logging at every step.

# quick_pipeline.py
import requests
from prefect import flow, task

@task
def fetch_post(post_id: int) -> dict:
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

@task
def print_title(post: dict) -> None:
    print(f"Post title: {post['title']}")

@flow(name="fetch-post-flow")
def run_pipeline(post_id: int = 1) -> None:
    post = fetch_post(post_id)
    print_title(post)

if __name__ == "__main__":
    run_pipeline(post_id=3)

Output:

13:42:01.123 | INFO    | prefect.engine - Created flow run 'cerulean-wolf' for flow 'fetch-post-flow'
13:42:01.201 | INFO    | Task run 'fetch_post-0' - Finished in state Completed()
Post title: ea molestias quasi exercitationem repellat qui ipsa sit aut
13:42:01.340 | INFO    | Task run 'print_title-0' - Finished in state Completed()
13:42:01.341 | INFO    | Flow run 'cerulean-wolf' - Finished in state Completed()

Each @task decorator wraps a regular function and gives it Prefect’s observability: state tracking, structured logs, and retry capability. The @flow decorator marks the entry point that orchestrates those tasks. Prefect generates a unique run name (like cerulean-wolf) automatically, which you can look up in the UI later.

The rest of this article digs into what happens when tasks fail, how to cache expensive results so they don’t rerun needlessly, and how to schedule the whole pipeline to run on a cron-style interval.

What Is Prefect and Why Use It?

Prefect is a Python-native workflow orchestration tool. At its core it answers one question: “When a step in my pipeline fails, what should happen?” The naive answer is “crash and let me deal with it.” Prefect’s answer is “retry it, log exactly why it failed, keep track of what succeeded, and notify me if it ultimately can’t recover.”

Think of Prefect like a project manager for your code. You define the tasks (individual units of work) and the flow (the order and dependencies between them). Prefect handles the coordination: running tasks in order, passing results between them, catching failures, and reporting status. You still write all the business logic in plain Python — Prefect just wraps it with supervision.

Feature	Raw Python Script	Prefect Flow
Automatic retries	Manual try/except loops	Built-in: retries=3
Structured logging	print() or custom logger	Auto-captured per task run
Run history	None (or log files)	Searchable UI with state history
Result caching	Manual file/pickle logic	Built-in: cache_key_fn
Scheduling	cron + bare scripts	Native deployments + schedules
Parallelism	threading/multiprocessing	Task runners (concurrent, Dask)

Prefect is not the only orchestration option — Airflow and Luigi are popular alternatives. The key difference is the developer experience: Prefect works with plain Python functions, needs zero XML or DAG configuration files, and starts a local server in one command. Airflow is a better fit for teams that already have it deployed and need its plugin ecosystem; Prefect is faster to get off the ground for Python-first workflows.

Installing Prefect

Prefect 3.x installs from PyPI like any other package. Using a virtual environment keeps your orchestration dependencies separate from your project’s other libraries.

# install_prefect.sh
python -m venv prefect-env
source prefect-env/bin/activate   # Windows: prefect-env\Scripts\activate
pip install prefect requests

Output:

Successfully installed prefect-3.x.x ...

Verify the install by checking the version:

# verify_prefect.py
import prefect
print(prefect.__version__)

Output:

3.4.1

To launch the local Prefect server (which gives you the monitoring UI), open a second terminal and run:

# in a separate terminal
prefect server start

The server starts on http://127.0.0.1:4200 by default. Keep it running while you execute flows so run history gets captured. You don’t need the server to run flows — they work locally without it — but it is well worth having for debugging.

Creating Tasks and Flows

The building blocks of every Prefect pipeline are tasks and flows. A task is a single unit of work — typically one function that does one thing. A flow is the container that calls those tasks in the right order and passes results between them.

Tasks and flows are created by decorating regular Python functions. The decorators accept configuration arguments that control behavior without changing the underlying logic.

# tasks_and_flows.py
from prefect import flow, task
import requests

@task(name="fetch-users", log_prints=True)
def fetch_users(limit: int) -> list:
    """Fetch user records from a public API."""
    url = "https://jsonplaceholder.typicode.com/users"
    response = requests.get(url, params={"_limit": limit}, timeout=10)
    response.raise_for_status()
    users = response.json()
    print(f"Fetched {len(users)} users")
    return users

@task(name="extract-emails", log_prints=True)
def extract_emails(users: list) -> list:
    """Pull email addresses from a list of user records."""
    emails = [u["email"] for u in users if "email" in u]
    print(f"Extracted {len(emails)} email addresses")
    return emails

@flow(name="user-email-pipeline", log_prints=True)
def user_email_pipeline(limit: int = 5) -> list:
    users = fetch_users(limit)
    emails = extract_emails(users)
    return emails

if __name__ == "__main__":
    result = user_email_pipeline(limit=3)
    print("Result:", result)

Output:

Fetched 3 users
Extracted 3 email addresses
Result: ['Sincere@april.biz', 'Shanna@melissa.tv', 'Nathan@yesenia.net']

The log_prints=True argument tells Prefect to capture every print() call inside that task or flow and route it through Prefect’s logger so it appears in the UI’s run history. Without it, print() still works but the output is not linked to the run record.

Notice that extract_emails receives the return value of fetch_users directly. Prefect passes results between tasks as regular Python objects — there is no serialization ceremony or special data store required for basic in-memory use.

Adding Retries and Error Handling

The most valuable feature for production pipelines is retry logic. Real-world APIs return 503s, database connections time out, and file systems occasionally misbehave. Rather than wrapping every task in try/except loops, you declare the retry policy at decoration time.

# retries_example.py
import requests
import random
from prefect import flow, task

@task(
    name="unreliable-api-call",
    retries=3,
    retry_delay_seconds=2,
    log_prints=True
)
def fetch_with_retries(post_id: int) -> dict:
    """Simulate an unreliable API call that randomly fails."""
    # Simulate 50% failure rate for demonstration
    if random.random() < 0.5:
        raise requests.exceptions.ConnectionError("Simulated network error")
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

@flow(name="retry-demo-flow")
def retry_demo() -> None:
    post = fetch_with_retries(1)
    print(f"Retrieved: {post['title'][:50]}")

if __name__ == "__main__":
    retry_demo()

Output (when a retry occurs):

13:52:01.101 | INFO    | Task run 'unreliable-api-call-0' - Retrying in 2.0 seconds...
13:52:03.205 | INFO    | Task run 'unreliable-api-call-0' - Retrying in 2.0 seconds...
13:52:05.311 | INFO    | Task run 'unreliable-api-call-0' - Finished in state Completed()
Retrieved: sunt aut facere repellat provident occaecati

With retries=3 and retry_delay_seconds=2, Prefect automatically attempts the task up to three more times with a two-second pause between each try. If all retries are exhausted, the task transitions to a Failed state, the flow marks the run as failed, and the failure is recorded in the UI with the full traceback — no extra code required.

For exponential backoff (doubling the wait time on each retry, which is gentler on flaky APIs), use a list:

# exponential_backoff.py
from prefect import task

@task(retries=4, retry_delay_seconds=[1, 2, 4, 8])
def fetch_with_backoff(url: str) -> dict:
    import requests
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

The retry delays list maps directly to each retry attempt: first retry waits 1 second, second waits 2, third waits 4, fourth waits 8. This pattern prevents thundering-herd situations where every retrying client hammers a recovering service simultaneously.

Caching Task Results

Some tasks are expensive to run — calling a paid API, running a slow database query, or processing a large file. If those tasks produce the same output for the same input, re-running them on every pipeline execution is wasteful. Prefect’s result caching lets you skip redundant work automatically.

# caching_example.py
import requests
from datetime import timedelta
from prefect import flow, task
from prefect.tasks import task_input_hash

@task(
    name="cached-api-fetch",
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(minutes=10),
    log_prints=True
)
def fetch_post_cached(post_id: int) -> dict:
    """Fetch a post -- result is cached per post_id for 10 minutes."""
    print(f"Making real API call for post {post_id}...")
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

@flow(name="caching-demo")
def caching_demo() -> None:
    # First call -- makes the real API request
    post = fetch_post_cached(1)
    print(f"First run: {post['title'][:40]}")

    # Second call with same argument -- returns cached result
    post_again = fetch_post_cached(1)
    print(f"Second run (cached): {post_again['title'][:40]}")

if __name__ == "__main__":
    caching_demo()

Output:

Making real API call for post 1...
First run: sunt aut facere repellat provident occae
13:55:12.001 | INFO    | Task run 'cached-api-fetch-1' - Finished in state Cached(type=COMPLETED)
Second run (cached): sunt aut facere repellat provident occae

The cache_key_fn=task_input_hash argument tells Prefect to generate a cache key from the task’s inputs (in this case post_id). When the same task runs with the same arguments within the cache expiration window, Prefect returns the stored result immediately — notice the Cached(type=COMPLETED) state in the output, and the absence of the “Making real API call” print statement on the second run.

Set cache_expiration based on how frequently the underlying data actually changes. For API data that refreshes hourly, timedelta(hours=1) is appropriate. For reference data that rarely changes, timedelta(days=1) keeps things responsive without hammering the source.

Structured Logging with Prefect

Prefect captures structured log output per task run, making it easy to trace exactly what each step did without sifting through a monolithic log file. The preferred approach is to use Prefect’s built-in logger via get_run_logger().

# logging_example.py
import requests
from prefect import flow, task, get_run_logger

@task(name="fetch-comments")
def fetch_comments(post_id: int) -> list:
    logger = get_run_logger()
    logger.info(f"Fetching comments for post {post_id}")
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}/comments"
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        comments = response.json()
        logger.info(f"Retrieved {len(comments)} comments")
        return comments
    except requests.exceptions.RequestException as err:
        logger.error(f"Failed to fetch comments: {err}")
        raise

@task(name="count-emails")
def count_unique_emails(comments: list) -> int:
    logger = get_run_logger()
    emails = {c["email"] for c in comments if c.get("email")}
    logger.info(f"Found {len(emails)} unique commenter emails")
    return len(emails)

@flow(name="comment-analysis")
def comment_analysis(post_id: int = 1) -> int:
    comments = fetch_comments(post_id)
    count = count_unique_emails(comments)
    return count

if __name__ == "__main__":
    total = comment_analysis(post_id=1)
    print(f"Unique commenters: {total}")

Output:

14:01:03.021 | INFO    | Task run 'fetch-comments-0' - Fetching comments for post 1
14:01:03.412 | INFO    | Task run 'fetch-comments-0' - Retrieved 5 comments
14:01:03.413 | INFO    | Task run 'count-emails-0' - Found 5 unique commenter emails
14:01:03.414 | INFO    | Flow run 'amber-falcon' - Finished in state Completed()
Unique commenters: 5

get_run_logger() returns a standard Python logger that Prefect automatically associates with the current task or flow run. Every message is timestamped, tagged with the run name, and visible in the Prefect UI’s log panel for that specific run. Use logger.info() for routine progress, logger.warning() for unexpected-but-handled situations, and logger.error() before re-raising exceptions so the failure has context in the UI.

Scheduling Flows with Deployments

Running a flow on demand with python my_flow.py is useful for development. In production you typically want it to run on a schedule — every hour, once a day, or on a cron expression. Prefect handles this through deployments.

A deployment is a named configuration that links a flow to a schedule and an execution environment. The simplest way to create one is to call .serve() on your flow, which registers the deployment and starts a worker that listens for scheduled runs.

# scheduled_flow.py
import requests
from prefect import flow, task

@task(retries=2, retry_delay_seconds=5)
def fetch_latest_post() -> dict:
    """Fetch the most recently created post."""
    response = requests.get(
        "https://jsonplaceholder.typicode.com/posts",
        params={"_sort": "id", "_order": "desc", "_limit": 1},
        timeout=10
    )
    response.raise_for_status()
    return response.json()[0]

@flow(name="hourly-post-check", log_prints=True)
def hourly_post_check() -> None:
    post = fetch_latest_post()
    print(f"Latest post ID {post['id']}: {post['title'][:50]}")

if __name__ == "__main__":
    # Run on an interval (every 60 seconds for demo purposes)
    from prefect.schedules import IntervalSchedule
    from datetime import timedelta

    hourly_post_check.serve(
        name="hourly-post-deployment",
        interval=timedelta(seconds=60)  # Change to timedelta(hours=1) in production
    )

Output (runs every 60 seconds while the script is active):

Your deployment 'hourly-post-deployment' is running!
Press Ctrl+C to stop.
14:10:00.001 | INFO    | Flow run 'crimson-lynx' - Latest post ID 100: at nam consequatur ea labore ea harum
14:11:00.023 | INFO    | Flow run 'lilac-crane' - Latest post ID 100: at nam consequatur ea labore ea harum

The .serve() method blocks, keeps the script alive, and triggers the flow on the configured schedule. For cron-based schedules — such as running at 6am every weekday — use cron="0 6 * * 1-5" instead of interval. The Prefect UI at http://127.0.0.1:4200 shows upcoming scheduled runs, past run history, and any failures that need attention.

Real-Life Example: ETL Pipeline for User Data

Let’s put everything together into a practical ETL (Extract, Transform, Load) pipeline. This flow fetches user records from a public API, transforms them into a clean format, and writes the results to a local SQLite database — a pattern you can adapt for real data sources and destinations.

# etl_pipeline.py
import sqlite3
import requests
from datetime import timedelta
from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash

# -------------------------------------------------------
# EXTRACT
# -------------------------------------------------------
@task(
    name="extract-users",
    retries=3,
    retry_delay_seconds=[2, 4, 8],
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=1)
)
def extract_users(limit: int) -> list:
    logger = get_run_logger()
    logger.info(f"Extracting {limit} users from API")
    response = requests.get(
        "https://jsonplaceholder.typicode.com/users",
        params={"_limit": limit},
        timeout=15
    )
    response.raise_for_status()
    users = response.json()
    logger.info(f"Extracted {len(users)} user records")
    return users

# -------------------------------------------------------
# TRANSFORM
# -------------------------------------------------------
@task(name="transform-users")
def transform_users(users: list) -> list:
    logger = get_run_logger()
    cleaned = []
    for user in users:
        name_parts = user.get("name", "").split(" ", 1)
        first = name_parts[0] if name_parts else "Unknown"
        last = name_parts[1] if len(name_parts) > 1 else ""
        city = user.get("address", {}).get("city", "Unknown")
        cleaned.append({
            "user_id": user["id"],
            "first_name": first,
            "last_name": last,
            "email": user.get("email", "").lower().strip(),
            "city": city,
            "company": user.get("company", {}).get("name", "Unknown")
        })
    logger.info(f"Transformed {len(cleaned)} records")
    return cleaned

# -------------------------------------------------------
# LOAD
# -------------------------------------------------------
@task(name="load-to-sqlite")
def load_to_sqlite(records: list, db_path: str) -> int:
    logger = get_run_logger()
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS users (
            user_id INTEGER PRIMARY KEY,
            first_name TEXT,
            last_name TEXT,
            email TEXT UNIQUE,
            city TEXT,
            company TEXT
        )
    """)
    inserted = 0
    for rec in records:
        try:
            cur.execute("""
                INSERT OR REPLACE INTO users
                (user_id, first_name, last_name, email, city, company)
                VALUES (:user_id, :first_name, :last_name, :email, :city, :company)
            """, rec)
            inserted += 1
        except sqlite3.Error as err:
            logger.warning(f"Skipping record {rec['user_id']}: {err}")
    conn.commit()
    conn.close()
    logger.info(f"Loaded {inserted} records into {db_path}")
    return inserted

# -------------------------------------------------------
# FLOW
# -------------------------------------------------------
@flow(name="user-etl-pipeline", log_prints=True)
def user_etl_pipeline(
    limit: int = 10,
    db_path: str = "users.db"
) -> dict:
    raw_users = extract_users(limit)
    clean_users = transform_users(raw_users)
    count = load_to_sqlite(clean_users, db_path)
    summary = {"extracted": len(raw_users), "loaded": count, "db": db_path}
    print(f"Pipeline complete: {summary}")
    return summary

if __name__ == "__main__":
    result = user_etl_pipeline(limit=5, db_path="users.db")

Output:

14:20:01.101 | INFO    | Task run 'extract-users-0' - Extracting 5 users from API
14:20:01.512 | INFO    | Task run 'extract-users-0' - Extracted 5 user records
14:20:01.513 | INFO    | Task run 'transform-users-0' - Transformed 5 records
14:20:01.515 | INFO    | Task run 'load-to-sqlite-0' - Loaded 5 records into users.db
Pipeline complete: {'extracted': 5, 'loaded': 5, 'db': 'users.db'}

This pipeline demonstrates all the key Prefect patterns in one place: extraction with retries and caching (so re-running the pipeline within an hour skips the API call), defensive transformation (handling missing keys with .get() defaults), structured loading with per-record error handling, and a flow-level summary printed at the end. To extend it, replace the jsonplaceholder URL with your real data source, add a notification task at the end (email or Slack), and wrap it in a .serve() call to run it nightly.

Frequently Asked Questions

When should I use Prefect instead of Airflow?

Choose Prefect when you want to orchestrate Python workflows quickly without setting up a dedicated server or writing DAG configuration files. Prefect flows are plain Python functions decorated with @flow and @task, so the learning curve is minimal if you already know Python. Airflow is a better fit when your organization already runs it as a shared platform, or when you need its broad ecosystem of pre-built operators for Hadoop, Spark, and similar tools.

Can I run Prefect flows without starting the server?

Yes. Flows run locally without the server — just call your flow function like a normal Python function. The server at http://127.0.0.1:4200 is optional; it adds run history, log browsing, and scheduling visibility, but flows are fully functional without it. Start with local runs during development, then add the server when you want to monitor scheduled production runs.

How do I run tasks in parallel?

Submit multiple tasks concurrently using a ConcurrentTaskRunner or by calling tasks with .submit() inside a with block. By default, Prefect tasks run sequentially in the order they are called. For CPU-bound work, use a DaskTaskRunner from the prefect-dask package; for I/O-bound work, ConcurrentTaskRunner (the default when using .submit()) is usually enough and requires no extra dependencies.

Where does Prefect store cached results?

By default, Prefect stores task results in a local file-system cache in a folder called .prefect in the project directory. You can configure this to use cloud storage (AWS S3, GCS, Azure Blob) by setting a result storage block in the Prefect server settings. For most local pipelines the default storage works well; switch to cloud storage when multiple machines need to share results or when cached data should survive a machine restart.

How do I get notified when a flow fails?

Add an on_failure hook to your flow: @flow(on_failure=[send_alert]), where send_alert is a regular Python function that receives the flow run context. Inside that function you can call any notification library — send an email with smtplib, post to Slack via a webhook, or trigger a PagerDuty alert. The prefect-slack and prefect-email integrations in the Prefect extras provide ready-made notification blocks.

How do I pass secrets (API keys, passwords) to flows?

Never hardcode secrets in your flow code. For local development, use environment variables accessed via os.environ["API_KEY"]. For production Prefect deployments, use Prefect Blocks — encrypted secret stores that are created in the UI and referenced in code with Secret.load("my-api-key").get(). Blocks are stored encrypted in the Prefect backend and are never logged or exposed in run output.

Conclusion

Prefect transforms raw Python scripts into supervised, fault-tolerant pipelines with minimal changes to your existing code. In this article we covered the core building blocks: decorating functions with @task and @flow, configuring retries with exponential backoff, caching expensive task results with task_input_hash, capturing structured per-task logs with get_run_logger(), and scheduling automated runs with .serve(). The real-life ETL example showed how all these features combine into a production-ready pipeline.

A good next step is to extend the ETL pipeline: add a notification task at the end, swap jsonplaceholder.typicode.com for your real data source, and point the load step at a PostgreSQL or Snowflake target. Once that is working locally, wrap the flow in a .serve() call with a cron schedule and let it run unattended. The Prefect UI will show you every run, every log line, and every retry — turning what used to be a fragile cron job into something you can actually trust.

The official Prefect documentation at docs.prefect.io covers advanced features including deployments on Prefect Cloud, work pools for distributed execution, and integrations with AWS, GCP, and dbt. The concepts from this article transfer directly to those guides.

Related Articles

• How To Use Python celery-beat for Periodic Task Scheduling
• How To Use Python Tenacity for Retry Logic
• How To Use Python socketio for Real-Time Web Events