ONNX-Rückschluss auf Spark

ONNX (Open Neural Network Exchange) bietet eine tragbare, hardwareoptimierte Laufzeit für Machine Learning-Modelle. Durch die Konvertierung eines Modells in das ONNX-Format können Sie Batchinferenz auf Spark mit geringerer Latenz ausführen, ohne beim Vorhersagezeitpunkt vom ursprünglichen Trainings-Framework abhängig zu sein.

In diesem Artikel trainieren Sie ein LightGBM-Modell mit SynapseML, konvertieren es in das ONNX-Format und verwenden dann das ONNX-Modell, um die Ableitung von Spark in Microsoft Fabric durchzuführen.

Voraussetzungen

• Erwerben Sie ein Microsoft Fabric-Abonnement. Oder registrieren Sie sich für eine kostenlose Microsoft Fabric-Testversion.

• Melden Sie sich bei Microsoft Fabric an.

• Wechseln Sie zu Fabric, indem Sie den Benutzeroberflächenschalter auf der unteren linken Seite Ihrer Startseite verwenden.

  

• Fügen Sie Ihr Notizbuch an ein Seehaus an. Wählen Sie auf der linken Seite Ihres Notizbuchs "Hinzufügen" aus, um ein vorhandenes Seehaus hinzuzufügen oder eins zu erstellen.
• Fabric Runtime 1.2 oder höher.

Installieren Sie die erforderlichen Pakete

Führen Sie die folgende Zelle in Ihrem Notizbuch aus, um die erforderlichen Pakete zu installieren. Das onnxmltools-Paket ist in der Fabric Laufzeit nicht vorinstalliert.

%pip install onnxmltools --quiet

Überprüfen Sie nach Abschluss der Installation, ob die Pakete verfügbar sind:

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

Laden der Beispieldaten

Laden Sie das Datenset für die Insolvenzvorhersage aus öffentlichen 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))

Die angezeigte Tabelle enthält Spalten wie:

Trainieren eines LightGBM-Modells

Verwenden Sie VectorAssembler, um Feature-Spalten zu kombinieren, und trainieren Sie dann ein 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)

Überprüfen Sie, ob das Modell erfolgreich trainiert wurde:

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

Konvertieren des Modells in das ONNX-Format

Exportieren Sie das trainierte Modell in einen LightGBM-Booster und konvertieren Sie es dann in 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))

Überprüfen Sie, ob die ONNX-Konvertierung erfolgreich war:

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

Die Ausgabe zeigt die ONNX-Modellnutzlastgröße in Byte an (in der Regel ca. 800.000 Bytes).

Laden und Konfigurieren des ONNX-Modells

Laden Sie das ONNX-Paket in SynapseML ONNXModel und überprüfen Sie die Eingaben und Ausgaben des Modells:

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()))

Die Ausgabe listet die Eingabe- und Ausgabeknoten des Modells auf.

Konfigurieren Sie das Modell durch Zuordnen von Eingabe- und Ausgabespalten. Das FeedDict ordnet die Eingabenamen des ONNX-Modells den Spaltennamen des DataFrames zu. Die FetchDict ordnet die gewünschten Ausgabespaltennamen den Ausgabenamen des ONNX-Modells zu:

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

Inferenz ausführen

Erstellen Sie Testdaten, und transformieren Sie sie über das ONNX-Modell:

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))

Die Ausgabe sollte Spalten für features, prediction und probability enthalten:

Überprüfen Sie die ergebnisse der Ableitung:

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"

Die Ausgabe bestätigt, dass alle Testzeilen bewertet wurden und das Ergebnis DataFrame die features, predictionund probability Spalten enthält.

Problembehandlung

Bereinigen von Ressourcen

Wenn Sie den zwischengespeicherten Test DataFrame nicht mehr benötigen, lösen Sie ihn aus, um Clusterspeicher freizugeben:

testDf.unpersist()

Verwandte Inhalte

• Verwenden von Kernel SHAP, um ein tabellarisches Klassifizierungsmodell zu erklären
• So verwenden Sie SynapseML für die multivariate Anomalieerkennung
• So erstellen Sie eine Suchmaschine mit SynapseML