MLOps：在 Azure Databricks 上为 ML 模型构建 CI/CD 流水线

大多数 ML 团队擅长训练模型，但很少擅长将模型交付上线。一个能运行的 Notebook 与一个能够可靠服务生产流量的模型之间的差距，正是大多数 ML 项目停滞不前的地方。

在本教程中，我将带领你构建一个针对 Azure Databricks 上的 ML 模型的完整 CI/CD 流水线，使用：

• MLflow 进行实验跟踪、模型版本控制以及模型注册表
• Databricks Asset Bundles (DABs) 实现基础设施即代码和任务部署
• Azure DevOps / GitHub Actions 作为 CI/CD 编排器
• Delta Lake 作为特征数据和验证数据的存储
• Databricks Model Serving 作为生产端点

用例与上一篇文章中的流失预测模型相同，但这次我们完全聚焦于如何将该模型从 Notebook 可靠且可重复地部署到生产端点。

---

架构概览

---

CI/CD 流程

---

流水线阶段解析

| 阶段 | 工具 | 发生的事件 | 进入下一阶段的关卡 | | --- | --- | --- | --- | | CI | GitHub Actions | 代码检查、单元测试、bundle 验证 | 所有测试通过 | | 训练 | Databricks Job | 完整训练运行、MLflow 日志记录 | 任务退出码为0 | | 注册 | MLflow 注册表 | 模型版本化并移至 Staging | 训练成功时自动触发 | | 验证 | Databricks Job | 在保留集上检查指标阈值 | ROC-AUC >= 0.80 | | 提升 | MLflow 注册表 | 模型移至 Production | 验证通过 | | 部署 | Model Serving | 端点更新为新的模型版本 | 提升完成 | | 监控 | Databricks Lakehouse Monitoring | 部署后跟踪漂移和准确性 | 持续进行 |

---

步骤 1 — 使用 Databricks Asset Bundles 的项目结构

Databricks Asset Bundles (DABs) 允许你将任务、集群和流水线定义为代码，并通过 CLI 进行部署。这就是你的 IaC 层。

# databricks.yml
bundle:
  name: churn-prediction

variables:
  env:
    default: dev

targets:
  dev:
    mode: development
    default: true
    workspace:
      host: https://adb-xxxx.azuredatabricks.net

  staging:
    workspace:
      host: https://adb-xxxx.azuredatabricks.net

  production:
    mode: production
    workspace:
      host: https://adb-xxxx.azuredatabricks.net

resources:
  jobs:
    training_job:
      name: churn-training-${var.env}
      tasks:
        - task_key: feature_engineering
          notebook_task:
            notebook_path: ./notebooks/01_feature_engineering.py
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS3_v2
            num_workers: 4

        - task_key: train_and_register
          depends_on:
            - task_key: feature_engineering
          notebook_task:
            notebook_path: ./notebooks/02_train_and_register.py
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS3_v2
            num_workers: 2

    validation_job:
      name: churn-validation-${var.env}
      tasks:
        - task_key: validate_model
          notebook_task:
            notebook_path: ./notebooks/03_validate_and_promote.py
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS3_v2
            num_workers: 2

---

步骤 2 — 使用 MLflow 自动日志记录的训练任务

保持训练 Notebook 简洁且专注。让 MLflow 自动日志处理指标和参数记录的重活。

# notebooks/02_train_and_register.py
import mlflow
import mlflow.sklearn
from mlflow.models.signature import infer_signature
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, f1_score, precision_score, recall_score
from delta.tables import DeltaTable
import pandas as pd

mlflow.set_experiment('/churn-prediction/ci-cd-pipeline')
mlflow.sklearn.autolog(log_input_examples=True, log_model_signatures=True)

# Read Gold features and capture Delta version for reproducibility
gold_table    = DeltaTable.forName(spark, 'churn.gold.features')
delta_version = gold_table.history(1).select('version').collect()[0][0]

features_pdf = spark.table('churn.gold.features') \
    .filter("feature_date = current_date()") \
    .toPandas()

FEATURE_COLS = [
    'total_events', 'total_sessions', 'distinct_products',
    'events_last_30d', 'events_last_90d', 'days_since_last_event',
    'customer_tenure_days', 'avg_events_per_day',
    'recency_tier', 'engagement_score',
]
TARGET = 'churned'

X = features_pdf[FEATURE_COLS]
y = features_pdf[TARGET]
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

