ZenML 生产实战：用 e2e_batch 模板构建端到端 MLOps 项目

ZenML 生产实战：用 e2e_batch 模板构建端到端 MLOps 项目

【免费下载链接】zenmlZenML 🙏: One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml

本文基于 ZenML 生产指南的收官章节「An end-to-end project」，带你把部署 ZenML Server、定义 Stack、接入远程存储、云编排、计算扩缩容与 Git 代码仓库六大生产概念合而为一，完整走通一个可运行的端到端 MLOps 项目。读完后，你将掌握如何用zenml init模板快速生成项目、理解训练/部署/批量推理三条流水线的组织方式与配置文件，并通过项目自带的 CLI 控制整套流水线在本地或云端 Stack 上运行。

回顾：本指南覆盖的六个生产概念

在动手之前，先回顾生产指南（production guide）所铺垫的核心能力，这些概念正是本端到端项目要综合运用的：

• 部署 ZenML Server 的价值——Deploying ZenML
• 把基础设施配置抽象为 Stack——Understanding Stacks
• 接入远程存储（如 S3/GCS）——Connecting Remote Storage
• 在云上编排流水线——Orchestrating on the Cloud
• 配置流水线以扩展计算资源——Configuring the Pipeline to Scale Compute
• 关联 Git 代码仓库实现代码溯源——Connecting a Code Repository

本节将把上述概念全部组合起来，构建一个由 ZenML 驱动的端到端 MLOps 项目。生产指南整体定位见 Production Guide 入口，它建立在 Starter Guide 之上，帮助你从"在本地跑通 PoC"过渡到"在云端跑生产"。

快速开始：全新虚拟环境与依赖安装

官方建议从一个不依赖任何现有包的全新 Python 虚拟环境开始，然后安装两类依赖：

pip install "zenml[templates,server]" notebook zenml integration install sklearn -y

其中zenml[templates,server]的两个 extras 各有用途：templates提供后续zenml init --template所需的模板能力，server则允许你在本地直接拉起 ZenML Server（而不必依赖远程实例）；notebook支持在 Jupyter 中使用 ZenML。scikit-learn 集成则通过集成安装器以隔离方式引入。

用 ZenML 模板生成项目

接着用 ZenML 模板系统生成 e2e_batch 项目骨架：

mkdir zenml_batch_e2e cd zenml_batch_e2e zenml init --template e2e_batch --template-with-defaults # 保险起见，再安装一次项目依赖 pip install -r requirements.txt

• --template e2e_batch：指定使用端到端批处理（supervised ML + batch predictions）模板；
• --template-with-defaults：以默认参数回答模板问题，实现免交互生成。

备选方案：如果模板命令不可用，e2e 模板同样以示例形式存在于本仓库的 examples/e2e 目录中。你可以直接克隆 ZenML 仓库并进入examples/e2e，然后执行pip install -r requirements.txt和zenml init，即可得到与模板生成的项目等价的结构（项目 README 中对两条路径做了等价的说明）。

你将学到什么：e2e 项目的整体面貌

e2e 项目是一个覆盖 ZenML 主要使用场景的综合性项目模板：一组步骤（step）与流水线（pipeline），外加一个简单但实用的 CLI。它展示的是监督学习 + 批量预测场景下的核心 ZenML 概念，是在 starter 项目基础上叠加了更多进阶概念的版本。README 给出的默认项目属性为：

• 技术名称e2e_use_case，版本0.0.1，部署环境staging；
• 超参数与模型架构调优（配置来自 pipeline 的model_search_space参数）；
• 基于 accuracy 指标与当前已部署模型对比后，将训练出的模型晋级（promote）到staging；
• 基于 Evidently 报告的数据漂移检查；
• 流水线失败通知（notify_on_failure）。

实操建议（原文档强调）：在推进过程中，尝试把这些流水线跑在一个受追踪 Git 仓库上的远程云端 Stack上，以巩固前文学到的云编排与代码仓库概念。

项目目录结构：一条可复用的生产级骨架

从 examples/e2e 的目录组织看，项目遵循 ZenML 推荐的工程结构：

. ├── configs # 流水线配置文件 │ ├── deployer_config.yaml # 部署流水线配置 │ ├── inference_config.yaml # 批量推理流水线配置 │ └── train_config.yaml # 训练流水线配置 ├── pipelines # zenml.pipeline 实现 │ ├── batch_inference.py # [CD] 批量推理流水线 │ ├── deployment.py # [CD] 部署流水线 │ └── training.py # [CT] 训练流水线 ├── steps # 按逻辑分组的 zenml.steps 实现 │ ├── alerts # 流水线状态告警 │ ├── deployment # 部署训练好的模型对象 │ ├── data_quality # 基于漂移报告的质量门 │ ├── etl # 数据集 ETL 逻辑 │ ├── hp_tuning # 超参数与模型架构调优 │ ├── inference # 基于注册表模型的推理 │ ├── promotion # 判断新模型是否成为新的推理模型 │ └── training # 训练与评估模型 ├── utils # 辅助函数 ├── Makefile # 集成安装与本地 Stack 快速配置脚本 ├── requirements.txt # 额外 Python 依赖 └── run.py # 在 ZenML Stack 上运行流水线的 CLI

代码中所有需要你扩展的位置都用醒目的注释标出：

### ADD YOUR OWN CODE HERE - THIS IS JUST AN EXAMPLE ### ... ### YOUR CODE ENDS HERE ###

三条流水线：CT 训练、CD 部署与批量推理

e2e 项目由三条流水线组成，全部通过Model Control Plane（模型注册表 + 模型版本控制）串联：训练流水线创建并晋级一个带有训练好的模型对象的新模型版本；部署流水线基于"推理用"模型版本创建预测服务；批量推理流水线则使用该推理版本的模型做预测，并把新预测结果作为版本化数据产物存回、关联到该模型版本。这样三条流水线既紧密协作，又保证只有经过质量保障的模型版本才能产出交付给下游的预测。

训练流水线（CT）

pipelines/training.py 中e2e_use_case_training流水线按以下阶段组织（源码 L42-L138）：

1. ETL 阶段：data_loader加载 scikit-learn 的 Breast Cancer 数据集，train_data_splitter按test_size切分训练/测试集，train_data_preprocessor完成去 NA、MinMax 归一化、删列等预处理；
2. 超参调优阶段：对model_search_space参数中的每个模型配置动态生成一个hp_tuning_single_search步骤（步骤 id 形如hp_tuning_search_random_forest），再由hp_tuning_select_best_model汇聚所有搜索结果选出最优模型；
3. 训练阶段：model_trainer用最优配置训练模型，model_evaluator在 holdout 集上评估 accuracy，并支持质量门（见下文 CLI 参数）；
4. 晋级阶段：compute_performance_metrics_on_current_data计算新模型指标并与target_env（staging）中当前模型的指标对比，promote_with_metric_compare在新模型更优时执行晋级；
5. 通知：整个流水线以@pipeline(on_failure=notify_on_failure)装饰，失败时告警，成功后notify_on_success收尾。

部署流水线（CD）

pipelines/deployment.py 最为精简（源码 L23-L37）：单步deployment_deploy基于推理模型版本创建预测服务，随后notify_on_success(after=["deployment_deploy"])通知成功。

批量推理流水线（CD）

pipelines/batch_inference.py（源码 L35-L73）展示了 Model Control Plane 的典型用法——通过get_pipeline_context().model拿到当前推理模型版本，并从中取出训练期保存的产物保持 ETL 一致性：

• data_loader(random_state=model.get_artifact("random_state"), is_inference=True)——复用训练时的随机种子加载推理数据；
• inference_data_preprocessor复用preprocess_pipeline产物对推理数据做同样的预处理；
• 数据质量门：evidently_report_step以训练集dataset_trn为参考、以推理集为对比对象生成漂移报告（指标为DataQualityPreset），随后drift_quality_gate依据报告决定是否放行；
• inference_predict在质量门通过（after=["drift_quality_gate"]）后执行预测，结果作为版本化产物存回并与模型版本关联。

流水线配置文件：三个 YAML 的分工

configs/下的三个文件分别对应三条流水线，由run.py通过config_path选项注入。它们共同声明了 Docker 运行所需的集成依赖：

settings: docker: required_integrations: - aws - evidently - kubeflow - kubernetes - mlflow - sklearn - slack

