Microsoft MLOps项目实战：使用Azure ML管道自动化机器学习工作流

引言

在现代机器学习项目中，自动化是提高效率和可靠性的关键。本文将深入探讨如何使用Azure Machine Learning服务中的管道(Pipeline)功能来自动化机器学习工作流。我们将基于Microsoft MLOps项目中的实际案例，展示如何构建端到端的自动化机器学习流程。

准备工作

在开始构建自动化管道前，我们需要确保环境已正确配置：

1. Azure ML工作区：已创建并配置好工作区
2. 计算资源：准备好用于执行管道步骤的计算资源
3. 数据集：已将糖尿病数据集上传到工作区的默认数据存储中

!pip install --upgrade azureml-sdk[notebooks]

from azureml.core import Workspace
ws = Workspace.from_config()
print('工作区准备就绪:', ws.name)

理解ML管道的基本概念

机器学习管道由多个**步骤(Step)**组成，每个步骤可以：

• 运行独立的Python脚本
• 使用不同的计算资源
• 接收输入并产生输出
• 在特定条件下执行

管道的主要优势包括：

• 可重复性：确保每次运行都遵循相同流程
• 模块化：便于单独开发和测试各个组件
• 自动化：减少人工干预，提高效率

构建管道步骤

步骤1：创建训练脚本

我们首先创建一个训练脚本train_diabetes.py，该脚本将：

1. 从工作区加载数据集
2. 训练逻辑回归模型
3. 保存训练好的模型

%%writefile $experiment_folder/train_diabetes.py
# 导入所需库
import argparse
import joblib
from azureml.core import Run
import pandas as pd
from sklearn.linear_model import LogisticRegression

# 解析参数
parser = argparse.ArgumentParser()
parser.add_argument('--dataset_name', type=str, default="Diabetes Dataset")
parser.add_argument('--regularization', type=float, default=0.01)
parser.add_argument('--output_folder', type=str, default="diabetes_model")
args = parser.parse_args()

# 获取运行上下文
run = Run.get_context()

# 加载数据并训练模型
diabetes = Dataset.get_by_name(workspace=run.experiment.workspace, 
                             name=args.dataset_name).to_pandas_dataframe()
X = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure',
             'TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].values
y = diabetes['Diabetic'].values

model = LogisticRegression(C=1/args.reg_rate, solver="liblinear").fit(X, y)

# 保存模型
joblib.dump(value=model, filename=f"{args.output_folder}/model.pkl")
run.complete()

步骤2：创建模型注册脚本

接下来创建register_diabetes.py脚本，用于将训练好的模型注册到工作区：

%%writefile $experiment_folder/register_diabetes.py
import argparse
import joblib
from azureml.core import Model, Run

parser = argparse.ArgumentParser()
parser.add_argument('--model_folder', type=str, default="diabetes_model")
args = parser.parse_args()

run = Run.get_context()
model = joblib.load(f"{args.model_folder}/model.pkl")

Model.register(workspace=run.experiment.workspace,
             model_path=f"{args.model_folder}/model.pkl",
             model_name='diabetes_model',
             tags={'Training context':'Pipeline'})
run.complete()

配置计算环境

管道步骤可以在不同的计算资源上运行。我们需要配置计算目标和运行环境：

from azureml.core.compute import ComputeTarget, AmlCompute

# 创建或获取计算集群
cpu_cluster_name = "cpu-cluster"
try:
    cpu_cluster = ComputeTarget(workspace=ws, name=cpu_cluster_name)
except ComputeTargetException:
    compute_config = AmlCompute.provisioning_configuration(
        vm_size='STANDARD_D2_V2', 
        max_nodes=4)
    cpu_cluster = ComputeTarget.create(ws, cpu_cluster_name, compute_config)
    cpu_cluster.wait_for_completion(show_output=True)

# 配置运行环境
from azureml.core.runconfig import RunConfiguration
from azureml.core.conda_dependencies import CondaDependencies

pipeline_run_config = RunConfiguration()
pipeline_run_config.target = cpu_cluster
pipeline_run_config.environment.docker.enabled = True
pipeline_run_config.environment.python.conda_dependencies = \
    CondaDependencies.create(conda_packages=['pandas','scikit-learn'], 
                            pip_packages=['azureml-sdk'])

构建和运行管道

定义管道步骤

使用PythonScriptStep定义管道中的各个步骤，并通过PipelineData在步骤间传递数据：

from azureml.pipeline.core import PipelineData
from azureml.pipeline.steps import PythonScriptStep

# 创建数据引用
model_folder = PipelineData("model_folder", datastore=ws.get_default_datastore())

# 定义训练步骤
train_step = PythonScriptStep(
    name="Train Model",
    source_directory=experiment_folder,
    script_name="train_diabetes.py",
    arguments=['--dataset_name', 'Diabetes Dataset',
              '--regularization', 0.1,
              '--output_folder', model_folder],
    outputs=[model_folder],
    compute_target=cpu_cluster,
    runconfig=pipeline_run_config)

# 定义注册步骤
register_step = PythonScriptStep(
    name="Register Model",
    source_directory=experiment_folder,
    script_name="register_diabetes.py",
    arguments=['--model_folder', model_folder],
    inputs=[model_folder],
    compute_target=cpu_cluster,
    runconfig=pipeline_run_config)

创建并运行管道

将步骤组合成管道并提交执行：

from azureml.pipeline.core import Pipeline
from azureml.core import Experiment

# 构建管道
pipeline = Pipeline(workspace=ws, steps=[train_step, register_step])

# 提交执行
experiment = Experiment(workspace=ws, name='diabetes_pipeline')
pipeline_run = experiment.submit(pipeline)
pipeline_run.wait_for_completion()

验证结果

管道运行完成后，我们可以检查注册的模型：

from azureml.core import Model

for model in Model.list(ws):
    print(f"模型名称: {model.name}, 版本: {model.version}")
    for tag, value in model.tags.items():
        print(f"\t标签: {tag}: {value}")

进阶应用场景

在实际生产环境中，我们可以扩展这个基础管道：

1. 模型评估：添加评估步骤，比较新模型与现有模型的性能
2. 条件执行：根据评估结果决定是否注册新模型
3. CI/CD集成：与Azure DevOps等工具集成，实现持续集成和部署
4. 触发机制：设置数据变更或定期触发管道执行

总结

通过Azure ML管道，我们实现了机器学习工作流的自动化，包括：

• 数据准备和模型训练
• 模型评估和注册
• 工作流编排和自动化执行

这种自动化方法不仅提高了效率，还确保了流程的可重复性和可靠性，是MLOps实践中的重要组成部分。