🧠 Fundamentals

🎯 What is Modern Orchestration?

Modern orchestration tools like Dagster and Prefect are the next generation of workflow orchestration. They improve on Airflow with data-aware scheduling, dynamic workflows, native Python, better testing, and developer experience.

Why Modern Orchestration Matters

Data-aware: Track assets. Dynamic: Runtime DAGs. Python-native: No DSL. Testable: Unit tests. Observable: Rich UI. Modern: Better DX.

📌 Data-Aware📌 Dynamic📌 Python-Native📌 Observable
# Modern Orchestration — the big picture # Dagster and Prefect improve on Airflow. # CONCEPTUAL OVERVIEW print(""" === Modern Orchestration === Tools: - Dagster - Prefect Key ideas: - Data-aware: Track assets, not just tasks - Dynamic: DAGs at runtime - Python-native: Pure Python, no DSL - Testable: Unit tests - Observable: Rich UI, lineage - Developer-friendly: Local development Compared to Airflow: - More modern - Better DX - Data-aware - Dynamic - Testable """) # SIMPLE COMPARISON print("=== Simple Comparison ===") print(""" Airflow: - Task-centric - Static DAGs - Cron scheduling - Python operators - Mature Dagster: - Asset-centric - Software-defined assets - Data-aware scheduling - Type safety - Modern Prefect: - Flow-centric - Dynamic workflows - Python-native - Modern - Easy local dev """) # WHY IT MATTERS print("=== Why It Matters ===") print(""" Problems with Airflow: - Static DAGs - Task-centric (not data-centric) - Hard to test - Top-level code issues - Complex scheduling Modern tools solve: - Dynamic DAGs - Asset tracking - Easy testing - Pure Python - Data-aware """) # KEY CONCEPTS print("=== Key Concepts ===") print(""" Dagster: - Software-Defined Asset (SDA) - Asset graph - Ops (operations) - Jobs - Schedules - Sensors - Resources - IO managers Prefect: - Flow - Task - Deployment - Work pool - Block - Notification - Automations """)

🪐 Dagster

🎯 The Data-Aware Orchestrator

Dagster is a modern data orchestrator built around Software-Defined Assets (SDAs). Instead of defining tasks, you define the data assets your pipeline produces, and Dagster figures out the execution graph.

Dagster Characteristics

Asset-centric: Define data, not tasks. Type-safe: Python types. Testable: Unit tests. Data-aware: Track lineage. Modern UI: Rich observability. Python: Pure Python.

🪐 Assets🪐 Type-Safe🪐 Testable🪐 Data-Aware
# Dagster — deep dive print("=== What is Dagster? ===") print(""" Dagster: - Data orchestrator - Software-Defined Assets - Type-safe - Testable - Data-aware - Modern UI Created by: - Nick Schrock (GraphQL co-creator) - 2018 Key concepts: - Asset: Data artifact - Op: Operation - Job: Collection of ops - Graph: DAG of ops - Schedule: Cron - Sensor: Event-based - Resource: External system - IO Manager: Storage """) print("\n=== Software-Defined Asset ===") print(""" Asset: - Data artifact - Table, file, ML model - Declarative - Tracked Example: from dagster import asset @asset def raw_users(): return extract_users() @asset def cleaned_users(raw_users): return clean(raw_users) @asset def user_metrics(cleaned_users): return compute_metrics(cleaned_users) # Dagster builds the DAG automatically! """) print("\n=== Full Example ===") print(""" from dagster import asset, Definitions, ScheduleDefinition import pandas as pd @asset def raw_orders(): return pd.read_csv("s3://bucket/orders.csv") @asset def cleaned_orders(raw_orders): return raw_orders.dropna() @asset def daily_revenue(cleaned_orders): return cleaned_orders.groupby("date")["amount"].sum() # Define job daily_job = define_asset_job("daily_job", selection="*") # Schedule daily_schedule = ScheduleDefinition( job=daily_job, cron_schedule="0 0 * * *", ) defs = Definitions( assets=[raw_orders, cleaned_orders, daily_revenue], schedules=[daily_schedule], ) """) print("\n=== Dagster Pros & Cons ===") print(""" PROS: ✓ Data-aware ✓ Type-safe ✓ Testable ✓ Modern UI ✓ Asset lineage ✓ Python-native ✓ Great DX CONS: ✗ Newer ✗ Smaller community ✗ Steeper learning curve ✗ Different paradigm ✗ Less mature """)

🌊 Prefect

🎯 The Dynamic Orchestrator

Prefect is a modern workflow orchestration tool built around flows and tasks. It's Python-native, supports dynamic workflows, and is easy to develop and deploy.

Prefect Characteristics

Flows: Workflows. Tasks: Units of work. Dynamic: Runtime DAGs. Python: Pure Python. Easy dev: Local-first. Modern UI: Observability.

