ZenML 2026: 将 20+ 工具连接成生产级流水线的 MLOps 框架 —— 完整配置指南

python -m venv zenml-env
source zenml-env/bin/activate  # Linux/Mac
# zenml-env\Scripts\activate  # Windows

# 安装 ZenML 核心
pip install zenml

# 验证安装
zenml version
# Output: ZenML version 0.80.0

# 初始化 ZenML 仓库（创建 .zen 目录）
zenml init

# 检查状态
zenml status

# 注册本地制品仓库
zenml artifact-store register local_store --flavor=local --path=./artifacts

# 注册本地编排器
zenml orchestrator register local_orchestrator --flavor=local

# 创建组合技术栈
zenml stack register local_stack \
  -o local_orchestrator \
  -a local_store \
  --set

# 验证当前技术栈
zenml stack describe

from zenml import pipeline, step
import pandas as pd
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

@step
def load_data() -> pd.DataFrame:
    """Load the iris dataset."""
    iris = load_iris(as_frame=True)
    df = iris.frame
    return df

@step
def split_data(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
    """Split data into training and test sets."""
    X = df.drop("target", axis=1)
    y = df["target"]
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    return X_train, X_test, y_train, y_test

@step
def train_model(X_train: pd.DataFrame, y_train: pd.Series) -> RandomForestClassifier:
    """Train a Random Forest classifier."""
    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    clf.fit(X_train, y_train)
    return clf

@step
def evaluate_model(
    model: RandomForestClassifier,
    X_test: pd.DataFrame,
    y_test: pd.Series
) -> float:
    """Evaluate the trained model."""
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    print(f"Model accuracy: {accuracy:.4f}")
    return accuracy

@pipeline
def training_pipeline():
    """End-to-end ML training pipeline."""
    df = load_data()
    X_train, X_test, y_train, y_test = split_data(df)
    model = train_model(X_train, y_train)
    accuracy = evaluate_model(model, X_test, y_test)

if __name__ == "__main__":
    run = training_pipeline()
    print(f"Pipeline run completed: {run.name}")

python first_pipeline.py

# 安装 Airflow 集成
pip install zenml[airflow]

# 注册 Airflow 编排器
zenml orchestrator register airflow_orchestrator \
  --flavor=airflow \
  --local=True

# 切换到 Airflow 技术栈
zenml stack update local_stack -o airflow_orchestrator

# 安装 MLflow 集成
pip install zenml[mlflow]

# 启动 MLflow UI（在另一个终端中）
mlflow ui --port 5000

# 注册 MLflow 实验追踪器
zenml experiment-tracker register mlflow_tracker \
  --flavor=mlflow \
  --tracking_uri=http://localhost:5000

# 注册 MLflow 模型注册表
zenml model-registry register mlflow_registry \
  --flavor=mlflow \
  --uri=http://localhost:5000

# 更新技术栈
zenml stack update local_stack \
  -e mlflow_tracker \
  -r mlflow_registry

from zenml import pipeline, step
from zenml.client import Client
import mlflow
import mlflow.sklearn

@step(experiment_tracker="mlflow_tracker")
def train_model(X_train: pd.DataFrame, y_train: pd.Series) -> RandomForestClassifier:
    """Train with MLflow logging."""
    mlflow.autolog()  # Auto-log parameters, metrics, and model
    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    clf.fit(X_train, y_train)
    
    # Log custom metrics
    mlflow.log_param("n_estimators", 100)
    mlflow.log_metric("train_samples", len(X_train))
    
    return clf

# Register the model after training
@step(model_registry="mlflow_registry")
def register_model(
    model: RandomForestClassifier,
    accuracy: float
) -> str:
    """Register model to MLflow model registry."""
    if accuracy > 0.90:
        model_version = mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            registered_model_name="iris-classifier"
        )
        print(f"Model registered: {model_version}")
        return "iris-classifier"
    return "below-threshold"

# 注册 S3 制品仓库
zenml artifact-store register s3_store \
  --flavor=s3 \
  --path=s3://my-ml-bucket/zenml-artifacts \
  --aws_access_key_id=$AWS_ACCESS_KEY_ID \
  --aws_secret_access_key=$AWS_SECRET_ACCESS_KEY

