MLOps极致细节：17. Azure ML Pipeline（机器学习管道），模型训练，打包和注册

这两个章节中，我们将介绍Azure ML Pipeline的使用，并且结合MLFlow一起跟踪ML模型。此章节将通过一个案例详细介绍如何训练，测试，打包和注册模型。

正如此系列博客之前所描述，MLFlow是一个很好的MLOps管理的开源软件。这里我们可以使用MLFlow的Tracking模块记录和跟踪训练运行指标和模型项目，而不管试验环境是位于本地计算机、远程计算目标、虚拟机还是 Azure Databricks 群集上，并最终将其存储在 Azure 机器学习工作区中。如下图：

Win10
IDE：VSCode
Anaconda
代码地址

文章目录

MLOps极致细节：17. Azure ML Pipeline（机器学习管道），模型训练，打包和注册
- 1 初始化
- 2 模型训练，测试，打包和注册模型主代码
- 3 模型训练
- 4 模型测试/验证
- 5 模型打包转化为onnx格式
- 6 模型注册
- 7 结果

在上一章节中，我们已经对Azure 机器学习管道进行了简单的介绍，并且介绍了如何克隆代码并在本地，或者Azure Copute Instance VM中进行运行。

1 初始化

首先我们需要从Azure portal中获得subscription_id，resource_group，workspace_name这三个参数。我们进入Azure portal，Machine Vision 账号后就能看到，如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wwl9Ez7C-1647703354152)(./img/25.png)]

我们实例化azureml.core.Workspace，然后获得uri。我们可以使用mlflow.set_tracking_uri(uri)来将MLFlow之后所有Tracking的参数，模型都保存到这个uri中（默认是保存到本地）。

# 实例化workspace
workspace = Workspace(subscription_id, resource_group, workspace_name)
# Get the uri of the workspace, we will also use mlflow to record parameters and models trained.
uri = workspace.get_mlflow_tracking_uri()
mlflow.set_tracking_uri(uri)

接下来，我们通过Dataset.get_by_name导入在之前章节中介绍过的，存于Azure Storage中的数据（trainDataset），并且通过.to_pandas_dataframe()转换为pandas的df格式。然后将可以被预测使用的特征列拿出来，作为X值，而预测列futureWeather作为Y值。通过train_test_split，我们将此数据集拆分为训练数据集以及验证数据集。最后，最此数据集进行归一化。代码如下：

# 导入我们之前存于storage的数据
datasetTrn = Dataset.get_by_name(workspace, name='trainDataset')
print("Input dataset name: {0}; Version: {1}.".format(datasetTrn.name, datasetTrn.version))
# 转化为pandas df格式
dfTrn = datasetTrn.to_pandas_dataframe()
print("Shape of train dataset: {0}".format(dfTrn.shape))
# Features
X = dfTrn[['Temperature_C', 'Humidity', 'Wind_speed_kmph', 'Wind_bearing_degrees', 'Visibility_km', 'Pressure_millibars', 'currentWeather']].values
# Label (Values to predict)
y = dfTrn['futureWeather'].values
# 将数据差分成训练集和验证集
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=1)
# 去均值和方差的归一化
sc = StandardScaler()
X_train = sc.fit_transform(X_train)
X_val = sc.transform(X_val)

这个案例中，我们支持使用两种算法进行预测：支持向量机（SVM）以及随机森林（RF）。这里我们将不对两个算法进行详细解释，而是更加专注于如何使用Azure Machine Learning来进行模型训练，测试，打包和注册模型。

2 模型训练，测试，打包和注册模型主代码

对于SVM和RF，模型训练，测试，打包和注册模型的逻辑都是一样的：

新建一个Experiment。Experiment就类似于一个任务。在一个Experiment里面可以包含很多个run，比如说，你可以重复运行main.py代码，在Experiment Name不变的情况下，每运行一次，都会新建一个新的run，之后我们会详细介绍；
新建一个run。
实例化svmImp类或者rfImp类。我们将两个算法相关的代码放在各自的类中。
模型训练：svmImpTrn或者rfImpTrn。
模型测试/验证：svmImpVal或者rfImpVal。
模型打包转化为onnx格式：onnxModelSave。
模型注册：modelRegister以及modelRegisterSC。

代码如下：