🌊 Flows🌊 Tasks🌊 Dynamic🌊 Python-Native
# Prefect — deep dive print("=== What is Prefect? ===") print(""" Prefect: - Workflow orchestration - Flows and Tasks - Dynamic workflows - Python-native - Modern UI - Easy local dev Created by: - Jeremiah Lowin - 2018 Key concepts: - Flow: Workflow - Task: Unit of work - Deployment: Run configuration - Work pool: Infrastructure - Block: Configuration - Notification: Alerts - Automation: Triggers """) print("\n=== Flow & Task ===") print(""" Flow: - Workflow definition - Python function - @flow decorator Task: - Unit of work - Python function - @task decorator Example: from prefect import flow, task @task def extract(): return [1, 2, 3] @task def transform(data): return [x * 2 for x in data] @task def load(data): print(f"Loaded {len(data)} records") @flow def etl(): data = extract() transformed = transform(data) load(transformed) etl() """) print("\n=== Dynamic Workflows ===") print(""" Dynamic: - Runtime DAGs - Conditional logic - Loops - Map over lists Example: @flow def process_files(files): for file in files: process(file) # Dynamic! @flow def conditional(should_run): if should_run: task_a() else: task_b() """) print("\n=== Prefect Pros & Cons ===") print(""" PROS: ✓ Python-native ✓ Dynamic workflows ✓ Easy local dev ✓ Modern UI ✓ Good DX ✓ Flexible ✓ Async support CONS: ✗ Newer ✗ Smaller community ✗ Less mature ✗ Different paradigm ✗ Prefect 2.0 change """)

⚖️ vs Airflow

🎯 Modern vs Traditional

Dagster and Prefect improve on Airflow with data-aware scheduling, dynamic workflows, and better developer experience. But Airflow is more mature with a larger ecosystem.

Comparison

Airflow: Mature, task-centric. Dagster: Asset-centric. Prefect: Flow-centric. Airflow: Static DAGs. Modern: Dynamic. All: Python.

⚖️ Airflow⚖️ Dagster⚖️ Prefect⚖️ Modern
# Modern vs Airflow print("=== Comparison ===") print(f"{'Aspect':<25} {'Airflow':<20} {'Dagster':<20} {'Prefect':<20}") print("-" * 85) print(f"{'Paradigm':<25} {'Task':<20} {'Asset':<20} {'Flow':<20}") print(f"{'DAG definition':<25} {'Static':<20} {'Asset graph':<20} {'Dynamic':<20}") print(f"{'Data-aware':<25} {'No':<20} {'Yes':<20} {'Partial':<20}") print(f"{'Type-safe':<25} {'No':<20} {'Yes':<20} {'Partial':<20}") print(f"{'Testing':<25} {'Hard':<20} {'Easy':<20} {'Easy':<20}") print(f"{'Local dev':<25} {'OK':<20} {'Good':<20} {'Excellent':<20}") print(f"{'Dynamic':<25} {'Limited':<20} {'Good':<20} {'Excellent':<20}") print(f"{'UI':<25} {'Mature':<20} {'Modern':<20} {'Modern':<20}") print(f"{'Maturity':<25} {'Mature':<20} {'Modern':<20} {'Modern':<20}") print(f"{'Community':<25} {'Largest':<20} {'Growing':<20} {'Growing':<20}") print("\n=== When to Use Which ===") print(""" AIRFLOW: ✓ Mature ecosystem ✓ Large community ✓ Many integrations ✓ Enterprise adoption ✓ Task-centric workflows DAGSTER: ✓ Data-aware pipelines ✓ Asset lineage ✓ Type safety ✓ Modern DX ✓ Testing PREFECT: ✓ Dynamic workflows ✓ Python-native ✓ Easy local dev ✓ Modern UI ✓ Flexible """) print("\n=== Feature Comparison ===") print(f"{'Feature':<25} {'Airflow':<15} {'Dagster':<15} {'Prefect':<15}") print("-" * 70) print(f"{'DAGs':<25} {'Yes':<15} {'Yes':<15} {'Yes':<15}") print(f"{'Python':<25} {'Yes':<15} {'Yes':<15} {'Yes':<15}") print(f"{'Assets':<25} {'No':<15} {'Yes':<15} {'No':<15}") print(f"{'Dynamic':<25} {'Limited':<15} {'Good':<15} {'Excellent':<15}") print(f"{'Type-safe':<25} {'No':<15} {'Yes':<15} {'Partial':<15}") print(f"{'Testing':<25} {'Hard':<15} {'Easy':<15} {'Easy':<15}") print(f"{'Local dev':<25} {'OK':<15} {'Good':<15} {'Excellent':<15}") print(f"{'UI':<25} {'Mature':<15} {'Modern':<15} {'Modern':<15}")

📦 Software-Defined Assets

🎯 The Dagster Paradigm

Software-Defined Assets (SDAs) are Dagster's core abstraction. Instead of defining tasks that do things, you define assets that are things — tables, files, ML models. Dagster builds the execution graph from the asset dependencies.

Software-Defined Assets

Asset: Data artifact. Declarative: Define what, not how. Dependencies: Auto-detected. Lineage: Tracked. Materialize: Produce asset. Partition: Split asset.

