data-pipeline-orchestration

Data Pipeline Orchestration

Overview

A pipeline runs unattended, gets retried, gets backfilled for last March, and runs twice when someone clicks the button twice. Each stage therefore has to produce the same result for the same inputs no matter how often it runs, and fail loudly when its input is wrong.

Core principle: business logic lives in plain, tested functions; the orchestrator layer is a thin wrapper that wires partitions, resources and retries around them.

1. Detect the orchestrator — and keep it

You find	It is
@dg.asset, Definitions(...), dagster.yaml, workspace.yaml	Dagster (asset-based)
@flow, @task from prefect, prefect.yaml	Prefect
dags/ folder, @dag / DAG(...), from airflow	Airflow (Airflow 3 imports authoring APIs from airflow.sdk)
dvc.yaml stages	DVC pipeline (experiment-tracking-and-reproducibility)
a cron entry or a CI schedule calling a CLI	plain entrypoint

Never add a second orchestrator (rule stack-choice). A plain CLI entrypoint called by the existing scheduler is often the right size.

2. Thin orchestration, testable logic

# src/churn/features.py — pure, tested with fixtures
def build_daily_features(orders: pd.DataFrame, day: date) -> pd.DataFrame: ...

# src/churn/orchestration/assets.py — wiring only
import dagster as dg

daily = dg.DailyPartitionsDefinition(start_date="2025-01-01")

@dg.asset(partitions_def=daily)
def daily_features(context: dg.AssetExecutionContext, warehouse: WarehouseResource) -> None:
    day = date.fromisoformat(context.partition_key)
    features = build_daily_features(warehouse.orders_for(day), day)
    warehouse.overwrite_partition("features.daily", day, features)

The asset never computes anything itself; build_daily_features has unit tests that know nothing about Dagster.

3. Idempotency

❌	✅
INSERT INTO features SELECT ... on every run	delete-then-insert the partition in one transaction, MERGE on the key, or INSERT OVERWRITE the partition
df.to_parquet("out/features.parquet", append...)	write out/day=2026-10-01/part.parquet to a temp path, then rename
window_end = datetime.now()	window from the partition key or the run's data interval
file names with a timestamp of when it ran	paths keyed by what the data is for (partition, model version)