ONNX-inference op Spark

ONNX (Open Neural Network Exchange) biedt een draagbare, hardwaregeoptimeerde runtime voor machine learning-modellen. Door een model naar de ONNX-indeling te converteren, kunt u batchinferentie uitvoeren op Spark met een lagere latentie en zonder afhankelijk te zijn van het oorspronkelijke trainingsframework bij het voorspellen.

In dit artikel traint u een LightGBM-model met SynapseML, converteert u het naar de ONNX-indeling en gebruikt u vervolgens het ONNX-model om deductie uit te voeren op Spark in Microsoft Fabric.

Vereiste voorwaarden

• Haal een Microsoft Fabric-abonnement op. Of meld u aan voor een gratis proefversie van Microsoft Fabric.

• Meld u aan bij Microsoft Fabric.

• Schakel over naar Fabric met behulp van de ervaringsschakelaar aan de linkerkant van de startpagina.

  

• Koppel uw notitieblok aan een lakehouse. Selecteer aan de linkerkant van uw notitieblok Toevoegen om een bestaand lakehouse toe te voegen of maak er een.
• Fabric Runtime 1.2 of hoger.

De vereiste pakketten installeren

Voer de volgende cel in uw notebook uit om de vereiste pakketten te installeren. Het onnxmltools-pakket is niet vooraf geïnstalleerd in de Fabric-runtime.

%pip install onnxmltools --quiet

Nadat de installatie is voltooid, controleert u of de pakketten beschikbaar zijn:

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

De voorbeeldgegevens laden

Laad de gegevensset voor faillissementsvoorspelling van openbare 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))

De weergegeven tabel bevat kolommen zoals:

Een LightGBM-model trainen

Gebruik de VectorAssembler om kenmerkkolommen te combineren en train vervolgens een 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)

Controleer of het model succesvol is getraind:

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

Het model converteren naar ONNX-indeling

Exporteer het getrainde model naar een LightGBM booster en converteer het vervolgens naar 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))

Controleer of de ONNX-conversie is geslaagd:

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

De uitvoer toont de nettoladinggrootte van het ONNX-model in bytes (meestal ongeveer 800.000 bytes).

Het ONNX-model laden en configureren

Laad de ONNX-nettolading in een SynapseML ONNXModel en inspecteer de invoer en uitvoer van het model:

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

De uitvoer bevat de invoer- en uitvoerknooppunten van het model.

Configureer het model door invoer- en uitvoerkolommen toe te kaarten. De FeedDict invoernamen van het ONNX-model worden toegewezen aan DataFrame-kolomnamen. De FetchDict wijst gewenste uitvoerkolomnamen toe aan de uitvoernamen van het ONNX-model:

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

Deductie uitvoeren

Testgegevens maken en transformeren via het ONNX-model:

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

De uitvoer moet kolommen bevatten voor features, predictionen probability:

Controleer de gegenereerde resultaten van de deductie:

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"

De uitvoer bevestigt dat alle testrijen zijn gescoord en het resultaat dataframe de features, predictionen probability kolommen bevat.

Troubleshooting

De hulpbronnen opschonen

Als u het dataframe met cache niet meer nodig hebt, maakt u deze ongedaan om clustergeheugen vrij te maken:

testDf.unpersist()

Verwante inhoud

• Kernel SHAP gebruiken om een tabellair classificatiemodel uit te leggen
• SynapseML gebruiken voor anomaliedetectie met meerdere variabelen
• Een zoekmachine bouwen met SynapseML