🧠 Fundamentals

🎯 What is Apache Airflow?

Apache Airflow is an open-source workflow orchestration platform. It lets you define, schedule, and monitor workflows as code using Python. Airflow represents workflows as DAGs (Directed Acyclic Graphs) of tasks, making complex data pipelines manageable, observable, and reproducible.

Why Airflow Matters

Workflows as code: Python DAGs. Scheduling: Cron-like. Monitoring: UI + logs. Extensible: Operators, hooks. Scalable: Executors. Reproducible: Version-controlled.

📌 DAGs📌 Python📌 Scheduling📌 Orchestration
# Apache Airflow — the big picture # Airflow orchestrates workflows as code. # CONCEPTUAL OVERVIEW print(""" === Apache Airflow === What it is: - Workflow orchestration - DAGs (Directed Acyclic Graphs) - Python-based - Scheduled execution - Monitoring UI Key concepts: - DAG: Workflow definition - Task: Unit of work - Operator: Task template - Scheduler: Triggers DAGs - Executor: Runs tasks - Webserver: UI - Metadata DB: State What it's NOT: - Not a data processing engine - Not a streaming engine - Not a scheduler for sub-second jobs - Not a replacement for Spark/Flink """) # SIMPLE DAG EXAMPLE print("=== Simple DAG Example ===") print(""" from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator from datetime import datetime, timedelta default_args = { 'owner': 'data-team', 'depends_on_past': False, 'email_on_failure': True, 'retries': 1, 'retry_delay': timedelta(minutes=5), } with DAG( 'my_first_dag', default_args=default_args, description='A simple tutorial DAG', schedule_interval='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['example'], ) as dag: extract = BashOperator( task_id='extract', bash_command='python extract.py', ) transform = PythonOperator( task_id='transform', python_callable=transform_data, ) load = BashOperator( task_id='load', bash_command='python load.py', ) extract >> transform >> load """) # KEY PROPERTIES print(""" === Key Properties of Airflow === 1. WORKFLOWS AS CODE - Python DAGs - Version-controlled - Testable - Reviewable 2. SCHEDULING - Cron expressions - Timedelta - Presets (@daily, @hourly) - Backfill 3. DEPENDENCIES - Upstream/downstream - Branching - Trigger rules - Dynamic task mapping 4. MONITORING - Web UI - Logs - Alerts - Metrics 5. EXTENSIBILITY - Operators (Bash, Python, SQL, etc.) - Hooks (connections) - Sensors (waiting) - Plugins 6. SCALABILITY - Executors (Local, Celery, Kubernetes) - Distributed - Cloud-managed """) # REAL-WORLD SCALE print("=== Real-World Airflow Deployments ===") deployments = [ ('Airbnb', 'Original creator', 'Data pipelines'), ('Google', 'Cloud Composer', 'Managed service'), ('Amazon', 'MWAA', 'Managed service'), ('Microsoft', 'Azure Data Factory', 'Managed service'), ('Astronomer', 'Astro', 'Managed service'), ('Many enterprises', 'Self-managed', 'ETL, ML, analytics'), ] print(f"{'Company':<20} {'Offering':<25} {'Use Case':<25}") print("-" * 70) for company, offering, use_case in deployments: print(f"{company:<20} {offering:<25} {use_case:<25}")

🏗️ Architecture

🎯 Airflow Architecture

Airflow consists of Webserver (UI), Scheduler (triggers DAGs), Executor (runs tasks), Workers (execute tasks), and Metadata Database (state).

Architecture Components

Webserver: UI. Scheduler: Triggers. Executor: Dispatches. Workers: Execute. Metadata DB: State. DAG files: Python.

🏗️ Webserver🏗️ Scheduler🏗️ Executor🏗️ Metadata DB
# Airflow Architecture print("=== Architecture Diagram ===") print(""" ┌─────────────────────────────────────────┐ │ Users │ │ (Browser / CLI / API) │ └─────────────────┬───────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ Webserver │ │ (Flask UI) │ │ │ │ - DAG list │ │ - Task logs │ │ - Graph view │ │ - Admin │ └─────────────────┬───────────────────────┘ │ ┌───────────┼───────────┐ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │Scheduler│ │Executor │ │Metadata │ │ │ │ │ │ DB │ │- Parse │ │- Dispatch│ │- State │ │- Trigger│ │- Queue │ │- History│ │- Backfill│ │ │ │- Users │ └────┬────┘ └────┬────┘ └─────────┘ │ │ │ │ ▼ ▼ ┌─────────────────────────────────────────┐ │ Workers │ │ (Celery / Kubernetes / Local) │ │ │ │ - Run tasks │ │ - Report status │ │ - Handle retries │ └─────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ External Systems │ │ (Databases, APIs, Spark, S3, etc.) │ └─────────────────────────────────────────┘ """) # COMPONENTS print("\n=== Airflow Components ===") print(""" 1. WEBSERVER - Flask application - UI for DAGs, tasks, logs - REST API - Authentication 2. SCHEDULER - Parses DAG files - Triggers tasks - Manages dependencies - Backfills 3. EXECUTOR - Dispatches tasks - Types: - Sequential - Local - Celery - Kubernetes - Dask 4. WORKERS - Execute tasks - Report status - Handle retries - Scale horizontally 5. METADATA DATABASE - PostgreSQL (recommended) - MySQL - SQLite (dev only) - Stores: - DAG runs - Task instances - Variables - Connections - Users 6. DAG FILES - Python files - Defined in dags/ folder - Parsed by scheduler - Version-controlled """) # DEPLOYMENT MODES print("\n=== Deployment Modes ===") print(""" 1. LOCAL - Single machine - Development - SequentialExecutor or LocalExecutor 2. SELF-MANAGED - Multiple nodes - CeleryExecutor or KubernetesExecutor - PostgreSQL backend 3. CLOUD-MANAGED - Google Cloud Composer - Amazon MWAA - Azure Data Factory - Astronomer - No infrastructure management """) # IMPLEMENTATION print("\n=== Simulating Airflow Architecture ===") import time import random from collections import defaultdict class AirflowScheduler: """Simulate Airflow Scheduler.""" def __init__(self): self.dags = {} self.dag_runs = [] self.task_instances = [] def register_dag(self, dag_id, tasks, schedule): """Register DAG.""" self.dags[dag_id] = { 'dag_id': dag_id, 'tasks': tasks, 'schedule': schedule, 'active': True, } print(f"Scheduler: Registered DAG '{dag_id}' with {len(tasks)} tasks") def trigger_dag(self, dag_id, execution_date=None): """Trigger DAG run.""" if dag_id not in self.dags: return None dag = self.dags[dag_id] execution_date = execution_date or time.time() dag_run = { 'dag_id': dag_id, 'run_id': f"manual__{execution_date}", 'execution_date': execution_date, 'state': 'running', 'tasks': [], } # Schedule tasks for task_id in dag['tasks']: task_instance = { 'dag_id': dag_id, 'task_id': task_id, 'run_id': dag_run['run_id'], 'state': 'queued', 'start_time': None, 'end_time': None, } dag_run['tasks'].append(task_instance) self.task_instances.append(task_instance) self.dag_runs.append(dag_run) print(f"Scheduler: Triggered DAG '{dag_id}' ({len(dag['tasks'])} tasks)") return dag_run def get_dag_status(self, dag_id): """Get DAG status.""" runs = [r for r in self.dag_runs if r['dag_id'] == dag_id] return { 'dag_id': dag_id, 'total_runs': len(runs), 'running': len([r for r in runs if r['state'] == 'running']), 'success': len([r for r in runs if r['state'] == 'success']), } class AirflowExecutor: """Simulate Airflow Executor.""" def __init__(self, executor_type='local', slots=4): self.executor_type = executor_type self.slots = slots self.running_tasks = {} self.completed_tasks = [] def execute_task(self, task_instance): """Execute a task.""" if len(self.running_tasks) >= self.slots: return False task_instance['state'] = 'running' task_instance['start_time'] = time.time() self.running_tasks[task_instance['task_id']] = task_instance # Simulate execution duration = random.uniform(0.1, 0.5) time.sleep(duration) task_instance['state'] = 'success' task_instance['end_time'] = time.time() del self.running_tasks[task_instance['task_id']] self.completed_tasks.append(task_instance) return True def get_stats(self): return { 'type': self.executor_type, 'slots': self.slots, 'running': len(self.running_tasks), 'completed': len(self.completed_tasks), } # Demo print("=== Airflow Simulation ===") scheduler = AirflowScheduler() executor = AirflowExecutor('local', slots=2) # Register DAGs scheduler.register_dag( 'etl_pipeline', ['extract', 'transform', 'load'], '@daily' ) scheduler.register_dag( 'ml_training', ['preprocess', 'train', 'evaluate'], '@weekly' ) # Trigger DAG dag_run = scheduler.trigger_dag('etl_pipeline') print(f"\nExecuting tasks:") for task in dag_run['tasks']: executor.execute_task(task) print(f" {task['task_id']}: {task['state']}") # Status print(f"\nDAG status: {scheduler.get_dag_status('etl_pipeline')}") print(f"Executor stats: {executor.get_stats()}") # ARCHITECTURE SUMMARY print("\n=== Architecture Summary ===") print(f"{'Component':<25} {'Role':<40}") print("-" * 65) print(f"{'Webserver':<25} {'UI, REST API':<40}") print(f"{'Scheduler':<25} {'Parse, trigger DAGs':<40}") print(f"{'Executor':<25} {'Dispatch tasks':<40}") print(f"{'Workers':<25} {'Execute tasks':<40}") print(f"{'Metadata DB':<25} {'State, history':<40}") print(f"{'DAG files':<25} {'Python workflows':<40}")