# 更新技术栈以使用 S3
zenml stack update local_stack -a s3_store

# 注册 Docker 容器注册表
zenml container-registry register docker_registry \
  --flavor=default \
  --uri=myregistry.azurecr.io

# 构建并运行容器化流水线
zenml stack update local_stack -c docker_registry
zenml pipeline run first_pipeline.py --build-docker

pip install zenml[wandb]

zenml experiment-tracker register wandb_tracker \
  --flavor=wandb \
  --api_key=$WANDB_API_KEY \
  --project_name="zenml-mlops"

# stack.yaml —— 将完整的 MLOps 技术栈定义为代码
stack_name: production_stack
components:
  orchestrator:
    flavor: kubernetes
    configuration:
      kubernetes_context: prod-cluster
      namespace: ml-pipelines
  artifact_store:
    flavor: s3
    configuration:
      path: s3://prod-ml-artifacts/zenml
      authentication_secret: aws-s3-secret
  container_registry:
    flavor: default
    configuration:
      uri: 123456789.dkr.ecr.us-east-1.amazonaws.com
  experiment_tracker:
    flavor: mlflow
    configuration:
      tracking_uri: http://mlflow.internal:5000
  model_registry:
    flavor: mlflow
    configuration:
      uri: http://mlflow.internal:5000
  step_operator:
    flavor: sagemaker
    configuration:
      role: arn:aws:iam::123456789:role/SageMakerRole
      instance_type: ml.p3.2xlarge

zenml stack register -f stack.yaml --set

from zenml.step_operators import BaseStepOperator

@step(step_operator="sagemaker_gpu")
def train_deep_learning_model(X_train: pd.DataFrame, y_train: pd.Series):
    """通过 SageMaker 在 GPU 上训练，其他步骤在本地运行。"""
    import tensorflow as tf
    
    # 此步骤通过 SageMaker 在 ml.p3.2xlarge 上执行
    model = tf.keras.Sequential([
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(3, activation='softmax')
    ])
    
    model.compile(optimizer='adam', loss='sparse_categorical_crossentropy')
    model.fit(X_train, y_train, epochs=50, batch_size=32)
    
    return model

from zenml.pipelines import Schedule

# 每天凌晨 3 点 UTC 运行流水线
daily_schedule = Schedule(
    cron_expression="0 3 * * *",
    pipeline_name="training_pipeline",
    stack_name="production_stack"
)

zenml.pipeline_schedule register daily_schedule

@step(enable_cache=True)  # 默认行为
def expensive_preprocessing(df: pd.DataFrame) -> pd.DataFrame:
    """仅在输入 df 或此函数更改时重新运行。"""
    # 耗时 30 分钟的繁重转换
    return processed_df

# 需要时强制重新运行
zenml pipeline run training_pipeline.py --no-cache

# 注册数据库凭据的密钥
zenml secrets-manager register aws_secrets \
  --flavor=aws \
  --region_name=us-east-1

zenml stack update local_stack -x aws_secrets

# 创建密钥
zenml secrets-manager secret register db_credentials \
  --schema=username_password \
  --username=ml_user \
  --password=$DB_PASSWORD

from zenml.client import Client

@step
def load_from_database() -> pd.DataFrame:
    """使用 ZenML 密钥管理器的凭据加载数据。"""
    client = Client()
    credentials = client.get_secret("db_credentials")
    
    import psycopg2
    conn = psycopg2.connect(
        host="db.internal",
        user=credentials.username,
        password=credentials.password
    )
    df = pd.read_sql("SELECT * FROM training_data", conn)
    return df

# .github/workflows/ml-pipeline.yml
name: ML Pipeline CI
on:
  push:
    branches: [main]
  schedule:
    - cron: "0 2 * * *"

jobs:
  train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup ZenML
        run: |
          pip install zenml[mlflow,aws]
          zenml connect --url $ZENML_SERVER_URL --api-key $ZENML_API_KEY
      
      - name: Run training pipeline
        run: |
          zenml pipeline run training_pipeline.py \
            --stack production_stack \
            --build-docker
      
      - name: Notify on failure
        if: failure()
        run: |
          curl -X POST $SLACK_WEBHOOK \
            -d '{"text":"Pipeline failed! Check ZenML dashboard."}'