📦 Asset📦 Declarative📦 Dependencies📦 Lineage
# Software-Defined Assets print("=== What is an Asset? ===") print(""" Asset: - Data artifact - Table, file, model - Declarative - Tracked - Versioned Example: from dagster import asset @asset def raw_users(): return extract_users() @asset def cleaned_users(raw_users): return clean(raw_users) @asset def user_metrics(cleaned_users): return compute_metrics(cleaned_users) # Dagster builds DAG automatically! """) print("\n=== Asset Dependencies ===") print(""" Dependencies: - Function parameters - Auto-detected - Asset graph - Lineage Example: @asset def daily_revenue(cleaned_orders): # cleaned_orders is a dependency return cleaned_orders.groupby("date")["amount"].sum() """) print("\n=== Asset Materialization ===") print(""" Materialize: - Produce asset - Run computation - Store result Manual: materialize([daily_revenue]) Scheduled: daily_job = define_asset_job("daily_job", selection="*") ScheduleDefinition(job=daily_job, cron_schedule="0 0 * * *") Sensor: @sensor(job=daily_job) def new_data_sensor(context): if new_data_available(): yield RunRequest() """) print("\n=== Asset Partitions ===") print(""" Partitions: - Split asset by date, region, etc. - Incremental processing - Backfill Example: from dagster import DailyPartitionsDefinition daily_partitions = DailyPartitionsDefinition(start_date="2024-01-01") @asset(partitions_def=daily_partitions) def daily_orders(context): date = context.partition_key return fetch_orders(date) """) print("\n=== Asset Metadata ===") print(""" Metadata: - Description - Owner - Tags - Column schema - Row count - Preview Example: @asset( description="Daily revenue by product", owners=["team:analytics"], tags={"domain": "finance"}, metadata={"rows": 1000}, ) def daily_revenue(cleaned_orders): return ... """) # IMPLEMENTATION print("\n=== Simulating Assets ===") class Asset: """Simulate Dagster Asset.""" def __init__(self, name, compute_fn=None): self.name = name self.compute_fn = compute_fn self.dependencies = [] self.value = None self.materialized = False def __call__(self, *args, **kwargs): """Materialize asset.""" if self.compute_fn: self.value = self.compute_fn(*args, **kwargs) self.materialized = True return self.value class AssetGraph: """Simulate Dagster Asset Graph.""" def __init__(self): self.assets = {} self.edges = [] def add_asset(self, asset, dependencies=None): """Add asset.""" self.assets[asset.name] = asset if dependencies: for dep in dependencies: asset.dependencies.append(dep) self.edges.append((dep, asset.name)) print(f"AssetGraph: Added '{asset.name}'") def materialize_all(self): """Materialize all assets in order.""" from collections import deque # Topological sort in_degree = {name: 0 for name in self.assets} for src, dst in self.edges: in_degree[dst] += 1 queue = deque([name for name, deg in in_degree.items() if deg == 0]) order = [] while queue: name = queue.popleft() order.append(name) for src, dst in self.edges: if src == name: in_degree[dst] -= 1 if in_degree[dst] == 0: queue.append(dst) # Materialize in order for name in order: asset = self.assets[name] deps = [self.assets[d].value for d in asset.dependencies] asset(*deps) print(f" Materialized: {name}") return order # Demo print("=== Asset Demo ===") graph = AssetGraph() # Define assets raw_users = Asset('raw_users', lambda: ['user1', 'user2', 'user3']) cleaned_users = Asset('cleaned_users', lambda x: [u.upper() for u in x]) user_count = Asset('user_count', lambda x: len(x)) # Add to graph graph.add_asset(raw_users) graph.add_asset(cleaned_users, dependencies=['raw_users']) graph.add_asset(user_count, dependencies=['cleaned_users']) # Materialize print("\nMaterializing:") order = graph.materialize_all() print(f"\nOrder: {order}") print(f"user_count value: {graph.assets['user_count'].value}") # ASSET SUMMARY print("\n=== Asset Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'Asset':<25} {'Data artifact':<40}") print(f"{'Dependency':<25} {'Auto-detected':<40}") print(f"{'Materialize':<25} {'Produce asset':<40}") print(f"{'Partition':<25} {'Split asset':<40}") print(f"{'Metadata':<25} {'Description, owner':<40}") print(f"{'Lineage':<25} {'Tracked':<40}")

🔄 Flows & Tasks

🎯 The Prefect Paradigm

In Prefect, flows are workflows and tasks are units of work. Both are Python functions with decorators (@flow, @task). Prefect handles retries, caching, logging, and observability.

Flows & Tasks

Flow: Workflow. Task: Unit of work. Decorators: @flow, @task. Retries: Built-in. Caching: Built-in. State: Tracked.

