📚 课程目录第3章数据管道编排

3、数据管道编排：Airflow / Dagster / Kubeflow Pipelines 选型与部署

数据管道编排，说白了就是给数据流程找个靠谱的「调度大脑」。

我这些年见过不少团队，代码写得挺漂亮，结果一到调度就翻车——依赖写死、重试没配、监控全靠人工盯。嗯，这其实是个大坑。

今天咱们就聊聊三个主流工具：Airflow、Dagster、Kubeflow Pipelines。我会结合自己的踩坑经验，帮你理清选型思路和部署要点。

核心观点：没有银弹。选型取决于你的团队背景、基础设施和业务场景。别盲目追新，也别死守旧工具。

3.1 三个工具的核心差异

先看一张对比表，心里有个底。

维度	Airflow	Dagster	Kubeflow Pipelines
编程模型	DAG 定义 + Operator	软件定义资产 (SDA)	组件 + 管道 DSL
调度粒度	任务级	资产级	组件级
可观测性	日志 + XCom	内置资产血缘	ML 元数据追踪
部署复杂度	中等（需数据库）	中等（需数据库）	高（依赖 K8s）
社区生态	最成熟	快速增长	K8s 生态绑定

我个人习惯先看团队背景。如果团队全是 Python 工程师，Airflow 上手最快。如果团队有数据治理需求，Dagster 的资产感知能力会让你眼前一亮。如果你们已经在 K8s 上跑 ML 训练，Kubeflow Pipelines 是自然选择。

3.2 Airflow：老牌调度器，稳如老狗

Airflow 我用了快五年。说实话，它的问题我都门儿清——调度器性能瓶颈、DAG 解析慢、依赖管理麻烦。但它的生态太强了，几乎你能想到的 Operator 都有。

部署要点

Executor 选型：小团队用 LocalExecutor，生产环境至少 CeleryExecutor，K8s 环境用 KubernetesExecutor。
数据库：PostgreSQL 是首选，MySQL 也行但别用 SQLite 上生产。
调度器配置：core.parallelism 和 core.dag_concurrency 要调优，我见过有人没调这两个参数，结果任务全堵在队列里。

我的经验：Airflow 的 DAG 解析是单线程的。如果你的 DAG 文件超过 200 个，建议开启 dagbag_import_timeout 并考虑使用 DAG 序列化缓存。

# 一个典型的 Airflow DAG 示例
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'mlops',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'feature_engineering_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    def extract_features():
        # 特征提取逻辑
        pass

    def validate_features():
        # 特征验证逻辑
        pass

    extract = PythonOperator(
        task_id='extract_features',
        python_callable=extract_features,
    )

    validate = PythonOperator(
        task_id='validate_features',
        python_callable=validate_features,
    )

    extract >> validate

避坑指南：我曾经在生产环境遇到过 DAG 解析超时导致调度器崩溃。原因是某个 DAG 文件里 import 了一个很重的库。记住，DAG 文件里只做轻量 import，重逻辑放到 Operator 里。

3.3 Dagster：新一代数据管道，资产优先

Dagster 是我最近两年开始主推的工具。它的设计理念很对我胃口——把「数据资产」作为一等公民。你想想看，传统调度器只管任务跑没跑，但 Dagster 会告诉你每个表、每个特征文件的状态。

部署要点

代码位置 (Code Location)：Dagster 的核心概念。每个代码位置对应一个 Python 包，可以独立部署和版本管理。
Dagit UI：内置的 Web 界面，比 Airflow 的 UI 现代得多。支持实时日志、资产血缘图、运行历史。
运行模式：支持 Daemon 模式（生产推荐）和 Dagit 内嵌模式（开发用）。

我的建议：如果你从 Airflow 迁移到 Dagster，别急着全量切换。先挑一个数据链路做试点，感受一下「资产感知」带来的变化。我团队迁移时，第一个月只跑了三条管道，但很快就发现了之前 Airflow 里隐藏的数据质量问题。

# Dagster 资产定义示例
from dagster import asset, Output, AssetIn
import pandas as pd

@asset
def raw_transactions() -> pd.DataFrame:
    """从源系统加载原始交易数据"""
    # 实际代码会连接数据库或API
    return pd.DataFrame()