📊 DAGs

🎯 Directed Acyclic Graphs

A DAG (Directed Acyclic Graph) defines a workflow: tasks as nodes, dependencies as edges. "Acyclic" means no cycles — the workflow flows in one direction. DAGs are defined in Python.

DAGs

Directed: One-way flow. Acyclic: No loops. Tasks: Nodes. Dependencies: Edges. Python: Defined as code. Versioned: Git-managed.

📊 Directed📊 Acyclic📊 Tasks📊 Dependencies
# DAGs — Directed Acyclic Graphs print("=== What is a DAG? ===") print(""" DAG: - Directed: One-way dependencies - Acyclic: No cycles - Graph: Tasks as nodes - Dependencies: Edges Example: extract → transform → load Or: ┌─→ task_b ─┐ task_a ├─→ task_d └─→ task_c ─┘ DAGs are defined in Python. """) print("\n=== DAG Definition ===") print(""" from airflow import DAG from airflow.operators.bash import BashOperator from datetime import datetime, timedelta with DAG( 'my_dag', default_args={ 'owner': 'data-team', 'retries': 1, 'retry_delay': timedelta(minutes=5), }, description='My workflow', schedule_interval='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'daily'], ) as dag: task1 = BashOperator( task_id='extract', bash_command='python extract.py', ) task2 = BashOperator( task_id='transform', bash_command='python transform.py', ) task3 = BashOperator( task_id='load', bash_command='python load.py', ) # Dependencies task1 >> task2 >> task3 """) print("\n=== DAG Parameters ===") print(""" Key parameters: - dag_id: Unique name - schedule_interval: Cron or preset - start_date: First run - end_date: Last run (optional) - catchup: Run missed intervals - max_active_runs: Concurrency - default_args: Default task args - tags: Organization - doc_md: Documentation """) print("\n=== Dependencies ===") print(""" Linear: task1 >> task2 >> task3 Fan-out: task1 >> [task2, task3, task4] Fan-in: [task1, task2, task3] >> task4 Branching: task1 >> task2 task1 >> task3 Diamond: task1 / \\ task2 task3 \\ / task4 """) # IMPLEMENTATION print("\n=== Simulating DAGs ===") class DAG: """Simulate Airflow DAG.""" def __init__(self, dag_id, schedule='@daily', start_date=None): self.dag_id = dag_id self.schedule = schedule self.start_date = start_date self.tasks = {} self.dependencies = defaultdict(list) # task -> downstream self.upstream = defaultdict(list) # task -> upstream def add_task(self, task_id, operator='BashOperator'): """Add a task.""" self.tasks[task_id] = { 'task_id': task_id, 'operator': operator, 'upstream': [], 'downstream': [], } return task_id def set_dependency(self, upstream, downstream): """Set dependency.""" self.dependencies[upstream].append(downstream) self.upstream[downstream].append(upstream) self.tasks[upstream]['downstream'].append(downstream) self.tasks[downstream]['upstream'].append(upstream) def topological_sort(self): """Topological sort for execution order.""" from collections import deque in_degree = {t: len(self.upstream[t]) for t in self.tasks} queue = deque([t for t in self.tasks if in_degree[t] == 0]) result = [] while queue: task = queue.popleft() result.append(task) for downstream in self.dependencies[task]: in_degree[downstream] -= 1 if in_degree[downstream] == 0: queue.append(downstream) return result def get_info(self): return { 'dag_id': self.dag_id, 'schedule': self.schedule, 'tasks': len(self.tasks), 'dependencies': sum(len(v) for v in self.dependencies.values()), } # Demo print("=== DAG Demo ===") dag = DAG('etl_pipeline', schedule='@daily') # Add tasks dag.add_task('extract', 'BashOperator') dag.add_task('transform', 'PythonOperator') dag.add_task('load', 'BashOperator') dag.add_task('validate', 'PythonOperator') dag.add_task('notify', 'EmailOperator') # Dependencies dag.set_dependency('extract', 'transform') dag.set_dependency('transform', 'load') dag.set_dependency('load', 'validate') dag.set_dependency('validate', 'notify') print(f"DAG: {dag.dag_id}") print(f"Tasks: {list(dag.tasks.keys())}") print(f"Dependencies: {dict(dag.dependencies)}") print(f"Execution order: {dag.topological_sort()}") print(f"Info: {dag.get_info()}") # DAG SUMMARY print("\n=== DAG Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'DAG':<25} {'Directed Acyclic Graph':<40}") print(f"{'Task':<25} {'Unit of work':<40}") print(f"{'Dependency':<25} {'Upstream/downstream':<40}") print(f"{'Schedule':<25} {'Cron or preset':<40}") print(f"{'start_date':<25} {'First run':<40}") print(f"{'catchup':<25} {'Run missed intervals':<40}")

🔧 Operators

🎯 Task Templates

Operators are templates for tasks. Airflow has many built-in operators (Bash, Python, SQL, etc.), and you can create custom ones. Each operator defines what a task does.