🔄 Flow🔄 Task🔄 Retries🔄 Caching
# Flows & Tasks print("=== What is a Flow? ===") print(""" Flow: - Workflow - Python function - @flow decorator - Entry point Example: from prefect import flow @flow def my_flow(): print("Running flow") my_flow() """) print("\n=== What is a Task? ===") print(""" Task: - Unit of work - Python function - @task decorator - Retries, caching Example: from prefect import task @task(retries=3, retry_delay_seconds=60) def fetch_data(): return requests.get(url).json() """) print("\n=== Full Example ===") print(""" from prefect import flow, task from prefect.tasks import task_input_hash from datetime import timedelta @task( retries=3, retry_delay_seconds=60, cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1), ) def extract(url): return requests.get(url).json() @task def transform(data): return [x * 2 for x in data] @task def load(data): print(f"Loaded {len(data)} records") @flow(name="etl-flow", log_prints=True) def etl(url): data = extract(url) transformed = transform(data) load(transformed) if __name__ == "__main__": etl("https://api.example.com/data") """) print("\n=== Task Features ===") print(""" Retries: @task(retries=3, retry_delay_seconds=60) Caching: @task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1)) Timeouts: @task(timeout_seconds=300) Concurrency: @task(tags=["concurrent"]) Logging: @flow(log_prints=True) """) # IMPLEMENTATION print("\n=== Simulating Flows & Tasks ===") class Task: """Simulate Prefect Task.""" def __init__(self, name, fn, retries=0, cache=False): self.name = name self.fn = fn self.retries = retries self.cache = cache self.cache_store = {} self.stats = {'runs': 0, 'retries': 0, 'cached': 0} def __call__(self, *args, **kwargs): """Run task.""" self.stats['runs'] += 1 # Check cache cache_key = (args, tuple(sorted(kwargs.items()))) if self.cache and cache_key in self.cache_store: self.stats['cached'] += 1 print(f" [CACHED] {self.name}") return self.cache_store[cache_key] # Retry logic for attempt in range(self.retries + 1): try: result = self.fn(*args, **kwargs) if self.cache: self.cache_store[cache_key] = result return result except Exception as e: if attempt < self.retries: self.stats['retries'] += 1 print(f" [RETRY {attempt+1}] {self.name}: {e}") else: raise def get_stats(self): return self.stats class Flow: """Simulate Prefect Flow.""" def __init__(self, name): self.name = name self.tasks = [] self.runs = 0 def run(self): """Run flow.""" self.runs += 1 print(f"Flow: {self.name} (run {self.runs})") # Demo print("=== Flow & Task Demo ===") # Define tasks def extract_data(): return [1, 2, 3, 4, 5] def transform_data(data): return [x * 2 for x in data] def load_data(data): print(f" Loaded {len(data)} records") return len(data) # Create tasks extract_task = Task('extract', extract_data) transform_task = Task('transform', transform_data) load_task = Task('load', load_data, retries=2) # Create flow flow = Flow('etl_flow') # Run print("\nRunning flow:") flow.run() data = extract_task() transformed = transform_task(data) count = load_task(transformed) print(f"\nResult: {count} records") # Caching demo print("\n=== Caching Demo ===") cached_task = Task('cached_extract', extract_data, cache=True) print("First run:") r1 = cached_task() print("Second run:") r2 = cached_task() print(f"Stats: {cached_task.get_stats()}") # FLOW SUMMARY print("\n=== Flow & Task Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'Flow':<25} {'Workflow':<40}") print(f"{'Task':<25} {'Unit of work':<40}") print(f"{'Retries':<25} {'Automatic':<40}") print(f"{'Caching':<25} {'Built-in':<40}") print(f"{'Timeout':<25} {'Configurable':<40}") print(f"{'Logging':<25} {'Built-in':<40}")

⚡ Dynamic Workflows

🎯 Runtime DAGs

Dynamic workflows let you create tasks at runtime based on data. This is a major advantage of modern orchestrators over Airflow's static DAGs.

Dynamic Workflows

Runtime: Create tasks at runtime. Map: Over lists. Conditional: Branch. Loops: Iterate. Dagster: Dynamic outputs. Prefect: Native Python.

⚡ Runtime⚡ Map⚡ Conditional⚡ Loops
# Dynamic Workflows print("=== What are Dynamic Workflows? ===") print(""" Dynamic: - Tasks created at runtime - Based on data - Not known at DAG definition Example: - Process N files (N unknown) - Branch on condition - Map over list - Retry based on result """) print("\n=== Prefect Dynamic ===") print(""" from prefect import flow, task @task def process_file(file): return f"Processed {file}" @flow def process_all(files): # Dynamic: iterate over files at runtime for file in files: process_file(file) # Or map @flow def process_mapped(files): process_file.map(files) # Conditional @flow def conditional(should_run): if should_run: task_a() else: task_b() """) print("\n=== Dagster Dynamic ===") print(""" from dagster import op, job, DynamicOut, DynamicOutput, graph @op(out=DynamicOut()) def get_files(): files = ["a.csv", "b.csv", "c.csv"] for file in files: yield DynamicOutput(file, mapping_key=file) @op def process_file(file): return f"Processed {file}" @graph def process_all(): files = get_files() files.map(process_file) process_job = process_all.to_job() """) print("\n=== Why Dynamic Matters ===") print(""" Static (Airflow): - DAG defined at parse time - Fixed structure - Hard to branch/loop Dynamic (Prefect, Dagster): - DAG at runtime - Data-driven - Easy branching/looping Use cases: - Process N files - Map over list - Conditional branches - Dynamic fan-out """) # IMPLEMENTATION print("\n=== Simulating Dynamic Workflows ===") class DynamicFlow: """Simulate dynamic workflow.""" def __init__(self, name): self.name = name self.executed = [] def run(self, data): """Run dynamic flow.""" print(f"Flow: {self.name}") print(f" Data: {data}") # Dynamic: process each item for item in data: result = self.process(item) self.executed.append(result) return self.executed def process(self, item): """Process item.""" result = f"Processed {item}" print(f" {result}") return result class ConditionalFlow: """Simulate conditional flow.""" def __init__(self, name): self.name = name def run(self, condition): """Run conditional.""" print(f"Flow: {self.name}") if condition: print(" Branch A") return "A" else: print(" Branch B") return "B" # Demo print("=== Dynamic Workflow Demo ===") # Dynamic: process list flow = DynamicFlow('process_files') files = ['file1.csv', 'file2.csv', 'file3.csv'] results = flow.run(files) print(f"Results: {results}") # Conditional print() cond = ConditionalFlow('conditional_flow') print(f"Condition True: {cond.run(True)}") print(f"Condition False: {cond.run(False)}") # DYNAMIC SUMMARY print("\n=== Dynamic Workflow Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'Dynamic':<25} {'Runtime DAGs':<40}") print(f"{'Map':<25} {'Over list':<40}") print(f"{'Conditional':<25} {'Branch':<40}") print(f"{'Loop':<25} {'Iterate':<40}") print(f"{'Prefect':<25} {'Native Python':<40}") print(f"{'Dagster':<25} {'Dynamic outputs':<40}")

