Build a Data Pipeline with Prefect

Data pipelines move data from one place to another. They extract, transform, and load data. Doing this by hand is slow and error prone. Prefect makes it easier.

Prefect is a modern workflow orchestration tool for Python. It lets you define tasks and flows. It also handles retries, scheduling, logging, and monitoring. This guide shows you how to build a clean data pipeline with Prefect.

Why Use Prefect for Data Pipelines

Prefect is written in Python. You do not need YAML or complex config files. You just write Python functions.

Prefect gives you automatic retries, task caching, and observability out of the box. You can see every run in a UI. You can also set schedules and alerts.

It works well with tools you already use. For example, you can connect to Kafka. If you want to produce and consume Kafka messages in Python, Prefect can wrap those steps as tasks. You can also install kafka-python in Python and call it inside a flow.

Install Prefect

Start by installing Prefect with pip. Use a virtual environment to keep things clean.

pip install prefect

Check the version to make sure the install worked.

prefect version

Version:             3.x.x
API version:         0.8.x
Python version:      3.11.x

You can run flows locally without a server. That is perfect for learning and small jobs.

Understand Tasks and Flows

Prefect has two main building blocks. A task is a single unit of work. A flow is a container that runs tasks in order.

Use the @task decorator for tasks. Use the @flow decorator for flows. Tasks can return values. Flows pass those values between tasks.

This separation is useful. Each task can retry on its own. Each task can be cached. Each task shows up as a step in the UI.

Build Your First Flow

Let us build a small pipeline. It fetches data, cleans it, and saves it. We use simple Python to keep the focus on Prefect.

from prefect import flow, task

@task
def extract():
    # Simulate pulling raw data
    data = [1, 2, 3, 4, 5]
    print(f"Extracted {len(data)} records")
    return data

@task
def transform(data):
    # Clean and scale the data
    cleaned = [x * 10 for x in data]
    print(f"Transformed {len(cleaned)} records")
    return cleaned

@task
def load(data):
    # Simulate saving the data
    print(f"Loaded {len(data)} records")
    return True

@flow(name="simple-etl")
def etl_flow():
    raw = extract()
    clean = transform(raw)
    load(clean)

if __name__ == "__main__":
    etl_flow()

Run the script from your terminal.

python etl.py

Extracted 5 records
Transformed 5 records
Loaded 5 records

That is a working pipeline. Each step is a task. The flow wires them together.

Add Retries and Caching

Real pipelines fail. Networks drop. APIs time out. Prefect lets you retry tasks without extra code.

Pass retries and retry_delay_seconds to the task decorator. The task will retry on failure. You can also cache results with cache_key_fn.

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(retries=3, retry_delay_seconds=5)
def fetch_from_api():
    # This may fail and retry automatically
    return {"status": "ok"}

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def expensive_transform(payload):
    # Cache the result for one hour
    return payload["status"].upper()

@flow
def retry_flow():
    payload = fetch_from_api()
    result = expensive_transform(payload)
    print(result)

if __name__ == "__main__":
    retry_flow()

OK

Retries and caching save time. They also make your pipeline more resilient.

Use Logging and Parameters

Prefect gives every flow a logger. Use get_run_logger to log messages. These logs appear in the UI and in your terminal.

Flows can also take parameters. This makes them reusable. You can trigger the same flow with different inputs.

from prefect import flow, task, get_run_logger

@task
def process(batch_size):
    logger = get_run_logger()
    logger.info(f"Processing batch of {batch_size}")
    return batch_size * 2

@flow
def parameterized_flow(batch_size: int = 10):
    result = process(batch_size)
    get_run_logger().info(f"Result: {result}")
    return result

if __name__ == "__main__":
    parameterized_flow(batch_size=25)

25:00:01.123 | INFO | Processing batch of 25
25:00:01.130 | INFO | Result: 50

Parameters make testing easy. You can run small batches first, then scale up.

Connect Real Data Sources

Most pipelines touch a database or a message queue. Prefect does not replace those tools. It orchestrates them.

For databases, you can query Oracle Database from Python inside a task. You can also install oracledb in Python and open a connection in your flow. If you use MySQL, you can install mysqlclient in Python and run queries the same way.

For streaming data, you can install pika in Python and read from RabbitMQ. Tasks can also wrap Kafka producers and consumers. Each step stays small and testable.

For text processing, you can add a task that uses tokenization with NLTK in Python. For structured data, you can compile and use .proto files in Python to parse messages. You can also install protobuf in Python when your pipeline reads binary payloads.

Prefect also pairs well with AI steps. You can call OpenAI function calling in Python from a task to enrich records. The flow keeps track of every call.

Schedule Your Pipeline

Pipelines should run on time. Prefect supports cron and interval schedules. You can set a schedule in code or in the UI.

from prefect import flow
from prefect.schedules import Cron

@flow
def daily_report():
    print("Running daily report")

if __name__ == "__main__":
    # Run every day at 6 AM
    daily_report.serve(name="daily-report", schedule=Cron("0 6 * * *"))

The serve method keeps the flow running. It checks the schedule and triggers runs. You can leave it running on a server or a small VM.

Monitor and Debug Runs

Start the Prefect UI to see your runs. Run this command in a terminal.

prefect server start

Open the URL shown in the terminal. You will see flows, tasks, and logs. You can click a failed run and read the traceback.

This visibility matters. Most pipeline bugs are data bugs. Good logs help you find them fast.

Best Practices

Keep tasks small. One task should do one thing. This makes retries cheap and debugging simple.

Pass data between tasks with return values. Avoid global state. It breaks caching and makes runs hard to reproduce.

Use idempotent tasks. If a task runs twice, the result should be the same. This is critical for retries.

Test tasks on their own. Because they are plain Python functions, you can call them directly in unit tests.

Add timeouts to network tasks. A stuck call can block a whole flow. Prefect lets you set a timeout on each task.

Conclusion

Prefect makes data pipelines simple and reliable. You write normal Python functions. Prefect adds retries, caching, logging, scheduling, and a UI.

Start small. Build a flow with three tasks. Add retries. Add a schedule. Then connect real sources like databases, message queues, and APIs.

With a few clean tasks and flows, you get a pipeline that is easy to read, easy to test, and easy to trust.