В этом руководстве мы создадим полный рабочий процесс экспериментирования и развёртывания машинного обучения (ML) с использованием MLflow. Мы начнём с запуска выделенного сервера отслеживания MLflow со структурированным бэкендом и хранилищем артефактов, что позволит нам отслеживать эксперименты масштабируемым и воспроизводимым образом.
Установка зависимостей и импорт библиотек
Мы устанавливаем все необходимые зависимости и импортируем библиотеки MLflow, scikit-learn и системные библиотеки, необходимые для отслеживания экспериментов и развёртывания. Мы определяем служебные функции, которые позволяют нам проверять доступность портов, ждать готовности сервера и безопасно завершать фоновые процессы.
«`python
import os
import time
import json
import shutil
import socket
import signal
import subprocess
from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import requests
from sklearn.datasets import loadbreastcancer
from sklearn.modelselection import traintest_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
rocaucscore,
accuracy_score,
precision_score,
recall_score,
f1_score,
confusion_matrix,
ConfusionMatrixDisplay,
)
import mlflow
import mlflow.sklearn
from mlflow.models.signature import infer_signature
«`
Конфигурация сервера отслеживания MLflow
Мы настраиваем хранилище данных MLflow и каталоги артефактов, чтобы создать структурированную, постоянную среду отслеживания экспериментов. Мы запускаем сервер отслеживания MLflow с базой данных SQLite и локальным хранилищем артефактов, что позволяет полностью отслеживать эксперименты и управлять ими.
«`python
BASEDIR = Path(«/content/mlflowcolab_demo»).resolve()
BACKENDDB = BASEDIR / «mlflow.db»
ARTIFACTROOT = BASEDIR / «mlartifacts»
os.makedirs(BASEDIR, existok=True)
os.makedirs(ARTIFACTROOT, existok=True)
HOST = «127.0.0.1»
PORT = 5000
TRACKING_URI = f»http://{HOST}:{PORT}»
if isport_open(HOST, PORT):
for p in range(5001, 5015):
if not isport_open(HOST, p):
PORT = p
TRACKING_URI = f»http://{HOST}:{PORT}»
break
print(«Using TRACKINGURI:», TRACKINGURI)
print(«Backend DB:», BACKEND_DB)
print(«Artifact root:», ARTIFACT_ROOT)
server_cmd = [
«mlflow»,
«server»,
«—host», HOST,
«—port», str(PORT),
«—backend-store-uri», f»sqlite:///{BACKEND_DB}»,
«—default-artifact-root», str(ARTIFACT_ROOT),
]
mlflow_server = subprocess.Popen(
server_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
waitforhttp(TRACKINGURI, timeout_s=45)
mlflow.settrackinguri(TRACKING_URI)
print(«MLflow server is up.»)
EXPERIMENT_NAME = «colab-advanced-mlflow-sklearn»
mlflow.setexperiment(EXPERIMENTNAME)
«`
Загрузка данных и подготовка к экспериментированию
Мы загружаем набор данных и готовим необходимые для машинного обучения эксперименты тренировочные и тестовые сплиты. Мы включаем автоматическое отслеживание MLflow, что позволяет автоматически отслеживать параметры, метрики и артефакты моделей без ручного вмешательства.
«`python
data = loadbreastcancer(as_frame=True)
df = data.frame.copy()
target_col = «target»
X = df.drop(columns=[target_col])
y = df[target_col].astype(int)
mlflow.sklearn.autolog(
loginputexamples=False,
logmodelsignatures=False,
silent=True
)
C_VALUES = [0.01, 0.1, 1.0, 3.0]
SOLVERS = [«liblinear», «lbfgs»]
best = {«auc»: -1.0, «run_id»: None, «params»: None}
«`
Гиперпараметрическая оптимизация
Мы выполняем вложенный гиперпараметрический поиск, обучая несколько моделей в структурированной иерархии родительских и дочерних запусков. Мы вычисляем метрики производительности и регистрируем их вместе с диагностическими артефактами, такими как матрицы путаницы, чтобы обеспечить детальный анализ экспериментов.
«`python
with mlflow.startrun(runname=»parentsweeprun») as parent_run:
mlflow.logparam(«dataset», «sklearnbreast_cancer»)
mlflow.logparam(«nfeatures», X_train.shape[1])
mlflow.logparam(«ntrain», X_train.shape[0])
mlflow.logparam(«ntest», X_test.shape[0])
for C in C_VALUES:
for solver in SOLVERS:
with mlflow.startrun(runname=f»childC={C}solver={solver}», nested=True) as child_run:
pipe = Pipeline([
(«scaler», StandardScaler()),
(«clf», LogisticRegression(
C=C,
solver=solver,
penalty=»l2″,
max_iter=2000,
random_state=42
))
])
pipe.fit(Xtrain, ytrain)
proba = pipe.predictproba(Xtest)[:, 1]
pred = (proba >= 0.5).astype(int)
auc = rocaucscore(y_test, proba)
acc = accuracyscore(ytest, pred)
prec = precisionscore(ytest, pred, zero_division=0)
rec = recallscore(ytest, pred, zero_division=0)
f1 = f1score(ytest, pred, zero_division=0)
mlflow.log_metrics({
«test_auc»: float(auc),
«test_accuracy»: float(acc),
«test_precision»: float(prec),
«test_recall»: float(rec),
«test_f1»: float(f1),
})
cm = confusionmatrix(ytest, pred)
disp = ConfusionMatrixDisplay(cm, displaylabels=data.targetnames)
fig, ax = plt.subplots(figsize=(5, 4))
disp.plot(ax=ax, values_format=»d»)
ax.set_title(f»Confusion Matrix (C={C}, solver={solver})»)
cmpath = BASEDIR / «confusion_matrix.png»
fig.tight_layout()
fig.savefig(cm_path, dpi=140)
plt.close(fig)
mlflow.logartifact(str(cmpath), artifact_path=»diagnostics»)
if auc > best[«auc»]:
best.update({
«auc»: float(auc),
«runid»: childrun.info.run_id,
«params»: {«C»: C, «solver»: solver}
})
mlflow.logdict(best, «bestrun_summary.json»)
print(«Best config:», best)
«`
Развёртывание модели
Мы развёртываем обученную модель MLflow в виде живого REST API-сервиса с использованием встроенной инфраструктуры обслуживания MLflow. Мы отправляем тестовый запрос на конечную точку развёрнутой модели, чтобы убедиться, что модель отвечает правильно и производит прогнозы в реальном времени.
«`python
best_C = best[«params»][«C»]
best_solver = best[«params»][«solver»]
final_pipe = Pipeline([
(«scaler», StandardScaler()),
(«clf», LogisticRegression(
C=best_C,
solver=best_solver,
penalty=»l2″,
max_iter=2000,
random_state=42
))
])
with mlflow.startrun(runname=»finalmodelrun») as final_run:
finalpipe.fit(Xtrain, y_train)
proba = finalpipe.predictproba(X_test)[:, 1]
pred = (proba >= 0.5).astype(int)
metrics = {
«testauc»: float(rocaucscore(ytest, proba)),
«testaccuracy»: float(accuracyscore(y_test, pred)),
«testprecision»: float(precisionscore(ytest, pred, zerodivision=0)),
«testrecall»: float(recallscore(ytest, pred, zerodivision=0)),
«testf1″: float(f1score(ytest, pred, zerodivision=0)),
}
mlflow.log_metrics(metrics)
mlflow.logparams({«C»: bestC, «solver»: best_solver, «model»: «LogisticRegression+StandardScaler»})
inputexample = Xtest.iloc[:5].copy()
signature = infersignature(inputexample, finalpipe.predictproba(input_example)[:, 1])
modelinfo = mlflow.sklearn.logmodel(
skmodel=finalpipe,
artifact_path=»model»,
signature=signature,
inputexample=inputexample,
registeredmodelname=None,
)
print(«Final runid:», finalrun.info.run_id)
print(«Logged model URI:», modelinfo.modeluri)
evaldf = Xtest.copy()
evaldf[«label»] = ytest.values
eval_result = mlflow.models.evaluate(
model=modelinfo.modeluri,
data=eval_df,
targets=»label»,
model_type=»classifier»,
evaluators=»default»,
)
eval_summary = {
«metrics»: {k: float(v) if isinstance(v, (int, float, np.floating)) else str(v)
for k, v in eval_result.metrics.items()},
«artifacts»: {k: str(v) for k, v in eval_result.artifacts.items()},
}
mlflow.logdict(evalsummary, «evaluation/eval_summary.json»)
«`
Завершение жизненного цикла машинного обучения
Мы завершили полный жизненный цикл машинного обучения, перейдя от отслеживания экспериментов к развёртыванию живых моделей в рамках унифицированного рабочего процесса MLflow. Мы создали структурированную среду, в которой каждый тренировочный запуск отслеживается, воспроизводится и поддаётся аудиту, что обеспечивает эффективное экспериментирование и сравнение моделей.
В этом руководстве мы продемонстрировали, как MLflow функционирует в качестве центрального слоя оркестрации для управления системами машинного обучения, обеспечивая масштабируемые, воспроизводимые и готовые к производству ML-конвейеры в облачной среде на основе ноутбуков.
1. Какие основные шаги включает в себя процесс экспериментирования и развёртывания машинного обучения с использованием MLflow?
Ответ:
Процесс включает в себя:
* установку зависимостей и импорт библиотек;
* конфигурацию сервера отслеживания MLflow;
* загрузку данных и подготовку к экспериментированию;
* гиперпараметрическую оптимизацию;
* развёртывание модели в виде живого REST API-сервиса с использованием встроенной инфраструктуры обслуживания MLflow.
2. Какие библиотеки и инструменты используются в данном руководстве для работы с MLflow?
Ответ:
В руководстве используются:
* MLflow;
* scikit-learn (sklearn);
* NumPy (numpy);
* Pandas (pd);
* Matplotlib (plt);
* другие системные библиотеки Python.
3. Какие параметры и метрики отслеживаются в процессе гиперпараметрической оптимизации?
Ответ:
В процессе гиперпараметрической оптимизации отслеживаются:
* параметры модели (C и solver);
* метрики производительности (AUC, точность, точность, полнота, F1-score);
* диагностические артефакты (матрицы путаницы).
4. Как обеспечивается воспроизводимость экспериментов в данном руководстве?
Ответ:
Воспроизводимость экспериментов обеспечивается за счёт:
* структурированной среды отслеживания экспериментов;
* автоматического отслеживания параметров, метрик и артефактов моделей;
* использования унифицированного рабочего процесса MLflow.
5. Какие шаги необходимо выполнить для развёртывания обученной модели в виде живого REST API-сервиса?
Ответ:
Для развёртывания обученной модели необходимо:
* обучить модель с использованием оптимальных гиперпараметров;
* отправить модель на сервер MLflow;
* убедиться, что модель отвечает правильно и производит прогнозы в реальном времени.