🔍 Observability

🎯 Understanding What Happened

Modern orchestrators provide rich observability: UI, logs, lineage, metrics, and alerts. Dagster and Prefect have modern UIs that show exactly what happened and why.

Observability

UI: Web interface. Logs: Structured. Lineage: Asset graph. Metrics: Performance. Alerts: Notifications. Traces: Distributed.

🔍 UI🔍 Logs🔍 Lineage🔍 Alerts
# Observability print("=== What is Observability? ===") print(""" Observability: - Understand system state - Debug issues - Optimize performance - Alert on problems Components: - UI: Web interface - Logs: Structured logging - Metrics: Performance - Lineage: Data flow - Alerts: Notifications - Traces: Distributed tracing """) print("\n=== Dagster UI ===") print(""" Dagster UI: - Asset graph - Asset catalog - Run history - Logs - Schedules - Sensors - Backfills - Lineage """) print("\n=== Prefect UI ===") print(""" Prefect UI: - Flow runs - Task runs - Logs - Deployments - Work pools - Blocks - Notifications - Automations """) print("\n=== Logging ===") print(""" Dagster: from dagster import get_dagster_logger logger = get_dagster_logger() logger.info("Message") Prefect: from prefect import get_run_logger logger = get_run_logger() logger.info("Message") """) print("\n=== Alerting ===") print(""" Dagster: - Sensors - Asset checks - Notifications Prefect: - Automations - Notifications - Blocks (Slack, email) """) print("\n=== Lineage ===") print(""" Lineage: - Track data flow - Source → Target - Impact analysis Dagster: - Asset graph - Auto-generated Prefect: - Flow runs - Task dependencies """) # IMPLEMENTATION print("\n=== Simulating Observability ===") class Observability: """Simulate observability.""" def __init__(self): self.runs = [] self.logs = [] self.metrics = [] self.alerts = [] def log_run(self, flow_name, status, duration): """Log run.""" self.runs.append({ 'flow': flow_name, 'status': status, 'duration': duration, 'timestamp': time.time(), }) def log(self, level, message): """Log message.""" self.logs.append({ 'level': level, 'message': message, 'timestamp': time.time(), }) def metric(self, name, value): """Record metric.""" self.metrics.append({ 'name': name, 'value': value, 'timestamp': time.time(), }) def alert(self, message, severity='info'): """Send alert.""" self.alerts.append({ 'message': message, 'severity': severity, 'timestamp': time.time(), }) def get_dashboard(self): """Get dashboard.""" return { 'runs': len(self.runs), 'successful': len([r for r in self.runs if r['status'] == 'success']), 'failed': len([r for r in self.runs if r['status'] == 'failed']), 'alerts': len(self.alerts), } # Demo print("=== Observability Demo ===") obs = Observability() # Log runs obs.log_run('etl_flow', 'success', 120) obs.log_run('ml_flow', 'success', 300) obs.log_run('report_flow', 'failed', 60) # Logs obs.log('INFO', 'Starting etl_flow') obs.log('INFO', 'Extracted 1000 rows') obs.log('ERROR', 'Failed to load data') # Metrics obs.metric('rows_processed', 1000) obs.metric('duration_seconds', 120) # Alerts obs.alert('Flow report_flow failed', 'error') obs.alert('High memory usage', 'warning') # Dashboard print(f"Dashboard: {obs.get_dashboard()}") print(f"\nLogs:") for log in obs.logs: print(f" [{log['level']}] {log['message']}") print(f"\nAlerts:") for alert in obs.alerts: print(f" [{alert['severity']}] {alert['message']}") # OBSERVABILITY SUMMARY print("\n=== Observability Summary ===") print(f"{'Component':<25} {'Description':<40}") print("-" * 65) print(f"{'UI':<25} {'Web interface':<40}") print(f"{'Logs':<25} {'Structured logging':<40}") print(f"{'Lineage':<25} {'Data flow':<40}") print(f"{'Metrics':<25} {'Performance':<40}") print(f"{'Alerts':<25} {'Notifications':<40}") print(f"{'Traces':<25} {'Distributed':<40}")

✅ Testing

🎯 Testable Workflows

Modern orchestrators make workflows testable. You can unit test individual tasks/assets without running the full pipeline. This is a major advantage over Airflow.

Testing

Unit tests: Test tasks. Integration: Test flows. Mock: External systems. Fixtures: Test data. Dagster: Asset tests. Prefect: Flow tests.