with mlflow.start_run() as run:
    params = {
        'n_estimators':  200,
        'max_depth':     5,
        'learning_rate': 0.05,
        'subsample':     0.8,
    }
    model = GradientBoostingClassifier(**params, random_state=42)
    model.fit(X_train, y_train)

    y_prob = model.predict_proba(X_test)[:, 1]
    y_pred = model.predict(X_test)

    mlflow.log_param('delta_feature_version', delta_version)
    mlflow.log_param('git_sha', dbutils.widgets.get('git_sha'))  # passed in by CI
    mlflow.log_metric('roc_auc',   roc_auc_score(y_test, y_prob))
    mlflow.log_metric('f1_score',  f1_score(y_test, y_pred))
    mlflow.log_metric('precision', precision_score(y_test, y_pred))
    mlflow.log_metric('recall',    recall_score(y_test, y_pred))

    signature = infer_signature(X_train, y_pred)
    mlflow.sklearn.log_model(
        model,
        artifact_path='churn-gbm',
        signature=signature,
        registered_model_name='churn-prediction-gbm',
        await_registration_for=300,
    )

    # Pass run ID to downstream tasks via job output
    dbutils.jobs.taskValues.set(key='run_id', value=run.info.run_id)
    print(f"Training complete. Run ID: {run.info.run_id}")

---

步骤 3 — 验证任务：在提升前根据指标设置关卡

切勿仅凭成功的训练运行就自动提升模型。在触及 Production 别名之前，始终在保留集或最新数据集上进行验证，并对照阈值进行检查。

# notebooks/03_validate_and_promote.py
import mlflow
from mlflow import MlflowClient
from sklearn.metrics import roc_auc_score
import pandas as pd

client    = MlflowClient()
MODEL_NAME = 'churn-prediction-gbm'

# Pick up run_id from the upstream training task
run_id = dbutils.jobs.taskValues.get(
    taskKey='train_and_register', key='run_id'
)

# Thresholds — fail the job if any metric is below these
THRESHOLDS = {
    'roc_auc':   0.80,
    'f1_score':  0.72,
    'precision': 0.70,
}

run     = client.get_run(run_id)
metrics = run.data.metrics

print("Validating metrics against thresholds...")
failures = []
for metric, threshold in THRESHOLDS.items():
    actual = metrics.get(metric, 0)
    status = 'PASS' if actual >= threshold else 'FAIL'
    print(f"  {metric}: {actual:.4f} (threshold: {threshold}) -> {status}")
    if actual < threshold:
        failures.append(f"{metric}={actual:.4f} below threshold {threshold}")

if failures:
    raise Exception(f"Validation failed: {', '.join(failures)}")

# Promote to Production alias if all thresholds pass
model_version = client.search_model_versions(
    filter_string=f"run_id='{run_id}'"
)[0].version

client.set_registered_model_alias(
    name=MODEL_NAME,
    alias='Production',
    version=model_version,
)

print(f"Model version {model_version} promoted to Production alias.")

---

步骤 4 — GitHub Actions CI/CD 工作流

这是将所有部分串联起来的粘合剂。一个工作流处理 PR 验证；另一个处理合并到 main 分支时的部署。

# .github/workflows/mlops.yml
name: MLOps CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

env:
  DATABRICKS_HOST:  ${{ secrets.DATABRICKS_HOST }}
  DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}

jobs:
  ci:
    name: Lint and Test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install databricks-cli databricks-sdk pytest flake8 mlflow

      - name: Lint
        run: flake8 notebooks/ src/ --max-line-length=120

      - name: Unit tests
        run: pytest tests/unit/ -v

      - name: Validate DAB bundle
        run: databricks bundle validate --target staging

  train-and-deploy:
    name: Train, Validate, Deploy
    runs-on: ubuntu-latest
    needs: ci
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4

      - name: Install Databricks CLI
        run: pip install databricks-cli databricks-sdk

      - name: Deploy bundle to staging
        run: databricks bundle deploy --target staging

      - name: Run training job
        run: |
          JOB_RUN_ID=$(databricks bundle run training_job \
            --var "git_sha=${{ github.sha }}" \
            --output json | jq -r '.run_id')
          echo "TRAINING_RUN_ID=$JOB_RUN_ID" >> $GITHUB_ENV

      - name: Run validation job
        run: |
          databricks bundle run validation_job --target staging
          echo "Validation passed. Promoting to production."

      - name: Deploy bundle to production
        run: databricks bundle deploy --target production

      - name: Update Model Serving endpoint
        run: |
          python scripts/update_serving_endpoint.py \
            --model-name churn-prediction-gbm \
            --alias Production

---

步骤 5 — 更新服务端点