Operators

Action: Do something. Transfer: Move data. Sensor: Wait for condition. Bash: Shell command. Python: Callable. SQL: Query.

🔧 Action🔧 Transfer🔧 Sensor🔧 Custom
# Operators print("=== What are Operators? ===") print(""" Operator: - Template for a task - Defines what to do - Reusable - Configurable Types: 1. Action Operators - BashOperator - PythonOperator - EmailOperator - HTTPOperator 2. Transfer Operators - S3ToRedshiftOperator - MySqlToHiveOperator - GCSToBigQueryOperator 3. Sensor Operators - FileSensor - S3KeySensor - ExternalTaskSensor - HttpSensor """) print("\n=== Common Operators ===") print(""" 1. BashOperator - Run shell command BashOperator( task_id='run_script', bash_command='python script.py', ) 2. PythonOperator - Run Python callable PythonOperator( task_id='process', python_callable=my_function, ) 3. EmailOperator - Send email EmailOperator( task_id='notify', to='team@example.com', subject='Done', html_content='Done', ) 4. SQL Operators - PostgresOperator - MySqlOperator - BigQueryOperator - SnowflakeOperator 5. Transfer Operators - S3ToRedshiftOperator - GCSToBigQueryOperator - S3ToSnowflakeOperator 6. Sensor Operators - FileSensor - S3KeySensor - ExternalTaskSensor """) print("\n=== Custom Operator ===") print(""" from airflow.models import BaseOperator from airflow.utils.decorators import apply_defaults class MyCustomOperator(BaseOperator): @apply_defaults def __init__(self, my_param, *args, **kwargs): super().__init__(*args, **kwargs) self.my_param = my_param def execute(self, context): # Do something self.log.info(f"Param: {self.my_param}") return "result" """) # IMPLEMENTATION print("\n=== Simulating Operators ===") class BaseOperator: """Simulate BaseOperator.""" def __init__(self, task_id, **kwargs): self.task_id = task_id self.kwargs = kwargs self.state = 'pending' self.result = None def execute(self, context): """Execute task.""" raise NotImplementedError class BashOperator(BaseOperator): """Simulate BashOperator.""" def execute(self, context): self.state = 'running' cmd = self.kwargs.get('bash_command', 'echo hello') self.result = f"Executed: {cmd}" self.state = 'success' return self.result class PythonOperator(BaseOperator): """Simulate PythonOperator.""" def execute(self, context): self.state = 'running' func = self.kwargs.get('python_callable') if func: self.result = func() else: self.result = "No callable" self.state = 'success' return self.result class EmailOperator(BaseOperator): """Simulate EmailOperator.""" def execute(self, context): self.state = 'running' to = self.kwargs.get('to', 'team@example.com') self.result = f"Email sent to {to}" self.state = 'success' return self.result # Demo print("=== Operator Demo ===") # Create operators bash_task = BashOperator( task_id='extract', bash_command='python extract.py' ) def process_data(): return "Processed 1000 records" python_task = PythonOperator( task_id='transform', python_callable=process_data ) email_task = EmailOperator( task_id='notify', to='data-team@example.com' ) # Execute context = {} for task in [bash_task, python_task, email_task]: result = task.execute(context) print(f" {task.task_id}: {task.state} — {result}") # OPERATOR SUMMARY print("\n=== Operator Summary ===") print(f"{'Operator':<25} {'Purpose':<40}") print("-" * 65) print(f"{'BashOperator':<25} {'Run shell command':<40}") print(f"{'PythonOperator':<25} {'Run Python callable':<40}") print(f"{'EmailOperator':<25} {'Send email':<40}") print(f"{'SQL Operators':<25} {'Run SQL':<40}") print(f"{'Transfer Operators':<25} {'Move data':<40}") print(f"{'Sensor Operators':<25} {'Wait for condition':<40}")

📋 Tasks

🎯 Units of Work

A task is a single unit of work in a DAG. Each task is an instance of an operator and has a unique task_id. Tasks have states (queued, running, success, failed, etc.) and can be retried.

Tasks

task_id: Unique name. Operator: Template. State: queued/running/success/failed. Retries: Automatic. Depends_on_past: Sequential. Trigger rules: When to run.

📋 task_id📋 State📋 Retries📋 Trigger Rules
# Tasks print("=== What is a Task? ===") print(""" Task: - Unit of work - Instance of operator - Has task_id (unique) - Has state - Can be retried Task instance: - One execution of a task - For a specific DAG run - Has start/end time - Has logs """) print("\n=== Task States ===") print(""" - none: Not yet scheduled - scheduled: Scheduled - queued: Waiting for executor - running: Executing - success: Completed successfully - failed: Failed - up_for_retry: Will retry - up_for_reschedule: Sensor reschedule - skipped: Skipped - upstream_failed: Upstream failed - deferred: Deferred to trigger """) print("\n=== Retries ===") print(""" Retries: - Automatic retry on failure - Configurable per task - Exponential backoff default_args = { 'retries': 3, 'retry_delay': timedelta(minutes=5), 'retry_exponential_backoff': True, 'max_retry_delay': timedelta(minutes=60), } """) print("\n=== Trigger Rules ===") print(""" Trigger rules: - all_success (default): All upstream succeeded - all_failed: All upstream failed - all_done: All upstream done - one_success: At least one upstream succeeded - one_failed: At least one upstream failed - none_failed: No upstream failed - none_failed_min_one_success: No failures, at least one success - none_skipped: No upstream skipped """) print("\n=== Task Dependencies ===") print(""" Linear: task1 >> task2 >> task3 Fan-out: task1 >> [task2, task3, task4] Fan-in: [task1, task2, task3] >> task4 Cross-dependencies: task1 >> task2 task1 >> task3 task2 >> task4 task3 >> task4 Dynamic: task1 >> task2 >> task3 task1 >> task3 # Skip task2 """) # IMPLEMENTATION print("\n=== Simulating Tasks ===") class Task: """Simulate Airflow Task.""" def __init__(self, task_id, operator='BashOperator', retries=0): self.task_id = task_id self.operator = operator self.retries = retries self.state = 'none' self.attempt = 0 self.start_time = None self.end_time = None self.logs = [] def execute(self, context=None): """Execute task.""" self.attempt += 1 self.state = 'running' self.start_time = time.time() self.logs.append(f"Attempt {self.attempt}") # Simulate execution import random success = random.random() > 0.3 # 70% success if success: self.state = 'success' self.logs.append("Success") else: if self.attempt <= self.retries: self.state = 'up_for_retry' self.logs.append(f"Failed, will retry ({self.attempt}/{self.retries})") else: self.state = 'failed' self.logs.append("Failed, no retries left") self.end_time = time.time() return self.state def get_info(self): return { 'task_id': self.task_id, 'state': self.state, 'attempt': self.attempt, 'retries': self.retries, } # Demo print("=== Task Demo ===") # Create tasks tasks = [ Task('extract', 'BashOperator', retries=2), Task('transform', 'PythonOperator', retries=1), Task('load', 'BashOperator', retries=3), ] # Execute print("Executing tasks:") for task in tasks: state = task.execute() print(f" {task.task_id}: {state}") # Retry demo print("\n=== Retry Demo ===") import random random.seed(42) retry_task = Task('flaky_task', 'PythonOperator', retries=3) for i in range(5): if retry_task.state in ['success', 'failed']: break state = retry_task.execute() print(f" Attempt {retry_task.attempt}: {state}") print(f"\nFinal: {retry_task.get_info()}") # TASK SUMMARY print("\n=== Task Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'task_id':<25} {'Unique identifier':<40}") print(f"{'Operator':<25} {'Task template':<40}") print(f"{'State':<25} {'Current status':<40}") print(f"{'Retries':<25} {'Automatic retry':<40}") print(f"{'Trigger rules':<25} {'When to run':<40}")

