Choose and implement data pipeline orchestration. Covers Airflow, Dagster, Prefect, DAG design, task dependencies, error handling, scheduling, and operational best practices.
This reference covers how to author, deploy and monitor data pipelines using three major orchestration platforms: Apache Airflow, Dagster, and Prefect. These systems manage dependencies, scheduling, execution state, and error recovery for workflows composed of discrete tasks. The patterns described here apply broadly to orchestration frameworks that support DAG-based workflow definition.
Each platform requires defining a directed acyclic graph (DAG) structure where nodes represent tasks and edges define execution dependencies.
In Airflow, a DAG is defined using Python classes:
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
dag = DAG(
'example_dag',
default_args={
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2023, 1, 1),
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
},
description='An example DAG',
schedule_interval=timedelta(days=1),
catchup=False,
)
task1 = BashOperator(
task_id='print_date',
bash_command='date',
dag=dag,
)Prefect uses a similar approach with Python functions:
from prefect import flow, task
@task
def print_date():
return "date"
@flow
def example_flow():
result = print_date()
return resultDagster uses a more explicit structure through Python classes:
from dagster import pipeline, solid
@solid
def print_date(context):
return "date"
@pipeline
def example_pipeline():
print_date()All platforms support defining dependencies between tasks. In Airflow, this happens at task definition time using >> or << operators:
task1 >> task2 >> task3Prefect defines upstream dependencies in the function signature:
@task
def process_data():
return "processed"
@task
def load_data(input_data):
return f"loaded {input_data}"
@flow
def example_flow():
data = process_data()
result = load_data(data)Dagster uses a more explicit dependency mechanism:
from dagster import graph
@graph
def example_graph():
data = print_date()
return load_data(data)Manual execution of pipelines works differently across platforms. In Airflow, you trigger a DAG run through the web UI or CLI:
airflow dags trigger example_dagPrefect allows triggering via CLI:
prefect flow run example_flowDagster supports both UI and CLI execution:
dagster pipeline execute -f example_pipeline.pyAirflow schedules workflows using schedule_interval in the DAG definition. The system uses a separate scheduler process that polls for jobs to run based on their cron expressions or timedelta intervals.
Prefect uses a similar concept with its schedule parameter:
from prefect import flow
from prefect.schedules import IntervalSchedule
@flow(schedule=IntervalSchedule(interval=timedelta(days=1)))
def example_flow():
passDagster manages scheduling through the schedule argument in pipeline definitions, typically using cron expressions or timedelta values.
All platforms support limiting concurrent task execution. Airflow uses max_active_tasks and max_active_runs parameters on DAGs:
dag = DAG(
'example_dag',
max_active_tasks=10,
max_active_runs=5,
)Prefect supports concurrency control through work pools:
from prefect.workers import ProcessWorker
# Configure worker with limited concurrent runs
worker = ProcessWorker(max_concurrent_runs=3)Dagster manages resource constraints through its execution engine, which can be configured to limit parallelism.
Airflow exposes execution state through a web UI at /admin and command-line interface:
airflow tasks list example_dag
airflow tasks state example_dag print_datePrefect provides a dashboard and CLI commands:
prefect flow run example_flow --watch
prefect flow run example_flow --logsDagster shows execution status through its UI and CLI:
dagster pipeline list
dagster pipeline execute -f example_pipeline.pyLogs are typically accessible via the platform's web UI or command-line tools. Airflow stores logs in a configured location, accessible via the web interface or CLI:
airflow tasks log example_dag print_datePrefect writes logs to stdout and can be configured for external logging systems:
from prefect import flow
@flow(log_prints=True)
def example_flow():
print("This will appear in logs")Dagster provides structured logging through its execution context.
Airflow retries failed tasks based on retries and retry_delay parameters:
task1 = BashOperator(
task_id='print_date',
bash_command='date',
retries=3,
retry_delay=timedelta(minutes=10),
)Prefect handles retries through its built-in retry mechanism:
@task(retries=3, retry_delay_seconds=600)
def example_task():
passDagster manages retries through task definitions:
@solid(
max_retries=3,
retry_delay=timedelta(minutes=10)
)
def example_solid(context):
passAirflow requires installing the apache-airflow package:
pip install apache-airflowPrefect uses prefect:
pip install prefectDagster uses dagster:
pip install dagsterAirflow typically requires a webserver, scheduler and database:
airflow webserver --port 8080
airflow schedulerPrefect runs with a server process:
prefect server startDagster uses a CLI for starting its service:
dagster webserverEach platform requires configuration through environment variables. Airflow uses AIRFLOW_HOME and AIRFLOW__CORE__SQL_ALCHEMY_CONN:
export AIRFLOW_HOME=/opt/airflow
export AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql://user:pass@host:port/dbPrefect sets configuration through PREFECT_API_URL and related environment variables.
Dagster uses DAGSTER_HOME and database connection strings in its config files.
All platforms store task execution state in databases. When this fails, tasks can appear to run multiple times or lose their execution history. In Airflow, if the metadata database becomes unavailable, tasks may be marked as failed even when they succeeded.
Dependencies that are not properly defined cause workflows to run incorrectly. In Airflow, an undefined dependency can lead to a task running before its prerequisites complete. Prefect's dynamic dependencies can silently fail if upstream tasks don't return values correctly.
Concurrency limits can cause deadlocks or excessive resource consumption. When maximum active tasks is set too low, pipelines may stall waiting for resources. Airflow's scheduler may not start new runs if all slots are taken.
Misconfigured DAG files cause execution failures. In Airflow, a syntax error in a DAG definition prevents the DAG from being loaded, which manifests as "No DAGs found" messages in the web UI. Prefect and Dagster also have similar behaviors where configuration errors block execution.
Timezone differences between systems can cause scheduling issues. Airflow uses start_date and timezone-aware datetime objects, which must be consistent across all systems. Incorrect timezone settings lead to missed or double-executions.
All platforms support task timeouts but may not handle them consistently. If a timeout occurs, the platform should mark the task as failed appropriately. In Airflow, tasks that exceed their configured timeout may leave resources in an inconsistent state.
Platforms support generating tasks dynamically at runtime. Airflow uses PythonOperator with dynamic task creation:
def generate_tasks():
for i in range(5):
BashOperator(
task_id=f'task_{i}',
bash_command=f'echo {i}'
)Prefect supports dynamic task generation through loops and conditionals:
@flow
def example_flow():
tasks = []
for i in range(5):
task = process_item(i)
tasks.append(task)Dagster generates tasks through runtime graph construction:
@graph
def example_graph():
tasks = []
for i in range(5):
task = process_item(i)
tasks.append(task)Data movement between platforms requires careful handling. Airflow often uses sensors to wait for external events, while Prefect and Dagster have built-in mechanisms for inter-platform communication through shared state or messaging systems.
Platforms support different recovery strategies. Airflow allows task-level recovery through retries and callbacks, Prefect provides automatic recovery mechanisms, and Dagster supports detailed error handling in its execution model.
Airflow's webserver may become unresponsive if the metadata database is under stress. The scheduler can also become unresponsive if it encounters DAG parsing errors that prevent it from starting new runs.
Prefect's flow execution model depends heavily on its API server being available. If the Prefect API becomes unavailable, flows may not register properly or may fail to report their status correctly.
Dagster's execution engine may have issues with complex dependency graphs, particularly when tasks have multiple upstream dependencies that must be satisfied in specific order.
This page was rewritten on 16 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy