第16课：Airflow调度

⚙️ Airflow调度

Apache Airflow是最流行的工作流编排平台，通过DAG定义、调度和监控数据管道。

🐍 Python实战：核心实战

# Airflow DAG模拟
import time, random
from datetime import datetime
from collections import defaultdict

class Task:
    def __init__(self, task_id, python_callable, retries=2):
        self.task_id = task_id; self.fn = python_callable; self.retries = retries
        self.downstream = []; self.status = 'pending'; self.duration_ms = 0
    def __rshift__(self, other): self.downstream.append(other); return other
    def execute(self, context):
        start = time.perf_counter()
        for attempt in range(1, self.retries+1):
            try:
                result = self.fn(context)
                self.status = 'success'; self.duration_ms = (time.perf_counter()-start)*1000
                return result
            except Exception as e:
                if attempt == self.retries: self.status = 'failed'; raise
                time.sleep(0.1 * attempt)

class DAG:
    def __init__(self, dag_id, schedule_interval='@daily'):
        self.dag_id = dag_id; self.schedule = schedule_interval; self.tasks = {}
    def add_task(self, task): self.tasks[task.task_id] = task; return task
    def run(self):
        print(f"DAG: {self.dag_id} ({self.schedule})")
        # Topological sort (simple)
        executed = set(); order = []
        while len(executed) < len(self.tasks):
            for tid, task in self.tasks.items():
                if tid in executed: continue
                upstream = [t.task_id for t in self.tasks.values() if tid in [d.task_id for d in t.downstream]]
                if all(u in executed for u in upstream):
                    executed.add(tid); order.append(task)
        for task in order:
            context = {'execution_date': datetime.now().isoformat()}
            try:
                task.execute(context)
                print(f"  ✅ {task.task_id}: {task.duration_ms:.0f}ms")
            except Exception as e:
                print(f"  ❌ {task.task_id}: {e}")
                break

# 构建DAG
dag = DAG('etl_pipeline', '@daily')
extract = dag.add_task(Task('extract', lambda ctx: {'records': random.randint(80,120)}))
clean = dag.add_task(Task('clean', lambda ctx: {'records': ctx.get('records',100)//2}))
transform = dag.add_task(Task('transform', lambda ctx: {'records': ctx.get('records',50)}))
load_dw = dag.add_task(Task('load_dw', lambda ctx: {'loaded': ctx.get('records',50)}))
load_es = dag.add_task(Task('load_es', lambda ctx: {'loaded': ctx.get('records',50)}))
notify = dag.add_task(Task('notify', lambda ctx: 'done'))

extract >> clean >> transform >> load_dw >> notify
transform >> load_es >> notify

dag.run()
print("✅ 验证通过 - Airflow DAG模拟运行正常")

📖 原理深入：数据编排与治理体系

🏆 最佳实践：数据编排与治理检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答

🔑 关键技术选型决策

数据量级？
├── <1GB/天 → SQLite/PostgreSQL + Python脚本
├── 1-100GB/天 → 数据仓库(Snowflake/BQ) + dbt
├── 100GB-1TB/天 → Spark + 数据湖
└── >1TB/天 → Spark/Flink + 湖仓一体

延迟要求？
├── T+1 → 批处理(Spark) + Airflow
├── 小时级 → 微批(Spark Streaming) + Airflow
├── 分钟级 → 流处理(Flink) + Kafka
└── 秒级 → 纯流(Flink) + Kafka + 缓存

数据类型？
├── 纯结构化 → 数据仓库
├── 结构化+半结构化 → 湖仓一体
└── 全类型 → 数据湖 + 元数据管理