Wnioskowanie ONNX na platformie Spark

ONNX (Open Neural Network Exchange) udostępnia przenośne, zoptymalizowane pod kątem sprzętu środowisko uruchomieniowe dla modeli uczenia maszynowego. Po przekonwertowaniu modelu do formatu ONNX można uruchamiać wnioskowanie wsadowe w Sparku z mniejszym opóźnieniem i bez zależności od oryginalnego frameworka uczenia na etapie predykcji.

W tym artykule wytrenujesz model LightGBM za pomocą usługi SynapseML, przekonwertujesz go na format ONNX, a następnie użyjesz modelu ONNX, aby przeprowadzić wnioskowanie na platformie Spark w Microsoft Fabric.

Wymagania wstępne

• Uzyskaj subskrypcję usługi Microsoft Fabric. Możesz też utworzyć konto bezpłatnej wersji próbnej usługi Microsoft Fabric.

• Zaloguj się do Microsoft Fabric.

• Przełącz na Fabric, używając przełącznika doświadczenia w dolnym lewym rogu twojej strony głównej.

  

• Dołącz swój notebook do lakehouse. Po lewej stronie notesnika wybierz pozycję Dodaj, aby dodać istniejący obiekt lakehouse lub go utworzyć.
• Fabric Runtime 1.2 lub nowsza wersja.

Instalowanie wymaganych pakietów

Uruchom następującą komórkę w notesie, aby zainstalować wymagane pakiety. Pakiet onnxmltools nie jest wstępnie zainstalowany w środowisku uruchomieniowym Fabric.

%pip install onnxmltools --quiet

Po zakończeniu instalacji sprawdź, czy pakiety są dostępne:

import onnxmltools
import lightgbm
print(f"onnxmltools version: {onnxmltools.__version__}")
print(f"lightgbm version: {lightgbm.__version__}")

Ładowanie przykładowych danych

Załaduj zestaw danych przewidywania bankructwa z publicznych Azure Blob Storage:

df = (
    spark.read.format("csv")
    .option("header", True)
    .option("inferSchema", True)
    .load(
        "wasbs://publicwasb@mmlspark.blob.core.windows.net/company_bankruptcy_prediction_data.csv"
    )
)

print(f"Rows: {df.count()}, Columns: {len(df.columns)}")
display(df.limit(5))

Wyświetlona tabela zawiera kolumny, takie jak:

Trenowanie modelu LightGBM

Użyj VectorAssembler, aby połączyć kolumny cech, a następnie wytrenuj LightGBMClassifier:

from pyspark.ml.feature import VectorAssembler
from synapse.ml.lightgbm import LightGBMClassifier

feature_cols = df.columns[1:]
featurizer = VectorAssembler(inputCols=feature_cols, outputCol="features")

train_data = featurizer.transform(df)["Bankrupt?", "features"]

model = (
    LightGBMClassifier(featuresCol="features", labelCol="Bankrupt?")
    .setDataTransferMode("bulk")
    .setEarlyStoppingRound(300)
    .setLambdaL1(0.5)
    .setNumIterations(1000)
    .setNumThreads(-1)
    .setMaxDeltaStep(0.5)
    .setNumLeaves(31)
    .setMaxDepth(-1)
    .setBaggingFraction(0.7)
    .setFeatureFraction(0.7)
    .setBaggingFreq(2)
    .setObjective("binary")
    .setIsUnbalance(True)
    .setMinSumHessianInLeaf(20)
    .setMinGainToSplit(0.01)
)

model = model.fit(train_data)

Sprawdź, czy model został pomyślnie wytrenowany:

print(f"Model type: {type(model).__name__}")
print(f"Number of features: {len(feature_cols)}")

Konwertowanie modelu na format ONNX

Wyeksportuj wytrenowany model do wzmacniacza LightGBM, a następnie przekonwertuj go na ONNX:

import lightgbm as lgb
from typing import Union
from lightgbm import Booster, LGBMClassifier
from onnxmltools.convert import convert_lightgbm
from onnxmltools.convert.common.data_types import FloatTensorType


def convert_to_onnx(lgbm_model: Union[LGBMClassifier, Booster], input_size: int) -> bytes:
    initial_types = [("input", FloatTensorType([-1, input_size]))]
    onnx_model = convert_lightgbm(
        lgbm_model, initial_types=initial_types, target_opset=13
    )
    return onnx_model.SerializeToString()


booster_model_str = model.getLightGBMBooster().modelStr().get()
booster = lgb.Booster(model_str=booster_model_str)
model_payload_ml = convert_to_onnx(booster, len(feature_cols))

Sprawdź, czy konwersja ONNX zakończyła się pomyślnie:

print(f"ONNX model payload size: {len(model_payload_ml)} bytes")
assert len(model_payload_ml) > 0, "ONNX conversion failed: empty payload"

Dane wyjściowe pokazują rozmiar ładunku modelu ONNX w bajtach (zazwyczaj około 800 000 bajtów).

Ładowanie i konfigurowanie modelu ONNX

Załaduj ładunek ONNX do kodu SynapseML ONNXModel i sprawdź dane wejściowe i wyjściowe modelu:

from synapse.ml.onnx import ONNXModel

onnx_ml = ONNXModel().setModelPayload(model_payload_ml)

print("Model inputs:" + str(onnx_ml.getModelInputs()))
print("Model outputs:" + str(onnx_ml.getModelOutputs()))

Dane wyjściowe zawierają listę węzłów wejściowych i wyjściowych modelu.

Skonfiguruj model, mapując kolumny wejściowe i wyjściowe. FeedDict mapuje nazwy wejściowe modelu ONNX na nazwy kolumn DataFrame. Element FetchDict mapuje żądane nazwy kolumn wyjściowych na nazwy wyjściowe modelu ONNX:

onnx_ml = (
    onnx_ml.setDeviceType("CPU")
    .setFeedDict({"input": "features"})
    .setFetchDict({"probability": "probabilities", "prediction": "label"})
    .setMiniBatchSize(5000)
)

Uruchamianie wnioskowania

Utwórz dane testowe i przekształć je za pomocą modelu ONNX:

from pyspark.ml.feature import VectorAssembler
import pandas as pd
import numpy as np

n = 10000
m = 95
test = np.random.rand(n, m)
testPdf = pd.DataFrame(test)
cols = list(map(str, testPdf.columns))
testDf = spark.createDataFrame(testPdf)
testDf = testDf.repartition(4)
testDf = (
    VectorAssembler()
    .setInputCols(cols)
    .setOutputCol("features")
    .transform(testDf)
    .drop(*cols)
    .cache()
)

display(onnx_ml.transform(testDf))

Dane wyjściowe powinny zawierać kolumny dla features, prediction i probability:

Sprawdź, czy wnioskowanie wygenerowało wyniki:

results = onnx_ml.transform(testDf)
print(f"Result count: {results.count()}")
print(f"Output columns: {results.columns}")
assert "prediction" in results.columns, "Missing prediction column"
assert "probability" in results.columns, "Missing probability column"

Dane wyjściowe potwierdzają, że wszystkie wiersze testowe zostały ocenione, a wynikowa ramka danych zawiera kolumny features, prediction i probability.

Troubleshooting

Uprzątnij zasoby

Jeśli nie potrzebujesz już zbuforowanej testowej ramki danych, usuń ją z pamięci, aby zwolnić pamięć klastra:

testDf.unpersist()

Powiązana treść

• Jak używać algorytmu SHAP jądra do wyjaśnienia modelu klasyfikacji tabelarycznej
• Jak używać usługi SynapseML do wykrywania anomalii wielowariancyjnych
• Jak utworzyć aparat wyszukiwania za pomocą usługi SynapseML