流水线的最后一步是更新 Databricks Model Serving 端点，使其指向新提升的 Production 模型版本。

# scripts/update_serving_endpoint.py
import argparse
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ServedModelInput, EndpointCoreConfigInput
from mlflow import MlflowClient

def update_serving_endpoint(model_name: str, alias: str):
    w      = WorkspaceClient()
    client = MlflowClient()

    # Resolve alias to concrete version
    model_version = client.get_model_version_by_alias(model_name, alias).version
    print(f"Deploying {model_name} version {model_version} (alias: {alias})")

    endpoint_name = 'churn-prediction-endpoint'
    served_model  = ServedModelInput(
        model_name=model_name,
        model_version=model_version,
        workload_size='Small',
        scale_to_zero_enabled=True,
    )

    try:
        # Update existing endpoint
        w.serving_endpoints.update_config(
            name=endpoint_name,
            served_models=[served_model],
        )
        print(f"Endpoint '{endpoint_name}' updated to version {model_version}.")
    except Exception:
        # Create if it doesn't exist yet
        w.serving_endpoints.create(
            name=endpoint_name,
            config=EndpointCoreConfigInput(served_models=[served_model]),
        )
        print(f"Endpoint '{endpoint_name}' created with version {model_version}.")

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--model-name', required=True)
    parser.add_argument('--alias',      required=True)
    args = parser.parse_args()
    update_serving_endpoint(args.model_name, args.alias)

---

工具对比

| 工具 | 在流水线中的角色 | 为什么不用替代方案 | | --- | --- | --- | | Databricks Asset Bundles | 任务和集群的 IaC | Terraform Databricks provider（更冗长，缺乏原生 Notebook 支持），手动 UI（不可复现） | | MLflow Registry aliases | Production / Staging 提升 | 基于 Stage 的提升（在 MLflow 2.x 中已弃用），手动版本跟踪（容易出错） | | GitHub Actions | CI/CD 编排器 | Azure DevOps（同样能运行，只需交换 yml 语法），Jenkins（运维开销更大） | | 验证任务中的指标关卡 | 自动化质量检查 | 人工审查（拖慢速度），没有关卡（有风险） | | Databricks Model Serving | 托管 REST 端点 | AKS 部署（更多控制，但运维量大得多），Azure ML 端点（增加额外服务依赖） | | dbutils.jobs.taskValues | 在任务间传递 run_id | 环境变量（在 DABs 中跨任务不可用），硬编码的 run 查找（脆弱） |

---

生产中需要注意的事项

固定你的 Databricks Runtime 版本。在 bundle 中使用 14.3.x-scala2.12 可以确保每次训练运行使用相同的 Spark 和库版本。浮动版本（latest）会导致静默的库漂移，破坏可复现性。

将密钥存储在 Azure Key Vault 中，而不仅仅是在 GitHub Secrets 中。GitHub Secrets 对 CI 令牌来说没问题，但对于 Databricks 任务在运行时使用的长期服务主体凭据，应使用 Azure Key Vault 作为后端，并通过 Databricks 密钥作用域进行引用。

从当前生产模型设置指标基准。你的阈值（ROC-AUC >= 0.80）应相对于当前 Production 模型在相同保留集上达到的指标，而不是一个任意数字。在验证任务中添加一个步骤，获取当前 Production 模型的指标，并将新模型与之进行比较。

用 git SHA 标记每个 MLflow 运行。在每次训练运行中将 git_sha 作为参数记录，意味着你可以始终将模型工件追溯到产生它的确切代码版本。这对事件响应至关重要。

服务端点启用缩放到零。对于非延迟敏感的模型，在服务端点上启用 scale_to_zero_enabled=True。对于不接收 24/7 流量的端点，这可以大幅降低成本。

---

总结

这里的模式很直接：代码变更触发 CI，CI 触发训练任务，训练任务注册模型，验证任务根据指标设置关卡，只有在那之后模型才被提升和部署。没有手动操作，没有跳过任何步骤。

使这个模式达到生产级别而不仅仅是自动化的原因，是结合了 Delta 版本控制以实现特征的可复现性、MLflow aliases 以实现清晰的提升语义，以及基于指标的关卡，从而确保一个更差的模型永远无法静默地替换一个更好的模型。

---

参考资料

• Databricks Asset Bundles 文档
• MLflow 模型注册表 — 别名和标签
• Databricks Model Serving
• Databricks SDK for Python
• GitHub Actions — 适用于 Databricks 的 CI/CD
• MLflow 自动日志记录
• Azure Key Vault — Databricks 密钥作用域
• Databricks Lakehouse Monitoring