⏰ Scheduling

🎯 When DAGs Run

Airflow schedules DAGs using cron expressions, timedelta, or presets. The scheduler triggers DAG runs at the specified intervals.

Scheduling

Cron: Standard syntax. Presets: @daily, @hourly. Timedelta: Python timedelta. start_date: First run. catchup: Run missed. backfill: Manual catchup.

⏰ Cron⏰ Presets⏰ start_date⏰ catchup
# Scheduling print("=== Schedule Types ===") print(""" 1. CRON EXPRESSIONS - Standard cron syntax - 5 fields: minute hour day month weekday '0 0 * * *' # Daily at midnight '0 */2 * * *' # Every 2 hours '0 9 * * 1-5' # Weekdays at 9am '0 0 1 * *' # First of month 2. PRESETS - @once: Once - @hourly: Every hour - @daily: Every day - @weekly: Every week - @monthly: Every month - @yearly: Every year - @continuous: Continuous 3. TIMEDELTA - Python timedelta from datetime import timedelta schedule_interval=timedelta(hours=2) 4. DATASET (NEW) - Data-driven scheduling - Trigger on dataset update schedule=[dataset1, dataset2] """) print("\n=== start_date and catchup ===") print(""" start_date: - When DAG first runs - Must be in the past - Used for backfill catchup: - True: Run all missed intervals - False: Only run latest - Default: True Example: start_date = datetime(2024, 1, 1) schedule = '@daily' catchup = False If today is 2024-01-10: catchup=True: Run Jan 1-10 (10 runs) catchup=False: Run Jan 10 only (1 run) """) print("\n=== Backfill ===") print(""" Backfill: - Run DAG for past dates - Manual or automatic - CLI: airflow dags backfill my_dag \\ --start-date 2024-01-01 \\ --end-date 2024-01-10 Use cases: - Reprocess data - Fix bugs - New DAG for historical data """) # IMPLEMENTATION print("\n=== Simulating Scheduling ===") from datetime import datetime, timedelta class AirflowScheduler: """Simulate Airflow scheduling.""" def __init__(self): self.dags = {} self.schedules = [] def register_dag(self, dag_id, schedule, start_date, catchup=True): """Register DAG with schedule.""" self.dags[dag_id] = { 'dag_id': dag_id, 'schedule': schedule, 'start_date': start_date, 'catchup': catchup, } print(f"Scheduler: Registered '{dag_id}' with schedule '{schedule}'") def calculate_runs(self, dag_id, current_date): """Calculate DAG runs between start_date and current_date.""" if dag_id not in self.dags: return [] dag = self.dags[dag_id] start = dag['start_date'] if not dag['catchup']: return [current_date] # Calculate intervals runs = [] current = start while current <= current_date: runs.append(current) if dag['schedule'] == '@hourly': current += timedelta(hours=1) elif dag['schedule'] == '@daily': current += timedelta(days=1) elif dag['schedule'] == '@weekly': current += timedelta(weeks=1) elif dag['schedule'] == '@monthly': current += timedelta(days=30) else: current += timedelta(days=1) return runs def get_schedule_info(self, dag_id): """Get schedule info.""" if dag_id not in self.dags: return None return self.dags[dag_id] # Demo print("=== Scheduling Demo ===") scheduler = AirflowScheduler() # Register DAGs scheduler.register_dag( 'daily_etl', '@daily', datetime(2024, 1, 1), catchup=True ) scheduler.register_dag( 'hourly_metrics', '@hourly', datetime(2024, 1, 1, 0), catchup=False ) # Calculate runs current = datetime(2024, 1, 5, 12) print(f"\nCurrent date: {current}") daily_runs = scheduler.calculate_runs('daily_etl', current) print(f"\nDaily ETL runs: {len(daily_runs)}") print(f" First: {daily_runs[0]}") print(f" Last: {daily_runs[-1]}") hourly_runs = scheduler.calculate_runs('hourly_metrics', current) print(f"\nHourly metrics runs: {len(hourly_runs)}") print(f" Runs: {hourly_runs}") # SCHEDULING SUMMARY print("\n=== Scheduling Summary ===") print(f"{'Schedule':<25} {'Description':<40}") print("-" * 65) print(f"{'@once':<25} {'Run once':<40}") print(f"{'@hourly':<25} {'Every hour':<40}") print(f"{'@daily':<25} {'Every day at midnight':<40}") print(f"{'@weekly':<25} {'Every week':<40}") print(f"{'@monthly':<25} {'Every month':<40}") print(f"{'Cron':<25} {'Custom schedule':<40}") print(f"{'timedelta':<25} {'Python timedelta':<40}") print(f"{'Dataset':<25} {'Data-driven':<40}")

⚙️ Executors

🎯 Running Tasks

Executors determine how and where tasks run. Airflow has several executors: Sequential, Local, Celery, Kubernetes, and Dask. Each has different scaling characteristics.

Executors

Sequential: Dev only. Local: Single machine. Celery: Distributed. Kubernetes: Container-native. Dask: Dask cluster. Scale: Depends on executor.

