Tworzenie usługi wnioskowania wsadowego¶

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
ml_client = MLClient.from_config(credential=credential)
print(f"Azure ML gotowe do pracy z obszarem roboczym {ml_client.workspace_name}")

import pandas as pd
import numpy as np
import joblib
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_auc_score
from azure.ai.ml.entities import Model
from azure.ai.ml.constants import AssetTypes

# wczytaj zbiór danych o cukrzycy
print("Wczytywanie danych...")
diabetes = pd.read_csv('data/diabetes.csv')

# Rozdziel cechy i etykiety
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)

# policz AUC
y_scores = model.predict_proba(X_test)
auc = roc_auc_score(y_test, y_scores[:,1])
print('AUC: ' + str(auc))

# Zapisz wytrenowany model do pliku
model_file = 'diabetes_model.pkl'
joblib.dump(value=model, filename=model_file)

# Zarejestruj model
registered_model = Model(
    path=model_file,
    type=AssetTypes.CUSTOM_MODEL,
    name="diabetes_model",
    description="A decision tree model that classifies patients by their likelihood of being diabetic.",
    tags={"Training context": "Inline Training"},
    properties={"AUC": str(auc), "Accuracy": str(acc)},
)
ml_client.models.create_or_update(registered_model)

print('Model wytrenowany i zarejestrowany.')

import pandas as pd
import os
from azure.ai.ml.entities import Data
from azure.ai.ml.constants import AssetTypes

# Wczytaj dane o cukrzycy (nowi pacjenci, dla których chcemy predykcji)
diabetes = pd.read_csv('data/diabetes2.csv')
# Wylosuj 100 obserwacji, biorąc same kolumny z cechami (bez etykiety Diabetic)
sample = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure','TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].sample(n=100).values

# Utwórz folder lokalny
batch_folder = './batch-data'
os.makedirs(batch_folder, exist_ok=True)
print("Folder utworzony!")

# Zapisz każdą obserwację w osobnym pliku
print("Zapisywanie plików...")
for i in range(100):
    fname = str(i+1) + '.csv'
    sample[i].tofile(os.path.join(batch_folder, fname), sep=",")
print("Pliki zapisane!")

# Zarejestruj folder z plikami jako zasób danych typu uri_folder
print("Wysyłanie plików i rejestrowanie zasobu danych...")
batch_data_set = Data(
    path=batch_folder,
    type=AssetTypes.URI_FOLDER,
    description="A folder of new patient observations to be scored in batch",
    name="diabetes_batch_data",
)
ml_client.data.create_or_update(batch_data_set)

print("Gotowe!")

from azure.ai.ml.entities import AmlCompute

cluster_name = "aml-cluster"

# Sprawdź, czy klaster już istnieje
try:
    inference_cluster = ml_client.compute.get(cluster_name)
    print('Klaster już istnieje - używamy go.')
except Exception:
    # Utwórz klaster obliczeniowy Azure ML
    compute_config = AmlCompute(
        name=cluster_name,
        type="amlcompute",
        size="Standard_D2as_v4",
        min_instances=0,
        max_instances=3,
        idle_time_before_scale_down=300,
    )
    inference_cluster = ml_client.compute.begin_create_or_update(compute_config).result()

print(inference_cluster.name, "jest dostępny.")

import os
# Utwórz folder na pliki wdrożenia
experiment_folder = 'batch_deploy'
os.makedirs(experiment_folder, exist_ok=True)

print(experiment_folder)

%%writefile $experiment_folder/batch_diabetes.py
import os
import numpy as np
import joblib


def init():
    # Wykonuje się raz, przy inicjalizacji wdrożenia
    global model

    # Zmienną AZUREML_MODEL_DIR ustawia wdrożenie wsadowe; wskazuje ona
    # folder z plikami zarejestrowanego modelu
    model_path = os.path.join(os.getenv("AZUREML_MODEL_DIR"), "diabetes_model.pkl")
    model = joblib.load(model_path)