✅ Unit Tests✅ Integration✅ Mock✅ Fixtures
# Testing print("=== Why Testing Matters ===") print(""" Testing: - Catch bugs early - Refactor safely - Document behavior - CI/CD Airflow: - Hard to test - Requires scheduler - Top-level code issues Modern: - Easy to test - Pure Python functions - No scheduler needed """) print("\n=== Dagster Testing ===") print(""" from dagster import asset, materialize @asset def my_asset(): return [1, 2, 3] def test_my_asset(): result = materialize([my_asset]) assert result.success assert result.output_for_node("my_asset") == [1, 2, 3] """) print("\n=== Prefect Testing ===") print(""" from prefect import flow, task from prefect.testing.utilities import prefect_test_harness @task def my_task(x): return x * 2 @flow def my_flow(x): return my_task(x) def test_my_task(): assert my_task(5) == 10 def test_my_flow(): with prefect_test_harness(): assert my_flow(5) == 10 """) print("\n=== Mocking ===") print(""" Dagster: from unittest.mock import patch @patch("my_module.external_api") def test_asset(mock_api): mock_api.return_value = {"data": [1, 2, 3]} result = materialize([my_asset]) assert result.success Prefect: from unittest.mock import patch @patch("my_module.external_api") def test_flow(mock_api): mock_api.return_value = {"data": [1, 2, 3]} result = my_flow() assert result == 3 """) print("\n=== CI/CD ===") print(""" GitHub Actions: - Run tests on PR - Lint - Deploy Example: name: Test on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 - run: pip install -r requirements.txt - run: pytest """) # IMPLEMENTATION print("\n=== Simulating Testing ===") class TestRunner: """Simulate test runner.""" def __init__(self): self.tests = [] self.results = [] def add_test(self, name, test_fn): """Add test.""" self.tests.append({'name': name, 'fn': test_fn}) def run(self): """Run tests.""" for test in self.tests: try: test['fn']() self.results.append({'name': test['name'], 'status': 'PASS'}) except AssertionError as e: self.results.append({'name': test['name'], 'status': 'FAIL', 'error': str(e)}) except Exception as e: self.results.append({'name': test['name'], 'status': 'ERROR', 'error': str(e)}) return self.results def report(self): """Report results.""" passed = len([r for r in self.results if r['status'] == 'PASS']) total = len(self.results) print(f"\nTests: {passed}/{total} passed") for r in self.results: status = "✓" if r['status'] == 'PASS' else "✗" print(f" {status} {r['name']}") if 'error' in r: print(f" {r['error']}") # Demo print("=== Testing Demo ===") # Define functions to test def extract_data(): return [1, 2, 3, 4, 5] def transform_data(data): return [x * 2 for x in data] def load_data(data): return len(data) # Define tests def test_extract(): result = extract_data() assert len(result) == 5 assert result[0] == 1 def test_transform(): result = transform_data([1, 2, 3]) assert result == [2, 4, 6] def test_load(): result = load_data([1, 2, 3]) assert result == 3 def test_failing(): result = transform_data([1, 2, 3]) assert result == [1, 2, 3] # Wrong! # Run tests runner = TestRunner() runner.add_test('test_extract', test_extract) runner.add_test('test_transform', test_transform) runner.add_test('test_load', test_load) runner.add_test('test_failing', test_failing) runner.run() runner.report() # TESTING SUMMARY print("\n=== Testing Summary ===") print(f"{'Test Type':<25} {'Description':<40}") print("-" * 65) print(f"{'Unit test':<25} {'Test task/asset':<40}") print(f"{'Integration':<25} {'Test flow':<40}") print(f"{'Mock':<25} {'External systems':<40}") print(f"{'Fixtures':<25} {'Test data':<40}") print(f"{'CI/CD':<25} {'Automated':<40}")

🚢 Deployment

🎯 Running in Production

Modern orchestrators offer flexible deployment: local, Docker, Kubernetes, cloud. Dagster and Prefect both have cloud offerings and can run anywhere.

Deployment

Local: Development. Docker: Containers. Kubernetes: Cloud-native. Cloud: Managed. Dagster: Dagster Cloud. Prefect: Prefect Cloud.

🚢 Local🚢 Docker🚢 Kubernetes🚢 Cloud
# Deployment print("=== Deployment Options ===") print(""" 1. LOCAL - Development - Single machine - No infrastructure 2. DOCKER - Containers - Reproducible - Portable 3. KUBERNETES - Cloud-native - Auto-scaling - Elastic 4. CLOUD - Dagster Cloud - Prefect Cloud - Managed """) print("\n=== Dagster Deployment ===") print(""" Local: dagster dev Docker: docker run -p 3000:3000 dagster/dagster Kubernetes: - Helm chart - Dagster Operator Cloud: - Dagster Cloud - Hybrid - Serverless """) print("\n=== Prefect Deployment ===") print(""" Local: prefect server start Docker: docker run -p 4200:4200 prefecthq/prefect Kubernetes: - Helm chart - Prefect Worker Cloud: - Prefect Cloud - Managed - Work pools """) print("\n=== Deployment Comparison ===") print(f"{'Option':<25} {'Dagster':<20} {'Prefect':<20}") print("-" * 65) print(f"{'Local':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Docker':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Kubernetes':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Cloud':<25} {'Dagster Cloud':<20} {'Prefect Cloud':<20}") print(f"{'Self-hosted':<25} {'Yes':<20} {'Yes':<20}") print("\n=== Best Practices ===") print(""" 1. Use infrastructure as code 2. Containerize 3. Version control 4. CI/CD 5. Monitor 6. Alert 7. Document """) # IMPLEMENTATION print("\n=== Simulating Deployment ===") class Deployment: """Simulate deployment.""" def __init__(self, name, environment, orchestrator): self.name = name self.environment = environment self.orchestrator = orchestrator self.status = 'not_deployed' self.version = '0.0.0' def deploy(self, version): """Deploy.""" self.version = version self.status = 'deployed' print(f"Deployed {self.name} v{version} to {self.environment}") def get_info(self): return { 'name': self.name, 'environment': self.environment, 'orchestrator': self.orchestrator, 'version': self.version, 'status': self.status, } # Demo print("=== Deployment Demo ===") # Deploy deployments = [ Deployment('etl_pipeline', 'staging', 'dagster'), Deployment('ml_pipeline', 'production', 'prefect'), ] for d in deployments: d.deploy('1.0.0') for d in deployments: print(f" {d.get_info()}") # DEPLOYMENT SUMMARY print("\n=== Deployment Summary ===") print(f"{'Option':<25} {'Use Case':<40}") print("-" * 65) print(f"{'Local':<25} {'Development':<40}") print(f"{'Docker':<25} {'Containers':<40}") print(f"{'Kubernetes':<25} {'Cloud-native':<40}") print(f"{'Cloud':<25} {'Managed':<40}")