print("Start modelling with {0} method".format(model2Test))
if model2Test == "svm":# 新建一个新的experiment，这个和mlflow是一个道理myexperiment = Experiment(workspace, "SVMTest1")mlflow.set_experiment("SVMTest1")# 在此experiment下新建一个runwith mlflow.start_run() as new_run:mlflow.log_param("dataset_name", datasetTrn.name)mlflow.log_param("dataset_version", datasetTrn.version)# 实例化svmImp类svmImp_ = svmImp(datasetTrn, X_train, y_train, X_val, y_val,new_run)svmImp_.svmImpTrn()     svmImp_.svmImpVal()svmImp_.onnxModelSave()svmImp_.modelRegister(workspace)svmImp_.modelRegisterSC(sc,workspace)print ("run id:", new_run.info.run_id)mlflow.end_run()
elif model2Test == "rf":# 新建一个新的experiment，这个和mlflow是一个道理myexperiment = Experiment(workspace, "RFTest1")mlflow.set_experiment("RFTest1")# 在此experiment下新建一个runwith mlflow.start_run() as new_run:mlflow.log_param("dataset_name", datasetTrn.name)mlflow.log_param("dataset_version", datasetTrn.version)# 实例化rfImp类rfImp_ = rfImp(datasetTrn, X_train, y_train, X_val, y_val,new_run)rfImp_.rfImpTrn()rfImp_.rfImpVal()rfImp_.onnxModelSave()rfImp_.modelRegister(workspace)rfImp_.modelRegisterSC(sc,workspace)print ("run id:", new_run.info.run_id)mlflow.end_run()

3 模型训练

支持向量机（Support Vector Machine）进行模型训练代码如下：

def svmImpTrn(self):'''Model training with SVM method. First we use grid search to find out the best svm hyper-parameters.Then we implement them for prediction.'''print("We use Grid Search to find out the best hyper-parameters for svm training.")svc_grid = GridSearchCV(self.svc, self.parameters)svc_grid.fit(self.X_train, self.y_train)print("Get parameters of svc method: ",svc_grid.get_params(deep=True))print("Now we implement the best hyperparameter to svm training.")self.svc = SVC(C=svc_grid.get_params(deep=True)['estimator__C'], kernel=svc_grid.get_params(deep=True)['estimator__kernel'])self.svc.fit(self.X_train, self.y_train)mlflow.log_param("C", svc_grid.get_params(deep=True)['estimator__C'])mlflow.log_param("Kernel", svc_grid.get_params(deep=True)['estimator__kernel'])

随机森林（Random Forest）进行模型训练代码如下：

def rfImpTrn(self):'''Modelling with random forest method'''self.rf.fit(self.X_train, self.y_train)mlflow.log_param("max_depth", 10)mlflow.log_param("random_state", 0)mlflow.log_param("n_estimators", 100)

这里需要注意，在我们使用svm的时候，我们先使用了GridSearchCV方法搜索最优超参数，然后再用这个超参数进行模型的训练，所以整体上所需要花的时间比较长，大概15分钟左右。另外，我们通过mlflow.log_param来记录我们希望保存的一些参数值。关于MLFlow Tracking的使用，请参见此系列博客之前的文章。这里由于我们已经在前面设置了mlflow.set_tracking_uri(uri)，所以所有这些日志数据都会被保存在Azure Machine Learning中，我们之后的章节会解释。

4 模型测试/验证

支持向量机（Support Vector Machine）进行模型测试/验证代码如下：

def svmImpVal(self):'''Model testing with SVM method.We finally get accuracy, fscore, precision and recall values.'''predicted_svc = self.svc.predict(self.X_val)self.acc = accuracy_score(self.y_val, predicted_svc)self.fscore = f1_score(self.y_val, predicted_svc, average="macro")self.precision = precision_score(self.y_val, predicted_svc, average="macro")self.recall = recall_score(self.y_val, predicted_svc, average="macro")print("Validation result: Accuracy: {0}; Fscore: {1}; Precision: {2}; Recall: {3}"\.format(self.acc, self.fscore, self.precision, self.recall))repo = git.Repo(search_parent_directories=True)self.sha = repo.head.object.hexsha# Log to AzureML and MLflowmlflow.log_param("Test_accuracy", self.acc)mlflow.log_param("Precision", self.precision)mlflow.log_param("Test_accuracy", self.recall)mlflow.log_param("F-Score", self.fscore)mlflow.log_param("Git-sha", self.sha)mlflow.sklearn.log_model(self.svc, 'outputs')

随机森林（Random Forest）进行模型测试/验证代码如下：

def rfImpVal(self):'''Model testing with random forest method.We finally get accuracy, fscore, precision and recall values.'''predicted_rf = self.rf.predict(self.X_val)self.acc = accuracy_score(self.y_val, predicted_rf)self.fscore = f1_score(self.y_val, predicted_rf, average="macro")self.precision = precision_score(self.y_val, predicted_rf, average="macro")self.recall = recall_score(self.y_val, predicted_rf, average="macro")repo = git.Repo(search_parent_directories=True)self.sha = repo.head.object.hexsha# Log to AzureML and MLflowmlflow.log_param("Test_accuracy", self.acc)mlflow.log_param("Precision", self.precision)mlflow.log_param("Test_recall", self.recall)mlflow.log_param("F-Score", self.fscore)mlflow.log_param("Git-sha", self.sha)mlflow.sklearn.log_model(self.rf, 'outputs')