@asset(ins={"raw_transactions": AssetIn()})
def cleaned_transactions(raw_transactions: pd.DataFrame) -> pd.DataFrame:
    """清洗交易数据，去除空值和异常值"""
    cleaned = raw_transactions.dropna()
    return cleaned

@asset(ins={"cleaned_transactions": AssetIn()})
def daily_aggregates(cleaned_transactions: pd.DataFrame) -> pd.DataFrame:
    """按天聚合交易数据"""
    daily = cleaned_transactions.groupby('date').agg({'amount': 'sum'})
    return daily

3.4 Kubeflow Pipelines：ML 工作流的 K8s 原生方案

Kubeflow Pipelines（KFP）是专门为 ML 工作流设计的。它的组件模型天然适合训练、评估、部署这种流程。但说实话，它的学习曲线比前两个都陡。

部署要点

K8s 集群：至少需要 4 核 8G 的节点，推荐 8 核 16G 以上。
存储：需要配置 MinIO 或 S3 兼容存储，用于存放组件输出和 ML 元数据。
组件镜像：每个组件都是一个容器镜像，建议用 Kaniko 或 BuildKit 构建。

避坑指南：我曾经在 KFP 上部署一个训练管道，结果因为组件镜像太大（超过 5G），每次拉取都要 10 分钟。后来我把基础镜像拆小，把训练数据挂载为 Volume，速度提升了 5 倍。

# Kubeflow Pipelines 组件定义示例
from kfp import dsl
from kfp.components import create_component_from_func

@create_component_from_func
def train_model(
    data_path: str,
    model_path: str,
    learning_rate: float = 0.01,
) -> str:
    """训练模型组件"""
    import pickle
    from sklearn.linear_model import LogisticRegression
    import pandas as pd

    data = pd.read_parquet(data_path)
    X = data.drop('target', axis=1)
    y = data['target']

    model = LogisticRegression(learning_rate=learning_rate)
    model.fit(X, y)

    with open(model_path, 'wb') as f:
        pickle.dump(model, f)

    return model_path

@dsl.pipeline(
    name='training_pipeline',
    description='端到端模型训练管道'
)
def training_pipeline(data_path: str):
    train_op = train_model(
        data_path=data_path,
        model_path='/models/model.pkl',
        learning_rate=0.01,
    )

3.5 选型决策矩阵

我整理了一个决策矩阵，你可以根据实际情况打分。

评估维度	权重	Airflow	Dagster	Kubeflow Pipelines
团队 Python 熟练度	高	★★★★★	★★★★☆	★★★☆☆
数据质量要求	中	★★★☆☆	★★★★★	★★★★☆
K8s 基础设施	高	★★★★☆	★★★★☆	★★★★★
ML 工作流占比	中	★★★☆☆	★★★★☆	★★★★★
运维复杂度容忍度	低	★★★★☆	★★★★☆	★★☆☆☆

3.6 部署实战：最小化生产部署

不管选哪个工具，部署时都有一些共性原则。我直接给一个最小化生产部署清单。

数据库：PostgreSQL 13+，开启连接池（PgBouncer 或内置池）。
消息队列：Airflow 用 Redis/RabbitMQ，Dagster 用 RabbitMQ，KFP 用 K8s 内部通信。
日志收集：统一输出到 stdout，用 EFK/ELK 收集。
监控告警：Prometheus + Grafana，关注调度延迟、任务失败率、队列积压。
CI/CD：DAG/资产定义和管道代码走 GitOps 流程，合并到 main 自动部署。

我的部署原则：先让管道跑起来，再谈优化。别一开始就追求高可用、多副本。单节点跑一个月，把坑踩完，再上集群。

3.7 总结

选型没有标准答案。Airflow 适合传统 ETL 团队，Dagster 适合数据质量敏感的场景，Kubeflow Pipelines 适合 ML 原生团队。

我个人建议：如果你还在犹豫，先从 Airflow 开始。它最成熟，踩坑成本最低。等团队成熟了，再考虑迁移到 Dagster 或 KFP。

嗯，工具只是手段。把数据管道跑稳、跑快、跑对，才是目的。