⚙️ Sequential⚙️ Local⚙️ Celery⚙️ Kubernetes
# Executors print("=== Executor Types ===") print(""" 1. SEQUENTIAL EXECUTOR - Single-threaded - Development only - SQLite backend - No parallelism 2. LOCAL EXECUTOR - Multi-process - Single machine - PostgreSQL backend - Parallel tasks 3. CELERY EXECUTOR - Distributed - Multiple workers - Redis/RabbitMQ broker - Production 4. KUBERNETES EXECUTOR - Container-native - Each task = pod - Auto-scaling - Cloud-native 5. DASK EXECUTOR - Dask cluster - Distributed - Python-native """) print("\n=== Comparison ===") print(f"{'Executor':<25} {'Parallelism':<20} {'Scale':<20}") print("-" * 65) print(f"{'Sequential':<25} {'1':<20} {'Dev':<20}") print(f"{'Local':<25} {'Many (1 machine)':<20} {'Medium':<20}") print(f"{'Celery':<25} {'Many (N machines)':<20} {'Large':<20}") print(f"{'Kubernetes':<25} {'Many (N pods)':<20} {'Elastic':<20}") print(f"{'Dask':<25} {'Many (Dask)':<20} {'Large':<20}") print("\n=== Configuration ===") print(""" # airflow.cfg [core] executor = CeleryExecutor # Celery broker_url = redis://localhost:6379/0 result_backend = db+postgresql://airflow:airflow@localhost/airflow # Kubernetes [kubernetes] namespace = airflow worker_container_repository = apache/airflow worker_container_tag = 2.7.0 """) print("\n=== When to Use Which ===") print(""" SEQUENTIAL: - Development - Testing LOCAL: - Single machine - Small workloads - Development CELERY: - Production - Distributed - Large workloads KUBERNETES: - Cloud-native - Elastic - Container-based DASK: - Python-native - Distributed - Data science """) # IMPLEMENTATION print("\n=== Simulating Executors ===") class Executor: """Base Executor.""" def __init__(self, name, max_workers): self.name = name self.max_workers = max_workers self.running = [] self.completed = [] def submit(self, task): """Submit task.""" if len(self.running) >= self.max_workers: return False self.running.append(task) task['state'] = 'running' return True def complete(self, task): """Complete task.""" if task in self.running: self.running.remove(task) task['state'] = 'success' self.completed.append(task) def get_stats(self): return { 'name': self.name, 'max_workers': self.max_workers, 'running': len(self.running), 'completed': len(self.completed), } class SequentialExecutor(Executor): def __init__(self): super().__init__('Sequential', 1) class LocalExecutor(Executor): def __init__(self, workers=4): super().__init__('Local', workers) class CeleryExecutor(Executor): def __init__(self, workers=16): super().__init__('Celery', workers) class KubernetesExecutor(Executor): def __init__(self, workers=100): super().__init__('Kubernetes', workers) # Demo print("=== Executor Demo ===") executors = [ SequentialExecutor(), LocalExecutor(4), CeleryExecutor(16), KubernetesExecutor(100), ] # Submit tasks tasks = [{'task_id': f'task_{i}', 'state': 'queued'} for i in range(10)] print("Submitting 10 tasks to each executor:") for executor in executors: submitted = 0 for task in tasks: if executor.submit(task.copy()): submitted += 1 print(f" {executor.name}: {submitted}/{len(tasks)} submitted") print(f" {executor.get_stats()}") # EXECUTOR SUMMARY print("\n=== Executor Summary ===") print(f"{'Executor':<25} {'Use Case':<40}") print("-" * 65) print(f"{'Sequential':<25} {'Development':<40}") print(f"{'Local':<25} {'Single machine':<40}") print(f"{'Celery':<25} {'Distributed production':<40}") print(f"{'Kubernetes':<25} {'Cloud-native':<40}") print(f"{'Dask':<25} {'Python data science':<40}")

🔌 Connections & Hooks

🎯 Connecting to External Systems

Connections store credentials and connection info for external systems. Hooks use connections to interact with those systems. Both are managed in the Airflow UI or CLI.

Connections & Hooks

Connection: Credentials. Hook: Interface. Types: Postgres, MySQL, S3, etc. UI: Manage connections. CLI: airflow connections. Security: Encrypted.

🔌 Connections🔌 Hooks🔌 Credentials🔌 External Systems
# Connections & Hooks print("=== What is a Connection? ===") print(""" Connection: - Stores credentials - Connection info - External systems - Managed in UI Fields: - Conn Id: Unique name - Conn Type: postgres, mysql, s3, etc. - Host - Schema - Login - Password - Port - Extra: JSON """) print("\n=== What is a Hook? ===") print(""" Hook: - Interface to external system - Uses connection - Provides methods - Reusable Example: from airflow.providers.postgres.hooks.postgres import PostgresHook hook = PostgresHook(postgres_conn_id='my_postgres') conn = hook.get_conn() cursor = conn.cursor() cursor.execute("SELECT * FROM users") """) print("\n=== Common Connection Types ===") print(""" - postgres - mysql - sqlite - s3 - gcs - azure_blob - http - ssh - slack - discord - aws - google_cloud - kubernetes """) print("\n=== Managing Connections ===") print(""" UI: Admin → Connections → Add CLI: airflow connections add my_postgres \\ --conn-type postgres \\ --conn-host localhost \\ --conn-schema mydb \\ --conn-login user \\ --conn-password pass Environment: AIRFLOW_CONN_MY_POSTGRES='postgres://user:pass@localhost:5432/mydb' """) # IMPLEMENTATION print("\n=== Simulating Connections & Hooks ===") class Connection: """Simulate Airflow Connection.""" def __init__(self, conn_id, conn_type, host, schema, login, password, port=None): self.conn_id = conn_id self.conn_type = conn_type self.host = host self.schema = schema self.login = login self.password = password self.port = port def get_uri(self): """Get connection URI.""" return f"{self.conn_type}://{self.login}:{self.password}@{self.host}:{self.port}/{self.schema}" def get_info(self): return { 'conn_id': self.conn_id, 'type': self.conn_type, 'host': self.host, 'schema': self.schema, } class BaseHook: """Simulate Airflow Hook.""" def __init__(self, conn_id): self.conn_id = conn_id self.conn = None def get_conn(self): """Get connection.""" raise NotImplementedError class PostgresHook(BaseHook): """Simulate PostgresHook.""" def __init__(self, postgres_conn_id): super().__init__(postgres_conn_id) def get_conn(self): """Get connection.""" return Connection( self.conn_id, 'postgres', 'localhost', 'mydb', 'user', 'pass', 5432 ) def run(self, sql): """Run SQL.""" conn = self.get_conn() return f"Executed: {sql} on {conn.host}/{conn.schema}" class S3Hook(BaseHook): """Simulate S3Hook.""" def __init__(self, aws_conn_id='aws_default'): super().__init__(aws_conn_id) def get_conn(self): """Get connection.""" return Connection( self.conn_id, 'aws', 's3.amazonaws.com', None, 'access_key', 'secret_key' ) def list_keys(self, bucket, prefix=''): """List keys.""" return [f"{prefix}file1.csv", f"{prefix}file2.csv"] # Demo print("=== Connection & Hook Demo ===") # Create connection conn = Connection( 'my_postgres', 'postgres', 'localhost', 'mydb', 'user', 'pass', 5432 ) print(f"Connection: {conn.get_info()}") print(f"URI: {conn.get_uri()}") # Use hook print("\nUsing PostgresHook:") pg_hook = PostgresHook('my_postgres') result = pg_hook.run("SELECT * FROM users") print(f" {result}") print("\nUsing S3Hook:") s3_hook = S3Hook() keys = s3_hook.list_keys('my-bucket', 'data/') print(f" Keys: {keys}") # CONNECTION SUMMARY print("\n=== Connection Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'Connection':<25} {'Credentials, host, port':<40}") print(f"{'Hook':<25} {'Interface to system':<40}") print(f"{'conn_id':<25} {'Unique name':<40}") print(f"{'UI':<25} {'Manage connections':<40}") print(f"{'CLI':<25} {'airflow connections':<40}")

📦 Variables & XComs

🎯 Sharing Data

Variables are global key-value stores for configuration. XComs (cross-communications) let tasks share small amounts of data.

Variables & XComs

Variables: Global config. XComs: Task-to-task. Small data: Metadata. Not for: Large data. UI: Manage. CLI: Manage.

