Migrate from Prefect to Mage Pro

​

Why migrate from Prefect to Mage Pro?

• Visual pipeline development and debugging
• Native data integration (ETL/ELT workflows)
• Mixing SQL and Python naturally in the same pipeline
• Streaming or event-driven data processing
• Built-in data previews and interactive development
• AI-powered pipeline development

​

Mage Pro vs Prefect: Benefits Overview

​

Step-by-Step Migration Instructions

​

Mapping Prefect Concepts to Mage Pro

​

Flows → Pipelines

from prefect import flow, task

@flow(name="my-etl-flow")
def my_etl_flow():
    data = extract_data()
    transformed = transform_data(data)
    load_data(transformed)

• Create a new pipeline in the UI
• Add blocks visually or via code
• Dependencies are defined by connecting blocks in the UI

​

Tasks → Blocks

@task
def extract_data():
    return pd.read_csv("data.csv")

@task
def transform_data(df):
    return df.dropna()

@task
def load_data(df):
    df.to_sql("table", engine)

• extract_data → Data loader block (Python or connector)
• transform_data → Transformer block (Python, SQL, or dbt block)
• load_data → Data exporter block (Python or connector)

​

Prefect Blocks → Mage Blocks

​

Scheduling & Deployments

from prefect import serve
from prefect.schedules import CronSchedule

@flow
def scheduled_flow():
    ...

if __name__ == "__main__":
    scheduled_flow.serve(
        cron=CronSchedule(cron="0 0 * * *")
    )

• Create a schedule trigger in the pipeline UI
• Configure cron expression, interval, or event-based triggers
• Use the Deployment App to manage deployments across environments (dev, staging, prod)
• No complex deployment setup needed—triggers and deployments are configured via UI

​

Secrets & Configuration

from prefect.blocks.system import Secret

@task
def use_secret():
    api_key = Secret.load("my-api-key")

• Use workspace variables or pipeline variables
• Access via variables dictionary in blocks
• Built-in secret manager with encryption, or integrate with external secret managers (AWS Secrets Manager, HashiCorp Vault, etc.)
• Secrets are isolated per workspace and pipeline

​

Convert Prefect Flows with AI Sidekick

​

How to Use It

1. Click the “Ask AI” button in the top-right corner of the Mage Pro UI.
2. Paste your Prefect flow code — including @flow, @task, or flow-based workflows.
3. Ask: “Convert this Prefect flow to a Mage pipeline.”
4. Sidekick will:
   • Parse the flow structure and task relationships
   • Generate the corresponding Mage blocks (data loader, transformer, data exporter)
   • Define block dependencies and scheduling logic
   • Convert Prefect blocks to Mage connectors where applicable
   • Optimize the generated code for performance and best practices
5. Review and insert the generated pipeline directly into your project.

​

Why Use Sidekick?

• Faster migration: turn entire flows into Mage pipelines in seconds
• Less error-prone: Sidekick understands scheduling, dependencies, and task types
• Code optimization: automatically optimizes generated code for performance
• Context-aware: uses your project setup and prior block structure
• Fully editable: review, tweak, and insert blocks before saving

​

Tips for Migrating Complex Flows

​

Handling Prefect Subflows

@flow
def parent_flow():
    result = subflow()

@flow
def subflow():
    ...

• Split into separate pipelines and trigger them via API or use dynamic blocks for conditional logic
• Use block groups to organize related blocks within a pipeline

​

Prefect Task Runners & Concurrency

from prefect.task_runners import ConcurrentTaskRunner

@flow(task_runner=ConcurrentTaskRunner())
def parallel_flow():
    ...

• Configure concurrency settings in pipeline configuration
• Blocks with no dependencies run in parallel automatically
• Use executor configuration for custom parallelism on Kubernetes or Docker

​

Prefect Retries & Error Handling

@task(retries=3, retry_delay_seconds=60)
def risky_task():
    ...

• Configure retry settings in block or pipeline configuration
• Set retry count, delay, and backoff strategy
• Built-in error handling and alerting

​

Prefect Caching

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def cached_task():
    ...

• Use block output caching in pipeline settings
• Configure cache expiration and invalidation rules
• Preview cached outputs in the UI

​

After Migration: What You Get with Mage Pro

• Unified data platform: orchestration + data integration + transformation + streaming in one tool
• Visual pipeline builder with real-time data previews
• AI-powered block generation, optimization, and debugging (via AI Sidekick)
• Git-backed version control and CI/CD with UI-based Deployment App
• Built-in access controls, audit logs, and workspace isolation
• Support for SQL, Python, R, streaming, dbt blocks, Delta Lake, Iceberg, and more
• 200+ native connectors for databases, APIs, and cloud storage
• Streaming pipelines for real-time data processing
• Flexible secrets management: built-in secret manager or integrate with external systems (AWS Secrets Manager, Vault, etc.)
• Auto-scaled executors on Kubernetes or Docker
• Transparent pricing: usage-based pricing (SaaS) or cluster/workspace license-based pricing (self-hosted)

Prefect excels at orchestration. But when you need data engineering, visual development, and AI assistance — Mage Pro is built for modern data teams.