🛠️ Ecosystem

🎯 Tools and Integrations

Dagster and Prefect have growing ecosystems: integrations with dbt, Spark, Snowflake, and more. Each has its own ecosystem philosophy.

Ecosystem

Dagster: Integrations, dbt, Snowflake. Prefect: Blocks, integrations. Both: dbt, Spark, cloud. Growing: Communities. Modern: Python-first.

🛠️ Integrations🛠️ dbt🛠️ Spark🛠️ Cloud
# Ecosystem print("=== Dagster Ecosystem ===") print(""" Integrations: - dbt - Snowflake - BigQuery - Spark - Pandas - DuckDB - Airbyte - Fivetran - AWS, GCP, Azure Features: - Asset checks - Data quality - Lineage - Observability """) print("\n=== Prefect Ecosystem ===") print(""" Integrations: - dbt - Snowflake - BigQuery - Spark - AWS, GCP, Azure - Slack - Email - GitHub Features: - Blocks - Notifications - Automations - Work pools """) print("\n=== Comparison ===") print(f"{'Integration':<25} {'Dagster':<20} {'Prefect':<20}") print("-" * 65) print(f"{'dbt':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Snowflake':<25} {'Yes':<20} {'Yes':<20}") print(f"{'BigQuery':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Spark':<25} {'Yes':<20} {'Yes':<20}") print(f"{'AWS':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Slack':<25} {'Yes':<20} {'Yes':<20}") print(f"{'Airbyte':<25} {'Yes':<20} {'Yes':<20}") print("\n=== Which Ecosystem? ===") print(""" Dagster: - Data-aware - Asset checks - dbt integration - Modern data stack Prefect: - Flexible - Blocks - Automations - General-purpose """)

⚖️ Full Comparison

🎯 Dagster vs Prefect vs Airflow

Complete comparison of the three main orchestrators across all dimensions.

Comparison Table

Use this table as a quick reference for choosing the right orchestrator.

⚖️ Reference⚖️ Decision⚖️ Complete
# Full Comparison Table print("=== Dagster vs Prefect vs Airflow ===") print() # Print header print(f"{'Aspect':<25} {'Airflow':<20} {'Dagster':<20} {'Prefect':<20}") print("-" * 85) # Data rows = [ ('Paradigm', 'Task', 'Asset', 'Flow'), ('DAG definition', 'Static', 'Asset graph', 'Dynamic'), ('Data-aware', 'No', 'Yes', 'Partial'), ('Type-safe', 'No', 'Yes', 'Partial'), ('Testing', 'Hard', 'Easy', 'Easy'), ('Local dev', 'OK', 'Good', 'Excellent'), ('Dynamic', 'Limited', 'Good', 'Excellent'), ('UI', 'Mature', 'Modern', 'Modern'), ('Maturity', 'Mature', 'Modern', 'Modern'), ('Community', 'Largest', 'Growing', 'Growing'), ('Cloud', 'Many', 'Dagster Cloud', 'Prefect Cloud'), ('Language', 'Python', 'Python', 'Python'), ('Scheduling', 'Cron', 'Cron/Sensor', 'Cron/Event'), ('Retries', 'Yes', 'Yes', 'Yes'), ('Lineage', 'Limited', 'Excellent', 'Good'), ('Observability', 'Good', 'Excellent', 'Excellent'), ('Learning curve', 'Medium', 'Steep', 'Low'), ] for row in rows: print(f"{row[0]:<25} {row[1]:<20} {row[2]:<20} {row[3]:<20}") print() print("=== Quick Guide ===") print(""" Choose AIRFLOW when: - Mature ecosystem - Large community - Many integrations - Enterprise adoption Choose DAGSTER when: - Data-aware pipelines - Asset lineage - Type safety - Modern DX Choose PREFECT when: - Dynamic workflows - Python-native - Easy local dev - Flexible """) print("=== Decision Flow ===") print(""" Start │ ├─ Data-aware? ──Yes──→ Dagster │ │ │ No │ ↓ ├─ Dynamic? ──Yes──→ Prefect │ │ │ No │ ↓ ├─ Mature? ──Yes──→ Airflow │ │ │ No │ ↓ └─ Either → Choose based on team """)

🎯 Decision Guide

🎯 Choosing the Right Orchestrator