📦 Variables📦 XComs📦 Config📦 Sharing
# Variables & XComs print("=== Variables ===") print(""" Variable: - Global key-value store - Configuration - Secrets (with backend) - Managed in UI Example: from airflow.models import Variable # Get my_var = Variable.get("my_variable") # Set Variable.set("my_variable", "my_value") # JSON config = Variable.get("config", deserialize_json=True) """) print("\n=== XComs ===") print(""" XCom (Cross-Communication): - Task-to-task data - Small data - Metadata - Not for large data Push: task_instance.xcom_push(key='result', value=data) # Or return from task def my_task(): return "result" Pull: value = task_instance.xcom_pull(key='result', task_ids='upstream_task') # Or via context def my_task(**context): value = context['ti'].xcom_pull(task_ids='upstream_task') """) print("\n=== Example ===") print(""" from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def extract(**context): data = {'records': 1000} context['ti'].xcom_push(key='extracted_data', value=data) return data def transform(**context): data = context['ti'].xcom_pull(key='extracted_data', task_ids='extract') transformed = {'records': data['records'] * 2} return transformed def load(**context): data = context['ti'].xcom_pull(task_ids='transform') print(f"Loading {data['records']} records") with DAG('etl', schedule='@daily', start_date=datetime(2024,1,1)) as dag: t1 = PythonOperator(task_id='extract', python_callable=extract) t2 = PythonOperator(task_id='transform', python_callable=transform) t3 = PythonOperator(task_id='load', python_callable=load) t1 >> t2 >> t3 """) # IMPLEMENTATION print("\n=== Simulating Variables & XComs ===") class VariableStore: """Simulate Airflow Variables.""" def __init__(self): self.variables = {} def set(self, key, value): """Set variable.""" self.variables[key] = value print(f"Variable: Set '{key}'") def get(self, key, default=None): """Get variable.""" return self.variables.get(key, default) def delete(self, key): """Delete variable.""" if key in self.variables: del self.variables[key] print(f"Variable: Deleted '{key}'") class XComStore: """Simulate Airflow XComs.""" def __init__(self): self.xcoms = defaultdict(dict) # task_id -> {key -> value} def push(self, task_id, key, value): """Push XCom.""" self.xcoms[task_id][key] = value print(f"XCom: {task_id} pushed '{key}'") def pull(self, task_id, key, default=None): """Pull XCom.""" return self.xcoms[task_id].get(key, default) def get_all(self, task_id): return self.xcoms[task_id] # Demo print("=== Variables & XComs Demo ===") # Variables print("1. Variables:") var_store = VariableStore() var_store.set('environment', 'production') var_store.set('max_retries', 3) var_store.set('config', {'batch_size': 1000, 'timeout': 300}) print(f" environment: {var_store.get('environment')}") print(f" max_retries: {var_store.get('max_retries')}") print(f" config: {var_store.get('config')}") # XComs print("\n2. XComs:") xcom_store = XComStore() # Task 1 pushes xcom_store.push('extract', 'row_count', 1000) xcom_store.push('extract', 'source', 'postgres') # Task 2 pulls row_count = xcom_store.pull('extract', 'row_count') source = xcom_store.pull('extract', 'source') print(f" transform pulled: row_count={row_count}, source={source}") # Task 2 pushes xcom_store.push('transform', 'transformed_count', row_count * 2) # Task 3 pulls transformed = xcom_store.pull('transform', 'transformed_count') print(f" load pulled: transformed_count={transformed}") print(f"\nAll XComs:") for task_id, xcoms in xcom_store.xcoms.items(): print(f" {task_id}: {xcoms}") # VARIABLE SUMMARY print("\n=== Variables & XComs Summary ===") print(f"{'Concept':<25} {'Purpose':<40}") print("-" * 65) print(f"{'Variable':<25} {'Global config':<40}") print(f"{'XCom':<25} {'Task-to-task data':<40}") print(f"{'xcom_push':<25} {'Send data':<40}") print(f"{'xcom_pull':<25} {'Receive data':<40}") print(f"{'Small data only':<25} {'Not for large datasets':<40}")

📡 Sensors

🎯 Waiting for Conditions

Sensors are special operators that wait for a condition to be met before proceeding. They're essential for event-driven workflows.

Sensors

Wait: For condition. Poll: Check periodically. Timeout: Fail if too long. Reschedule: Free up slot. Types: File, S3, HTTP, SQL. Custom: User-defined.

📡 Wait📡 Poll📡 Timeout📡 Reschedule
# Sensors print("=== What is a Sensor? ===") print(""" Sensor: - Waits for condition - Polls periodically - Times out - Blocks or reschedules Types: - FileSensor - S3KeySensor - HttpSensor - SqlSensor - ExternalTaskSensor - DateTimeSensor - Custom """) print("\n=== Common Sensors ===") print(""" 1. FileSensor - Wait for file FileSensor( task_id='wait_for_file', filepath='/data/input.csv', poke_interval=60, timeout=3600, ) 2. S3KeySensor - Wait for S3 key S3KeySensor( task_id='wait_for_s3', bucket_name='my-bucket', bucket_key='data/*.csv', poke_interval=300, timeout=7200, ) 3. ExternalTaskSensor - Wait for another DAG ExternalTaskSensor( task_id='wait_for_upstream', external_dag_id='upstream_dag', external_task_id='final_task', ) 4. SqlSensor - Wait for SQL condition SqlSensor( task_id='wait_for_data', conn_id='my_db', sql='SELECT COUNT(*) FROM table', success=lambda x: x > 0, ) """) print("\n=== Sensor Modes ===") print(""" 1. POKE (default) - Occupies worker slot - Polls periodically - Good for short waits 2. RESCHEDULE - Frees worker slot - Reschedules itself - Good for long waits FileSensor( task_id='wait', mode='reschedule', poke_interval=300, ) """) # IMPLEMENTATION print("\n=== Simulating Sensors ===") import time import random class Sensor: """Simulate Airflow Sensor.""" def __init__(self, task_id, poke_interval=5, timeout=60, mode='poke'): self.task_id = task_id self.poke_interval = poke_interval self.timeout = timeout self.mode = mode self.state = 'pending' self.pokes = 0 def poke(self): """Check condition.""" raise NotImplementedError def execute(self): """Execute sensor.""" self.state = 'running' start = time.time() while time.time() - start < self.timeout: self.pokes += 1 if self.poke(): self.state = 'success' return True time.sleep(0.1) # Simulate poll interval self.state = 'failed' return False class FileSensor(Sensor): """Simulate FileSensor.""" def __init__(self, task_id, filepath, **kwargs): super().__init__(task_id, **kwargs) self.filepath = filepath self.poke_count = 0 def poke(self): """Check if file exists.""" self.poke_count += 1 # Simulate: file appears after 3 pokes return self.poke_count >= 3 class S3KeySensor(Sensor): """Simulate S3KeySensor.""" def __init__(self, task_id, bucket, key, **kwargs): super().__init__(task_id, **kwargs) self.bucket = bucket self.key = key self.poke_count = 0 def poke(self): """Check if S3 key exists.""" self.poke_count += 1 return self.poke_count >= 5 # Demo print("=== Sensor Demo ===") # FileSensor print("1. FileSensor:") file_sensor = FileSensor( 'wait_for_file', '/data/input.csv', poke_interval=1, timeout=10, ) result = file_sensor.execute() print(f" Result: {result}, State: {file_sensor.state}") print(f" Pokes: {file_sensor.pokes}") # S3KeySensor print("\n2. S3KeySensor:") s3_sensor = S3KeySensor( 'wait_for_s3', 'my-bucket', 'data/file.csv', poke_interval=1, timeout=10, ) result = s3_sensor.execute() print(f" Result: {result}, State: {s3_sensor.state}") print(f" Pokes: {s3_sensor.pokes}") # SENSOR SUMMARY print("\n=== Sensor Summary ===") print(f"{'Sensor':<25} {'Waits For':<40}") print("-" * 65) print(f"{'FileSensor':<25} {'File exists':<40}") print(f"{'S3KeySensor':<25} {'S3 key exists':<40}") print(f"{'HttpSensor':<25} {'HTTP response':<40}") print(f"{'SqlSensor':<25} {'SQL condition':<40}") print(f"{'ExternalTaskSensor':<25} {'Another DAG':<40}") print(f"{'DateTimeSensor':<25} {'Specific time':<40}")