5 模型打包转化为onnx格式

在上一步中对已训练模型进行测试/验证后，我们可以将模型序列化为文件以导出到测试或生产环境中。序列化文件会带来兼容性挑战，比如，在使用不同框架进行模型训练（模型1使用TF进行模型训练，而模型2使用sklearn，他们的模型自然不能相互使用）。一种解决办法就是同一转换成ONNX格式。这里，我们通过convert_sklearn将模型序列化为XXX.onnx，以便进一步将模型导出和导入到测试和生产环境中。

支持向量机（Support Vector Machine）代码如下：

def onnxModelSave(self):'''Convert svm model into ONNX format file and save to local path.'''initial_type = [('float_input', FloatTensorType([None, 6]))]onx = convert_sklearn(self.svc, initial_types=initial_type)with open("outputs/svc.onnx", "wb") as f:f.write(onx.SerializeToString())

随机森林（Random Forest）代码如下：

def onnxModelSave(self):'''Convert random forest model into ONNX format file and save to local path.'''initial_type = [('float_input', FloatTensorType([None, 6]))]onx = convert_sklearn(self.rf, initial_types=initial_type)with open("outputs/rf.onnx", "wb") as f:f.write(onx.SerializeToString())

6 模型注册

在这一步中，我们将在上一步中进行了序列化或容器化的模型注册并存储在Azure的模型注册表中。

支持向量机（Support Vector Machine）代码如下：

def modelRegister(self,workspace):'''Register Model on AzureML workstation'''model = Model.register(model_path = './outputs/svc.onnx', # this points to a local file model_name = "support-vector-classifier", # this is the name the model is registered astags = {'dataset': self.dataset.name, 'version': self.dataset.version, 'hyparameter-C': '1', 'testdata-accuracy': '0.9519'}, model_framework='pandas==0.23.4',description = "Support vector classifier to predict weather at port of Turku",workspace = workspace)print('Name:', model.name)print('Version:', model.version)# Save the model to the outputs directory for capturemlflow.sklearn.log_model(self.svc, 'outputs/svc.onnx')

随机森林（Random Forest）代码如下：

def modelRegister(self,workspace):'''Register Model on AzureML workstation'''# Register Model on AzureML WSmodel = Model.register(model_path = './outputs/rf.onnx', # this points to a local file model_name = "random-forest-classifier", # this is the name the model is registered astags = {'dataset': self.dataset.name, 'version': self.dataset.version, 'hyparameter-C': '1', 'testdata-accuracy': '0.9548'}, model_framework='pandas==0.23.4',description = "Random forest classifier to predict weather at port of Turku",workspace = workspace)print('Name:', model.name)print('Version:', model.version)# Save the model to the outputs directory for capturemlflow.sklearn.log_model(self.rf, 'outputs/rf.onnx')

并且我们也将归一化的参数注册起来。

def modelRegisterSC(self, sc, workspace):'''Register sc (StandardScaler) for the data standardrization parameters.'''with open('./outputs/scaler.pkl', 'wb') as scaler_pkl:# sc = StandardScaler()pickle.dump(sc, scaler_pkl)# Register Model on AzureML WSscaler = Model.register(model_path = './outputs/scaler.pkl', # this points to a local file model_name = "scaler", # this is the name the model is registered astags = {'dataset': self.dataset.name, 'version': self.dataset.version}, model_framework='pandas==0.23.4',description = "Scaler used for scaling incoming inference data",workspace = workspace)print('Name:', scaler.name)print('Version:', scaler.version)

7 结果

此处以随机森林距离，当我们成功运行main.py后，我们回到Azure Portal的Azure Machine Learning账号，然后进入Machine Learning Studio，我们先看一下Datasets，我们能看到之前博客中注册的三个数据集：testDataset，trainDataset，trainDataset。这里我们只用了trainDataset这个。

我们看一下Experiments，如下图：

这个RFTest1就是我们自己在代码中输入的Experiment Name（myexperiment = Experiment(workspace, "RFTest1")）。

我们点进去看一下，里面罗列了许多run的记录

我们点进去刚才我们跑代码的那个run

我们发现，所有刚才代码中的mlflow.log_param所记录下来的值都展示在了这里。当然，我们还有很多功能没有用，比如我们可以用metrics来记录某个参数的连续变化（就是MLFlow的metrics，一个功能），Images来保存镜像信息，Child runs的意思是，我们每一个run可以设置nest，可以有很多child run，比如，我们训练一个深度学习模型通常需要很多echo，那么每一个echo都可以是一个child run，我们可以在每个child run中保存每个loop的模型以及相关参数。我们如果点击Outputs+Logs，我们能看见mlflow.sklearn.log_model的结果。

最后，我们在左边栏点击Models，我们便能看到我们刚才注册的模型，以及对应版本。