Data Orchestration

Data Orchestration

Routes data pipeline and orchestration tasks to the right tool based on scope and data movement pattern.

Routing Table

dlt — Lightweight Data Loading

# pip install dlt

import dlt

# Load from REST API to DuckDB
@dlt.resource
def github_events():
    import requests
    response = requests.get("https://api.github.com/events")
    yield from response.json()

pipeline = dlt.pipeline(
    pipeline_name="github",
    destination="duckdb",
    dataset_name="events"
)

load_info = pipeline.run(github_events())
print(load_info)

Destinations: DuckDB, BigQuery, Snowflake, Redshift, Postgres, Filesystem, REST API Sources: REST API, SQL databases, cloud storage, custom Python generators

# Auto-infer schema from REST API
import dlt
from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {"base_url": "https://api.example.com"},
    "resources": [
        {"name": "users", "endpoint": "/users"},
        {"name": "orders", "endpoint": "/orders"}
    ]
})

pipeline = dlt.pipeline(destination="bigquery", dataset_name="app_data")
pipeline.run(source)

dagster — Asset-Based Pipeline Orchestration

# pip install dagster dagster-webserver

from dagster import asset, define_asset_job, Definitions

@asset
def raw_orders():
    """Fetch raw orders from source system."""
    return fetch_orders_from_api()

@asset
def cleaned_orders(raw_orders):
    """Clean and validate order data."""
    df = pd.DataFrame(raw_orders)
    return df.dropna().reset_index(drop=True)

@asset
def orders_report(cleaned_orders):
    """Generate summary report."""
    return cleaned_orders.groupby("status").size().to_dict()

defs = Definitions(
    assets=[raw_orders, cleaned_orders, orders_report],
    jobs=[define_asset_job("orders_pipeline")]
)

# Start Dagster UI
dagster dev  # opens at localhost:3000

# Run job from CLI
dagster job execute -j orders_pipeline

Dagster strengths: Asset lineage graph, data catalog, partitioned backfills, built-in sensors/schedules, software-defined assets

prefect — Python-Native Workflow Orchestration

# pip install prefect

from prefect import flow, task
from prefect.deployments import Deployment

@task(retries=3, retry_delay_seconds=10)
def extract_data(source_url: str) -> list:
    return requests.get(source_url).json()

@task
def transform_data(data: list) -> list:
    return [item for item in data if item.get("active")]

@task
def load_data(data: list, target: str) -> None:
    db.insert_many(target, data)

@flow(name="etl-pipeline", log_prints=True)
def etl_pipeline(source_url: str, target: str):
    raw = extract_data(source_url)
    clean = transform_data(raw)
    load_data(clean, target)

if __name__ == "__main__":
    etl_pipeline("https://api.example.com/data", "users")

# Start Prefect server
prefect server start  # localhost:4200

# Deploy flow
prefect deploy --name prod-etl

dbt-core — SQL Transformation Layer

# pip install dbt-core dbt-postgres  (or dbt-bigquery, dbt-snowflake)

# Initialize project
dbt init my_project
cd my_project

# Configure profiles.yml (~/.dbt/profiles.yml)
# my_project:
#   target: dev
#   outputs:
#     dev:
#       type: postgres
#       host: localhost
#       database: analytics
#       schema: dbt_dev

-- models/staging/stg_orders.sql
SELECT
    order_id,
    customer_id,
    status,
    created_at::date AS order_date
FROM {{ source('raw', 'orders') }}
WHERE status != 'cancelled'

-- models/marts/orders_summary.sql
SELECT
    order_date,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM {{ ref('stg_orders') }}
GROUP BY 1
ORDER BY 1

dbt run              # run all models
dbt run --select stg_orders   # run specific model
dbt test             # run data quality tests
dbt docs generate && dbt docs serve  # data catalog UI

bytewax — Real-Time Stream Processing

# pip install bytewax

from bytewax.dataflow import Dataflow
from bytewax.connectors.kafka import KafkaSource
from bytewax.connectors.stdio import StdOutSink

flow = Dataflow("order-processor")

# Consume from Kafka
stream = flow.input("kafka-in", KafkaSource(
    brokers=["localhost:9092"],
    topics=["orders"]
))

# Transform
def parse_order(msg):
    import json
    data = json.loads(msg.value)
    return (data["customer_id"], data["amount"])

def aggregate_by_customer(customer_id, amounts):
    return (customer_id, sum(amounts))

stream = stream.map("parse", parse_order)
stream = stream.reduce_window("aggregate", aggregate_by_customer)
stream.output("stdout", StdOutSink())

# Run
from bytewax.run import cli_main
cli_main(flow)

airbyte — Open-Source ELT Connectors

# Docker Compose (local)
git clone https://github.com/airbytehq/airbyte.git
cd airbyte
./run-ab-platform.sh
# Access at http://localhost:8000

# Or use Airbyte Cloud

Airbyte API:

import airbyte as ab

# Use a pre-built source connector
source = ab.get_source(
    "source-github",
    config={"credentials": {"personal_access_token": "$GITHUB_TOKEN"}, "repositories": ["myorg/myrepo"]}
)

# Read into pandas
result = source.read()
df = result["issues"].to_pandas()

Available connectors: Postgres, MySQL, Snowflake, BigQuery, S3, GitHub, Stripe, Salesforce, Google Sheets, HubSpot, Shopify (300+ connectors)

Decision Guide

"Load data from an API into a warehouse" → dlt (simplest) "Build a full data platform with asset catalog" → dagster "Orchestrate Python functions with retries/scheduling" → prefect "Transform data in a warehouse with SQL" → dbt-core "Process Kafka/event streams in real-time" → bytewax "Sync data between SaaS tools with pre-built connectors" → airbyte

Common Patterns

dlt + dbt (popular combo)

# 1. Load with dlt
pipeline = dlt.pipeline(destination="bigquery", dataset_name="raw")
pipeline.run(source)

# 2. Transform with dbt
# (run dbt after load in same orchestration job)
import subprocess
subprocess.run(["dbt", "run", "--select", "marts+"])

Prefect + dbt

from prefect_dbt.cli.commands import DbtCoreOperation

@flow
def dbt_flow():
    DbtCoreOperation(commands=["dbt run", "dbt test"]).run()

Environment Variables

# dlt destinations
DESTINATION__BIGQUERY__CREDENTIALS_FILE=credentials.json
DESTINATION__POSTGRES__CREDENTIALS=postgresql://user:pass@host/db

# Prefect
PREFECT_API_URL=http://localhost:4200/api

# dbt profiles
DBT_PROFILES_DIR=~/.dbt/

# Airbyte
AIRBYTE_API_KEY=          # for Airbyte Cloud