这意味着流水线打包镜像时会预装这些集成（例如 MLflow 系列组件支撑实验追踪/模型注册/模型部署，Evidently 支撑漂移分析，Slack 支撑告警，Kubernetes/Kubeflow 支撑云编排）。

configs/train_config.yaml 是三者中最信息量大的一个，除 settings 外还包含：

• 步骤级参数：model_trainer.parameters.name: e2e_use_case、promote_with_metric_compare.parameters.mlflow_model_name: e2e_use_case，以及notify_on_success.parameters.notify_on_success: False（默认不发成功通知）；
• Model Control Plane 配置：model.name: e2e_use_case、license: apache、tags 等元信息；
• 流水线级参数：target_env: staging决定模型晋级到哪个环境；
• model_search_space——超参搜索空间的核心，为每个候选模型声明model_package、model_class与search_grid，例如随机森林（sklearn.ensemble.RandomForestClassifier）的搜索网格：

parameters: target_env: staging model_search_space: random_forest: model_package: sklearn.ensemble model_class: RandomForestClassifier search_grid: criterion: [gini, entropy] max_depth: [2, 4, 6, 8, 10, 12] min_samples_leaf: {range: {start: 1, end: 10}} n_estimators: {range: {start: 50, end: 500, step: 25}} decision_tree: model_package: sklearn.tree model_class: DecisionTreeClassifier search_grid: criterion: [gini, entropy] max_depth: [2, 4, 6, 8, 10, 12] min_samples_leaf: {range: {start: 1, end: 10}}

想增删候选模型，只需修改这份 YAML 即可被训练流水线的动态步骤生成逻辑感知。而 configs/deployer_config.yaml 与 configs/inference_config.yaml 结构相近，关键在于model: {name: e2e_use_case, version: staging}——它们都锚定到 staging 这个模型版本，从而与训练流水线的晋级结果对齐；三者均带有extra.notify_on_failure: True。

用 CLI 运行整套流水线：run.py 参数详解

项目自带的 run.py 是一个基于 Click 的 CLI（入口源码 L34-L213）。默认按训练 → 部署 → 批量推理的顺序依次执行三条流水线，每条流水线的run_name带时间戳（如e2e_use_case_training_run_2026_01_01_08_00_00），并通过with_options(config_path=..., run_name=...)注入各自的 YAML 配置。

完整命令行参数（均可通过python run.py --help查看）：

几个有代表性的用法（源自--help内嵌示例）：

# 默认参数运行 python run.py # 关闭缓存 python run.py --no-cache # 不做超参调优、不做 NA 删除与归一化，删除列 A,B,C，测试集占 10% python run.py --no-drop-na --no-normalize --drop-columns A,B,C --test-size 0.1 # 精度质量门：训练集 90% / 测试集 85%，任一不达标则流水线失败 python run.py --min-train-accuracy 0.9 --min-test-accuracy 0.85 --fail-on-accuracy-quality-gates

本地跑通的完整步骤：Makefile 与本地 Stack

examples/e2e/README 给出的"开箱即跑"流程（对应 Makefile 中的 target）：

# 建立虚拟环境（如尚未建立） python3 -m venv .venv source .venv/bin/activate # 安装依赖与 ZenML 集成（对应 make setup） make setup # 可选：注册默认本地 Stack（对应 make install-stack-local） make install-stack-local # 本地启动 ZenML UI（推荐但可选） zenml login --local # 运行项目包含的流水线 python run.py

其中make setup实际执行为：

pip install -r requirements.txt zenml integration install aws sklearn mlflow slack evidently kubeflow kubernetes -y

make install-stack-local则演示了 Stack 组件注册的完整命令序列（默认stack_name=e2e_template_stack，可通过环境变量覆盖）：依次注册 MLflow 实验追踪器、模型注册表、模型部署器，注册 Evidently 数据校验器，最后用一个组合命令注册 Stack 并设为活跃 Stack：

zenml experiment-tracker register -f mlflow mlflow_local_$stack_name zenml model-registry register -f mlflow mlflow_local_$stack_name zenml model-deployer register -f mlflow mlflow_local_$stack_name zenml contenteditable="false">【免费下载链接】zenmlZenML 🙏: One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml