Tworzenie potoku¶
Poprzednie ćwiczenia prowadziły przez całą drogę od dostępu do danych, przez trenowanie modelu, aż po jego rejestrację - za każdym razem kolejne kroki uruchamiane były ręcznie z notatnika. Teraz zobaczysz, jak te kroki zautomatyzować, budując z nich potok (ang. pipeline).
Połączenie z obszarem roboczym¶
Na początek połącz się z obszarem roboczym przy użyciu Azure ML SDK v2.
Uwaga: Jeśli od poprzedniego ćwiczenia wygasła uwierzytelniona sesja z subskrypcją Azure, pojawi się prośba o ponowne zalogowanie.
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
ml_client = MLClient.from_config(credential=credential)
print(f"Gotowe do pracy z obszarem roboczym {ml_client.workspace_name}")
Przygotowanie danych dla potoku¶
W tym ćwiczeniu pracujesz na danych o cukrzycy. Krok trenujący w potoku czyta je jako pojedynczy plik CSV (uri_file), więc zamiast opierać się na współdzielonym zasobie diabetes_mltable (zarejestrowanym jako mltable, w konkretnej wersji) przekażesz jako wejście potoku lokalny plik data/diabetes.csv. SDK wyśle go do chmury automatycznie przy zlecaniu zadania potoku.
Skrypty dla kroków potoku¶
Potok w Azure ML SDK v2 składa się z komponentów (ang. component). Komponent to skrypt (albo inny program) opisany razem ze swoimi wejściami, wyjściami i środowiskiem; każdy może wykonywać się na innym środowisku obliczeniowym. Potok powstaje przez połączenie komponentów: wyjście jednego staje się wejściem następnego.
Zbudujesz tu prosty potok z dwóch komponentów - pierwszy wytrenuje model, drugi zarejestruje go w obszarze roboczym.
import os
# Utwórz folder na pliki kroków potoku
experiment_folder = 'diabetes_pipeline'
os.makedirs(experiment_folder, exist_ok=True)
print(experiment_folder)
Teraz napisz skrypt pierwszego kroku, czyli trenowania modelu. Skrypt przyjmuje dwa parametry: training_data (dane wejściowe) i output_folder (folder, w którym ma zapisać wytrenowany model). Azure ML konfiguruje śledzenie MLflow w każdym zadaniu automatycznie, więc skrypt może po prostu użyć mlflow do zapisania metryk i modelu.
%%writefile $experiment_folder/train_diabetes.py
# Import bibliotek
import argparse
import pandas as pd
import numpy as np
import mlflow
import mlflow.sklearn
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_auc_score
from sklearn.metrics import roc_curve
import matplotlib.pyplot as plt
# Odczytaj parametry
parser = argparse.ArgumentParser()
parser.add_argument('--training_data', type=str, dest='training_data', help='dane treningowe')
parser.add_argument('--output_folder', type=str, dest='output_folder', default="diabetes_model", help='folder wyjściowy')
args = parser.parse_args()
output_folder = args.output_folder
# wczytaj dane o cukrzycy (przekazane jako wejście komponentu)
print("Wczytywanie danych...")
diabetes = pd.read_csv(args.training_data)
# Oddziel cechy od etykiet
X, y = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure','TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].values, diabetes['Diabetic'].values
# Podziel dane na zbiór treningowy i testowy
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.30, random_state=0)
# Wytrenuj model drzewa decyzyjnego
print('Trenowanie modelu drzewa decyzyjnego')
model = DecisionTreeClassifier().fit(X_train, y_train)
# policz skuteczność
y_hat = model.predict(X_test)
acc = np.average(y_hat == y_test)
print('Skuteczność:', acc)
mlflow.log_metric('Accuracy', float(acc))
# policz pole pod krzywą ROC
y_scores = model.predict_proba(X_test)
auc = roc_auc_score(y_test,y_scores[:,1])
print('AUC: ' + str(auc))
mlflow.log_metric('AUC', float(auc))
# narysuj krzywą ROC
fpr, tpr, thresholds = roc_curve(y_test, y_scores[:,1])
fig = plt.figure(figsize=(6, 4))
# Linia przekątnej odpowiadająca losowemu zgadywaniu
plt.plot([0, 1], [0, 1], 'k--')
# Wartości FPR i TPR osiągnięte przez model
plt.plot(fpr, tpr)
plt.xlabel('Odsetek fałszywie pozytywnych')
plt.ylabel('Odsetek prawdziwie pozytywnych')
plt.title('Krzywa ROC')
mlflow.log_figure(fig, "ROC.png")
plt.show()
# Zapisz wytrenowany model w formacie MLflow, gotowy do odczytania przez kolejny krok
print("Zapisywanie modelu w folderze", output_folder)
# Format cloudpickle: domyślny skops odmawia zapisu drzewa decyzyjnego.
mlflow.sklearn.save_model(model, path=output_folder,
serialization_format="cloudpickle")
Skrypt drugiego kroku zarejestruje w obszarze roboczym model zapisany przez krok pierwszy. Ma jeden parametr, model_folder, czyli ścieżkę, pod którą model został zapisany. Rejestrowany jest cały folder w formacie MLflow - model nie jest po drodze wczytywany do pamięci ani zapisywany po raz drugi.
%%writefile $experiment_folder/register_diabetes.py
# Import bibliotek
import argparse
import os
import mlflow
# Odczytaj parametry
parser = argparse.ArgumentParser()
parser.add_argument('--model_folder', type=str, dest='model_folder', default="diabetes_model", help='lokalizacja modelu')
args = parser.parse_args()
model_folder = args.model_folder
# Zarejestruj model zapisany przez krok trenujący. Rejestrujemy folder wprost,
# bez wczytywania modelu do pamięci i zapisywania go po raz drugi.
print("Rejestrowanie modelu z " + model_folder)
mlflow.register_model(f"file://{os.path.abspath(model_folder)}", "diabetes_model")
print("Model zarejestrowany jako diabetes_model")
Przygotowanie środowiska obliczeniowego dla kroków potoku¶
Oba kroki uruchomisz na tym samym klastrze obliczeniowym, ale warto pamiętać, że każdy krok wykonuje się niezależnie - gdyby miało to sens, każdemu można przypisać inne środowisko obliczeniowe.
Najpierw pobierz (lub utwórz) klaster obliczeniowy.
from azure.ai.ml.entities import AmlCompute
cluster_name = "aml-cluster"
try:
pipeline_cluster = ml_client.compute.get(cluster_name)
print('Znaleziono istniejący klaster - zostanie użyty.')
except Exception:
# Jeśli nie istnieje, utwórz go
pipeline_cluster = AmlCompute(
name=cluster_name,
type="amlcompute",
size="Standard_D2as_v4",
min_instances=0,
max_instances=2,
idle_time_before_scale_down=300,
)
ml_client.compute.begin_create_or_update(pipeline_cluster).result()
print(f"Środowisko obliczeniowe '{pipeline_cluster.name}' jest gotowe do użycia.")
Kroki potrzebują środowiska Pythona z zainstalowanymi pakietami, których używają skrypty - utwórz je więc i zarejestruj.
from azure.ai.ml.entities import Environment
# Zdefiniuj zależności conda dla kroków potoku.
# mlflow przypięty: nowszy zrywa logowanie artefaktów przez azureml-mlflow.
conda_dependencies = {
"name": "diabetes-pipeline-env",
"channels": ["conda-forge"],
"dependencies": [
"python=3.10",
"scikit-learn",
"pandas",
"matplotlib",
"pip",
{"pip": ["mlflow<=3.15.0", "azureml-mlflow"]},
],
}
pipeline_env = Environment(
name="diabetes-pipeline-env",
description="Training environment for the diabetes pipeline",
conda_file=conda_dependencies,
image="mcr.microsoft.com/azureml/openmpi5.0-ubuntu24.04",
)
pipeline_env = ml_client.environments.create_or_update(pipeline_env)
print(f"Zarejestrowano środowisko '{pipeline_env.name}', wersja {pipeline_env.version}.")
Zbudowanie i uruchomienie potoku¶
Teraz można złożyć potok i go uruchomić.
Najpierw trzeba zdefiniować komponenty. Pierwszy z nich musi zapisać wytrenowany model do folderu, z którego odczyta go drugi. Każdy komponent działa we własnym kontenerze (a może nawet na innym środowisku obliczeniowym), więc jego wejścia i wyjścia deklaruje się jawnie, a Azure ML sam zadba o przeniesienie danych między miejscami składowania poszczególnych kroków. Dla folderu z modelem zdefiniujesz Output w komponencie trenującym i użyjesz go jako Input komponentu rejestrującego.
from azure.ai.ml import command, Input, Output
from azure.ai.ml.constants import AssetTypes
# Krok 1: wytrenuj model
train_step = command(
name="train_diabetes_model",
display_name="Train Model",
description="Trains a diabetes classification model",
inputs={"training_data": Input(type=AssetTypes.URI_FILE)},
outputs={"model_output": Output(type=AssetTypes.MLFLOW_MODEL)},
code=experiment_folder,
command="python train_diabetes.py --training_data ${{inputs.training_data}} --output_folder ${{outputs.model_output}}",
environment=f"{pipeline_env.name}:{pipeline_env.version}",
)
# Krok 2: zarejestruj wytrenowany model
register_step = command(
name="register_diabetes_model",
display_name="Register Model",
description="Registers the trained model in the workspace",
inputs={"model_folder": Input(type=AssetTypes.MLFLOW_MODEL)},
code=experiment_folder,
command="python register_diabetes.py --model_folder ${{inputs.model_folder}}",
environment=f"{pipeline_env.name}:{pipeline_env.version}",
)
print("Kroki potoku zdefiniowane")
Wszystko jest gotowe. Złóż potok ze zdefiniowanych komponentów przy użyciu dekoratora @dsl.pipeline i uruchom go jako zadanie.
from azure.ai.ml import dsl
# Zbuduj potok
@dsl.pipeline(description="Trains and registers a diabetes classification model")
def diabetes_training_pipeline(pipeline_input_data: Input(type=AssetTypes.URI_FILE)):
train_job = train_step(training_data=pipeline_input_data)
register_step(model_folder=train_job.outputs.model_output)
return {"trained_model": train_job.outputs.model_output}
pipeline_job = diabetes_training_pipeline(
pipeline_input_data=Input(type=AssetTypes.URI_FILE, path="data/diabetes.csv")
)
# Ustaw domyślne środowisko obliczeniowe dla kroków, które nie wskazują własnego
pipeline_job.settings.default_compute = "aml-cluster"
print("Potok zbudowany.")
# Zleć potok jako zadanie
pipeline_job = ml_client.jobs.create_or_update(
pipeline_job, experiment_name="diabetes-training-pipeline"
)
print("Potok przekazany do wykonania.")
# Wyświetlaj na bieżąco logi aż do zakończenia potoku
ml_client.jobs.stream(pipeline_job.name)
Wywołanie stream powyżej pokazuje postęp potoku aż do jego zakończenia. Zadania potoków możesz też obserwować na stronie Jobs w Azure Machine Learning studio.
Po zakończeniu potoku w obszarze roboczym powinna pojawić się nowa wersja modelu diabetes_model. Sprawdź to poniższym kodem.
for model in ml_client.models.list(name="diabetes_model"):
print(f"{model.name} wersja: {model.version}")
for tag_name, tag in (model.tags or {}).items():
print(f"\t{tag_name} : {tag}")
To prosty przykład, który pokazuje samą zasadę działania. W praktyce kroki potoku bywają znacznie bardziej rozbudowane - można na przykład ocenić model na danych testowych, porównać uzyskaną wartość AUC lub skuteczności z wcześniej zarejestrowanymi wersjami modelu i zarejestrować nowy model tylko wtedy, gdy okaże się lepszy.
Za pomocą rozszerzenia Azure Machine Learning dla Azure DevOps można połączyć potoki Azure ML z potokami Azure DevOps (tak, to mylące, że nazywają się tak samo) i włączyć ponowne trenowanie modelu w proces ciągłej integracji i ciągłego wdrażania (CI/CD). Potok build w Azure DevOps może na przykład uruchamiać potok Azure ML, który trenuje i rejestruje model, a rejestracja modelu może z kolei wyzwalać potok release, wdrażający model jako punkt końcowy (ang. endpoint) razem z aplikacją, która z niego korzysta.