✅ Best Practices

🎯 Writing Good DAGs

Follow best practices for maintainable, scalable, and reliable Airflow workflows.

Best Practices

Idempotent: Safe retries. Atomic: Small tasks. Deterministic: Same result. No top-level code: Avoid DB queries. Use templates: Jinja. Test: Unit tests.

✅ Idempotent✅ Atomic✅ Deterministic✅ Testable
# Best Practices print("=== DAG Design ===") print(""" 1. IDEMPOTENT TASKS - Safe to re-run - Same input → same output - No side effects on retry 2. ATOMIC TASKS - One thing per task - Small, focused - Easy to retry 3. DETERMINISTIC - Same input → same output - No random without seed - No time-dependent logic 4. NO TOP-LEVEL CODE - No DB queries - No API calls - No heavy imports - Only DAG definition 5. USE TEMPLATES - Jinja templates - {{ ds }} (date) - {{ params }} - {{ ti.xcom_pull() }} 6. TEST LOCALLY - Unit tests - Integration tests - airflow tasks test """) print("\n=== Code Examples ===") print(""" # BAD: Top-level code from airflow import DAG import pandas as pd df = pd.read_csv('s3://bucket/file.csv') # Executes on every parse! with DAG('bad_dag') as dag: ... # GOOD: Only DAG definition from airflow import DAG from airflow.operators.python import PythonOperator def process_data(): import pandas as pd df = pd.read_csv('s3://bucket/file.csv') return len(df) with DAG('good_dag') as dag: task = PythonOperator( task_id='process', python_callable=process_data, ) """) print("\n=== Idempotency ===") print(""" BAD: Not idempotent def insert_data(): # Inserts without checking db.insert(data) GOOD: Idempotent def insert_data(): # Upsert db.upsert(data, key='id') # Or delete + insert def replace_data(): db.delete('table') db.insert(data) """) print("\n=== Task Granularity ===") print(""" BAD: Too coarse task = PythonOperator( task_id='etl', python_callable=do_everything, # 1 hour ) GOOD: Atomic tasks extract = PythonOperator(task_id='extract', ...) transform = PythonOperator(task_id='transform', ...) load = PythonOperator(task_id='load', ...) extract >> transform >> load """) print("\n=== Configuration ===") print(""" Use Variables for config: batch_size = Variable.get('batch_size', default_var=1000) environment = Variable.get('environment') Use Connections for credentials: hook = PostgresHook(postgres_conn_id='my_db') """) # IMPLEMENTATION print("\n=== Best Practices Checklist ===") class DAGReview: """Review DAG against best practices.""" def __init__(self, name): self.name = name self.checks = [] def check(self, name, passed, message=''): """Add check result.""" self.checks.append({ 'name': name, 'passed': passed, 'message': message, }) def get_score(self): passed = sum(1 for c in self.checks if c['passed']) total = len(self.checks) return passed, total, passed / total * 100 if total > 0 else 0 def report(self): print(f"\n=== DAG Review: {self.name} ===") for check in self.checks: status = "✓" if check['passed'] else "✗" print(f" {status} {check['name']}") if check['message']: print(f" {check['message']}") passed, total, score = self.get_score() print(f"\n Score: {passed}/{total} ({score:.0f}%)") # Demo print("=== DAG Review Demo ===") review = DAGReview('etl_pipeline') # Run checks review.check('Idempotent tasks', True, 'Uses upsert') review.check('Atomic tasks', True, 'One thing per task') review.check('Deterministic', True, 'No random logic') review.check('No top-level code', True, 'Only DAG definition') review.check('Uses templates', False, 'Consider Jinja templates') review.check('Has retries', True, 'retries=2') review.check('Has alerts', False, 'Add email_on_failure') review.check('Documentation', True, 'doc_md set') review.report() # BEST PRACTICES SUMMARY print("\n=== Best Practices Summary ===") print(f"{'Practice':<30} {'Why':<35}") print("-" * 70) print(f"{'Idempotent tasks':<30} {'Safe retries':<35}") print(f"{'Atomic tasks':<30} {'Easy debugging':<35}") print(f"{'Deterministic':<30} {'Reproducible':<35}") print(f"{'No top-level code':<30} {'Fast parsing':<35}") print(f"{'Use templates':<30} {'Dynamic values':<35}") print(f"{'Test locally':<30} {'Catch bugs early':<35}")

⚖️ Airflow vs Alternatives

🎯 Comparing Orchestrators

Airflow is one of many workflow orchestrators. Compare with Dagster, Prefect, Argo Workflows, and Luigi.

Comparison

Airflow: Python, mature. Dagster: Data-aware. Prefect: Dynamic. Argo: Kubernetes. Luigi: Spotify. Choose: Based on needs.

⚖️ Airflow⚖️ Dagster⚖️ Prefect⚖️ Argo
# Airflow vs Alternatives print("=== Orchestrator Comparison ===") print(f"{'Aspect':<25} {'Airflow':<20} {'Dagster':<20} {'Prefect':<20}") print("-" * 85) print(f"{'Language':<25} {'Python':<20} {'Python':<20} {'Python':<20}") print(f"{'DAG definition':<25} {'Python':<20} {'Python':<20} {'Python':<20}") print(f"{'Scheduling':<25} {'Cron':<20} {'Cron':<20} {'Cron':<20}") print(f"{'UI':<25} {'Mature':<20} {'Modern':<20} {'Modern':<20}") print(f"{'Data-aware':<25} {'No':<20} {'Yes':<20} {'No':<20}") print(f"{'Dynamic':<25} {'Limited':<20} {'Good':<20} {'Excellent':<20}") print(f"{'Ecosystem':<25} {'Largest':<20} {'Growing':<20} {'Growing':<20}") print(f"{'Maturity':<25} {'Mature':<20} {'Modern':<20} {'Modern':<20}") print(f"{'Managed':<25} {'Many':<20} {'Dagster Cloud':<20} {'Prefect Cloud':<20}") print("\n=== When to Use Which ===") print(""" USE AIRFLOW: ✓ Mature ecosystem ✓ Large community ✓ Many integrations ✓ Enterprise adoption USE DAGSTER: ✓ Data-aware pipelines ✓ Software-defined assets ✓ Type safety ✓ Modern developer experience USE PREFECT: ✓ Dynamic workflows ✓ Python-native ✓ Easy local development ✓ Modern UI USE ARGO: ✓ Kubernetes-native ✓ Container workflows ✓ GitOps USE LUIGI: ✓ Simple pipelines ✓ Spotify ecosystem ✓ Batch jobs """) print("\n=== Feature Comparison ===") print(f"{'Feature':<25} {'Airflow':<15} {'Dagster':<15} {'Prefect':<15} {'Argo':<15}") print("-" * 85) print(f"{'DAGs':<25} {'Yes':<15} {'Yes':<15} {'Yes':<15} {'Yes':<15}") print(f"{'Python':<25} {'Yes':<15} {'Yes':<15} {'Yes':<15} {'No':<15}") print(f"{'YAML':<25} {'No':<15} {'No':<15} {'No':<15} {'Yes':<15}") print(f"{'Kubernetes':<25} {'Yes':<15} {'Yes':<15} {'Yes':<15} {'Native':<15}") print(f"{'Data-aware':<25} {'No':<15} {'Yes':<15} {'No':<15} {'No':<15}") print(f"{'Dynamic':<25} {'Limited':<15} {'Good':<15} {'Excellent':<15} {'Good':<15}") # COMPARISON SUMMARY print("\n=== Summary ===") print(f"{'Tool':<20} {'Best For':<40}") print("-" * 65) print(f"{'Airflow':<20} {'General-purpose orchestration':<40}") print(f"{'Dagster':<20} {'Data-aware pipelines':<40}") print(f"{'Prefect':<20} {'Dynamic Python workflows':<40}") print(f"{'Argo':<20} {'Kubernetes-native':<40}") print(f"{'Luigi':<20} {'Simple batch pipelines':<40}")