def run(mini_batch):
    # Wykonuje się dla każdej porcji plików (mini-batcha)
    resultList = []

    # przetwórz każdy plik z porcji
    for f in mini_batch:
        # Wczytaj dane rozdzielone przecinkami do tablicy
        data = np.genfromtxt(f, delimiter=',')
        # Zmień kształt na tablicę dwuwymiarową - model oczekuje zbioru obserwacji, nie jednej
        prediction = model.predict(data.reshape(1, -1))
        # Dopisz predykcję do wyników
        resultList.append("{}: {}".format(os.path.basename(f), prediction[0]))
    return resultList

%%writefile $experiment_folder/batch_env.yml
name: batch-environment
dependencies:
  - python=3.8
  - numpy
  - pip
  - pip:
      - scikit-learn
      - joblib

from azure.ai.ml.entities import Environment

batch_env = Environment(
    name="batch-environment",
    conda_file=f"{experiment_folder}/batch_env.yml",
    image="mcr.microsoft.com/azureml/openmpi5.0-ubuntu24.04",
)
print('Konfiguracja środowiska gotowa.')

from azure.ai.ml.entities import BatchEndpoint, ModelBatchDeployment, ModelBatchDeploymentSettings, CodeConfiguration
from azure.ai.ml.constants import BatchDeploymentOutputAction

endpoint_name = "diabetes-batch-endpoint"

# Utwórz punkt końcowy wsadowy
endpoint = BatchEndpoint(
    name=endpoint_name,
    description="A batch endpoint for scoring diabetes patient data",
)
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

# Pobierz zarejestrowany model, który chcesz wdrożyć
model = ml_client.models.get(name="diabetes_model", label="latest")

# Utwórz wdrożenie wsadowe
batch_deployment = ModelBatchDeployment(
    name="diabetes-batch-dpl",
    endpoint_name=endpoint_name,
    model=model,
    environment=batch_env,
    code_configuration=CodeConfiguration(code=experiment_folder, scoring_script="batch_diabetes.py"),
    compute=cluster_name,
    settings=ModelBatchDeploymentSettings(
        instance_count=2,
        max_concurrency_per_instance=2,
        mini_batch_size=5,
        output_action=BatchDeploymentOutputAction.APPEND_ROW,
        output_file_name="predictions.csv",
    ),
)
ml_client.batch_deployments.begin_create_or_update(batch_deployment).result()

# Ustaw to wdrożenie jako domyślne dla punktu końcowego
endpoint = ml_client.batch_endpoints.get(endpoint_name)
endpoint.defaults.deployment_name = batch_deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

print('Punkt końcowy wsadowy i wdrożenie utworzone.')

from azure.ai.ml import Input
from azure.ai.ml.constants import AssetTypes

# Pobierz najnowszą wersję zarejestrowanego zasobu danych wsadowych
batch_data_asset = ml_client.data.get(name="diabetes_batch_data", label="latest")

job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint_name,
    inputs={
        "input_data": Input(
            type=AssetTypes.URI_FOLDER,
            path=f"azureml:diabetes_batch_data:{batch_data_asset.version}",
        )
    },
)

ml_client.jobs.stream(job.name)

import glob
import shutil
import pandas as pd

shutil.rmtree('diabetes-results', ignore_errors=True)

# Wdrożenie uruchamia skrypt scoringowy w zadaniu podrzędnym - pobierz referencję do niego
scoring_job = list(ml_client.jobs.list(parent_job_name=job.name))[0]

# Pobierz jego wyjście (dla wdrożenia wsadowego modelu domyślnie nazywa się "score")
ml_client.jobs.download(name=scoring_job.name, download_path='diabetes-results', output_name='score')

# Odszukaj plik z predykcjami
result_file = glob.glob('diabetes-results/**/predictions.csv', recursive=True)[0]

# uporządkuj format wyników
df = pd.read_csv(result_file, delimiter=":", header=None)
df.columns = ["File", "Prediction"]

# Wyświetl pierwsze 20 wyników
df.head(20)

# ml_client.batch_endpoints.begin_delete(name=endpoint_name)