Use this guide to decide between Airflow, Dagster, and Prefect.

Decision Framework

Questions: Data-aware? Dynamic? Mature? Team? Answers: Lead to tool. Trade-offs: Maturity vs modern. Future: Modern wins. Start: Prefect or Airflow.

🎯 Questions🎯 Trade-offs🎯 Decision🎯 Future
# Decision Guide print("=== Decision Questions ===") print(""" 1. DATA-AWARE? - Yes → Dagster - No → Prefect or Airflow 2. DYNAMIC? - Yes → Prefect - No → Airflow 3. MATURE? - Yes → Airflow - No → Dagster or Prefect 4. TEAM? - Python → Prefect - Data → Dagster - Enterprise → Airflow 5. TESTING? - Important → Dagster or Prefect - Less → Airflow 6. CLOUD? - Managed → Any - Self-hosted → Any """) print("\n=== Decision Matrix ===") print(f"{'Requirement':<30} {'Best Choice':<20} {'Alternative':<20}") print("-" * 70) print(f"{'Data-aware':<30} {'Dagster':<20} {'Prefect':<20}") print(f"{'Dynamic workflows':<30} {'Prefect':<20} {'Dagster':<20}") print(f"{'Mature ecosystem':<30} {'Airflow':<20} {'—':<20}") print(f"{'Easy local dev':<30} {'Prefect':<20} {'Dagster':<20}") print(f"{'Asset lineage':<30} {'Dagster':<20} {'—':<20}") print(f"{'Type safety':<30} {'Dagster':<20} {'Prefect':<20}") print(f"{'Testing':<30} {'Dagster/Prefect':<20} {'Airflow':<20}") print(f"{'Enterprise':<30} {'Airflow':<20} {'Dagster':<20}") print("\n=== Decision Flowchart ===") print(""" START │ ├─ Data-aware? ──Yes──→ Dagster │ │ │ No │ ↓ ├─ Dynamic? ──Yes──→ Prefect │ │ │ No │ ↓ ├─ Mature? ──Yes──→ Airflow │ │ │ No │ ↓ └─ Either → Choose based on team """) print("\n=== Common Scenarios ===") print(""" 1. MODERN DATA STACK → Dagster 2. DYNAMIC PYTHON WORKFLOWS → Prefect 3. LARGE ENTERPRISE → Airflow 4. DATA-AWARE PIPELINES → Dagster 5. QUICK PROTOTYPING → Prefect 6. EXISTING AIRFLOW → Stay or migrate 7. ML PIPELINES → Dagster or Prefect 8. ETL PIPELINES → Any """)

🚀 Future

🎯 Where Modern Orchestration Is Going

The future is data-aware, dynamic, and AI-driven. Modern orchestrators are becoming the standard for data pipelines.

Future Directions

Data-aware: Standard. Dynamic: Runtime DAGs. AI: Auto-optimization. Cloud: Serverless. Unified: One platform. Open: Standards.

🚀 Data-Aware🚀 Dynamic🚀 AI🚀 Serverless
# Future of Modern Orchestration print("=== 1. Data-Aware Standard ===") print(""" Trend: Data-aware becomes standard Dagster: - Software-Defined Assets - Asset graph - Lineage Airflow: - Dataset scheduling - Data-aware (newer) Prefect: - Assets (coming) Future: - Data-aware default - Asset lineage - Data contracts """) print("\n=== 2. Dynamic Workflows ===") print(""" Trend: Dynamic DAGs standard Prefect: - Native Python - Runtime DAGs Dagster: - Dynamic outputs - Dynamic mapping Airflow: - Dynamic task mapping (newer) Future: - Dynamic default - Data-driven - Less boilerplate """) print("\n=== 3. AI and Automation ===") print(""" Trend: AI-powered orchestration Automation: - Auto-optimization - Auto-scaling - Auto-healing AI: - Anomaly detection - Predictive alerts - Smart scheduling Future: - Self-driving - Self-optimizing - AI-native """) print("\n=== 4. Serverless ===") print(""" Trend: Serverless orchestration Services: - Dagster Cloud - Prefect Cloud - Airflow managed Benefits: - No infrastructure - Pay per use - Auto-scaling Future: - Serverless default - Event-driven - Cost-effective """) print("\n=== 5. Unified Platform ===") print(""" Trend: One platform for all Features: - DAGs - ML pipelines - Data quality - Data lineage - Observability Future: - Unified analytics - Data-aware - ML integration """) print("\n=== 6. Open Standards ===") print(""" Trend: Open wins Standards: - OpenLineage - OpenMetadata - Great Expectations Future: - Standardization - Interoperability - No lock-in """) # TIMELINE print("\n=== Timeline ===") print(""" Airflow: 2014: Started 2016: Apache 2020: Airflow 2.0 Dagster: 2018: Started 2021: Dagster Cloud 2023: Dagster 1.0 Prefect: 2018: Started 2022: Prefect 2.0 2023: Prefect Cloud Future: 2030: Data-aware standard? 2035: AI-native? 2040: New paradigms? """) print("\n=== Final Thoughts ===") print(""" Modern orchestration: - Data-aware - Dynamic - Python-native - Testable - Observable Tools: - Dagster (assets) - Prefect (flows) - Airflow (mature) The future is: - Data-aware - Dynamic - AI-driven - Serverless - Unified "Orchestrate data, not just tasks." Keep exploring. Keep building. """)