💼 Applications

🎯 Real-World Use Cases

Airflow orchestrates workflows across every industry: ETL, ML, data warehousing, and more.

Applications

ETL: Data pipelines. ML: Training pipelines. Warehouse: Loading. Analytics: Reports. Batch: Processing. Cloud: Infrastructure.

💼 ETL💼 ML💼 Warehouse💼 Analytics
# Applications print("=== Industry Use Cases ===") print(f"{'Industry':<20} {'Use Case':<30} {'Frequency':<20}") print("-" * 70) print(f"{'E-commerce':<20} {'Order processing':<30} {'Hourly':<20}") print(f"{'Finance':<20} {'Risk reports':<30} {'Daily':<20}") print(f"{'Healthcare':<20} {'Patient data':<30} {'Daily':<20}") print(f"{'Media':<20} {'Content pipelines':<30} {'Hourly':<20}") print(f"{'Retail':<20} {'Inventory sync':<30} {'Hourly':<20}") print(f"{'Tech':<20} {'ML training':<30} {'Weekly':<20}") print("\n=== 1. ETL Pipelines ===") print(""" Use case: - Extract from sources - Transform data - Load to warehouse DAG: extract >> transform >> load >> validate >> notify Tools: - Airflow - Spark - dbt - Snowflake """) print("\n=== 2. ML Pipelines ===") print(""" Use case: - Data preparation - Feature engineering - Model training - Model evaluation - Deployment DAG: preprocess >> features >> train >> evaluate >> deploy Tools: - Airflow - MLflow - SageMaker - Kubeflow """) print("\n=== 3. Data Warehouse Loading ===") print(""" Use case: - Load data from sources - Transform for analytics - Update dashboards DAG: extract >> stage >> transform >> load >> refresh Tools: - Airflow - dbt - Fivetran - Snowflake """) print("\n=== 4. Analytics Reports ===") print(""" Use case: - Generate reports - Send emails - Update dashboards DAG: query >> format >> send_email >> update_dashboard Tools: - Airflow - Tableau - Looker - Metabase """) # IMPLEMENTATION print("\n=== Example: ETL DAG ===") class ETLDAG: """Simulate ETL DAG.""" def __init__(self): self.tasks = [] self.results = {} def add_task(self, name, func): self.tasks.append({'name': name, 'func': func}) def run(self): for task in self.tasks: result = task['func']() self.results[task['name']] = result print(f" {task['name']}: {result}") # Create ETL DAG etl = ETLDAG() etl.add_task('extract', lambda: "Extracted 10000 rows from Postgres") etl.add_task('transform', lambda: "Transformed 10000 rows, cleaned data") etl.add_task('load', lambda: "Loaded 10000 rows to Snowflake") etl.add_task('validate', lambda: "Validation passed") etl.add_task('notify', lambda: "Email sent to team") print("Running ETL DAG:") etl.run() # APPLICATION SUMMARY print("\n=== Application Summary ===") print(f"{'Use Case':<25} {'Description':<40}") print("-" * 65) print(f"{'ETL':<25} {'Data pipelines':<40}") print(f"{'ML':<25} {'Training pipelines':<40}") print(f"{'Warehouse':<25} {'Loading data':<40}") print(f"{'Analytics':<25} {'Reports':<40}") print(f"{'Batch':<25} {'Processing':<40}") print(f"{'Cloud':<25} {'Infrastructure':<40}")

🚀 Future

🎯 Where Airflow Is Going

Airflow is evolving: cloud-native, better scheduling, dynamic DAGs, and AI-driven orchestration.

Future Directions

Cloud: Managed services. Dynamic: DAGs at runtime. Data-aware: Dataset scheduling. AI: Auto-optimization. Serverless: No ops. Unified: One platform.

🚀 Cloud🚀 Dynamic🚀 Dataset🚀 AI
# Future of Airflow print("=== 1. Cloud-Native ===") print(""" Trend: Managed services Services: - Google Cloud Composer - Amazon MWAA - Azure Data Factory - Astronomer - Astronomer Astro Benefits: - No infrastructure - Auto-scaling - Pay per use Future: - Serverless default - Event-driven - Cost-effective """) print("\n=== 2. Dynamic DAGs ===") print(""" Trend: Runtime DAG generation Features: - Dynamic task mapping - Task groups - Expandable tasks Example: # Map over a list expand( task_id='process', my_input=task1.output, ) Future: - More dynamic - Data-driven - Less boilerplate """) print("\n=== 3. Dataset-Aware Scheduling ===") print(""" Trend: Data-driven scheduling Dataset: - Data asset - Trigger DAGs on update - Cross-DAG dependencies Example: # Producer @dag(schedule='@daily') def producer(): task = produce_data() task.outlets = [Dataset('s3://bucket/data')] # Consumer @dag(schedule=[Dataset('s3://bucket/data')]) def consumer(): ... Future: - Standard scheduling - Data lineage - Event-driven """) print("\n=== 4. 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=== 5. Kubernetes-Native ===") print(""" Trend: Kubernetes-first Features: - KubernetesExecutor - Pod-based tasks - Helm charts - Operators Benefits: - Cloud-native - Elastic - Container-based Future: - Kubernetes default - Serverless pods - Multi-cloud """) print("\n=== 6. Unified Platform ===") print(""" Trend: One platform for all Features: - DAGs - ML pipelines - Data quality - Data lineage - Observability Future: - Unified analytics - Data-aware - ML integration """) # TIMELINE print("\n=== Timeline ===") print(""" Past: 2014: Airflow started (Airbnb) 2015: Open sourced 2016: Apache incubator 2019: Apache top-level 2020: Airflow 2.0 2023: Airflow 2.7 Present: 2024: Airflow 2.9 2025: Dataset scheduling Future: 2030: Serverless? 2035: AI-native? 2040: New paradigms? """) print("\n=== Final Thoughts ===") print(""" Airflow is: - Workflow orchestration - DAGs as code - Python-based - Mature - Widely adopted The future is: - Cloud-native - Dynamic - Data-aware - AI-driven - Unified "Orchestrate everything. Airflow is the conductor of your data symphony." Keep exploring. Keep building. """)