Batchdeductie uitvoeren met behulp van een Spark DataFrame

Op deze pagina wordt beschreven hoe u batchdeductie uitvoert op een Spark DataFrame met behulp van een geregistreerd model in Databricks. De werkstroom is van toepassing op verschillende machine learning- en Deep Learning-modellen, waaronder TensorFlow, PyTorch en scikit-learn. Het bevat aanbevolen procedures voor het laden van gegevens, modeldeductie en het afstemmen van prestaties.

Voor modeldeductie voor Deep Learning-toepassingen raadt Azure Databricks de volgende werkstroom aan. Zie bijvoorbeeld notebooks die TensorFlow en PyTorch gebruiken Batch-deductievoorbeelden.

Werkstroom voor modeldeductie

Databricks raadt de volgende werkstroom aan voor het uitvoeren van batchdeductie met behulp van Spark DataFrames.

Stap 1: Omgeving instellen

Zorg ervoor dat uw cluster een compatibele Databricks ML Runtime-versie uitvoert die overeenkomt met de trainingsomgeving. Het model, geregistreerd met MLflow, bevat de vereisten die kunnen worden geïnstalleerd om ervoor te zorgen dat de trainings- en deductieomgevingen overeenkomen.

requirements_path = os.path.join(local_path, "requirements.txt")
if not os.path.exists(requirements_path):
    dbutils.fs.put("file:" + requirements_path, "", True)

%pip install -r $requirements_path
%restart_python

Stap 2: Gegevens laden in Spark DataFrames

Afhankelijk van het gegevenstype gebruikt u de juiste methode om gegevens in een Spark DataFrame te laden:

Stap 3: Model laden uit modelregister

In dit voorbeeld wordt een model uit het Databricks-modelregister gebruikt voor deductie.

predict_udf = mlflow.pyfunc.spark_udf(spark, model_uri)

Stap 4: Modeldeductie uitvoeren met Pandas UDF's

Pandas UDF's maken gebruik van Apache Arrow voor efficiënte gegevensoverdracht en pandas voor verwerking. De typische stappen voor inferentie met pandas udfs zijn:

1. Laad het getrainde model: gebruik MLflow om een Spark UDF te maken voor deductie.
2. Invoergegevens vooraf verwerken: zorg ervoor dat het invoerschema overeenkomt met de modelvereisten.
3. Modelvoorspelling uitvoeren: gebruik de UDF-functie van het model in het DataFrame.

df_result = df_spark.withColumn("prediction", predict_udf(*df_spark.columns))

4. (Aanbevolen) Sla voorspellingen op in Unity Catalog.

In het volgende voorbeeld worden voorspellingen opgeslagen in Unity Catalog.

df_result.write.mode("overwrite").saveAsTable(output_table)

Prestaties afstemmen voor modeldeductie

Deze sectie bevat enkele tips voor foutopsporing en het afstemmen van prestaties voor modeldeductie in Azure Databricks. Voor een overzicht, zie de Batch-inferentie uitvoeren met behulp van een Spark DataFrame.

Normaal gesproken zijn er twee hoofdonderdelen in modeldeductie: pijplijn voor gegevensinvoer en modeldeductie. De pijplijn voor gegevensinvoer is zwaar belast met data-I/O en modelinferentie vereist veel rekenkracht. Het bepalen van het knelpunt van de werkstroom is eenvoudig. Hier volgen enkele benaderingen:

• Verminder het model tot een triviaal model en meet de voorbeelden per seconde. Als het verschil tussen de eind- en eindtijd tussen het volledige model en het triviale model minimaal is, is de pijplijn voor gegevensinvoer waarschijnlijk een knelpunt, anders is modeldeductie het knelpunt.
• Als u modeldeductie uitvoert met GPU, controleert u het GPU-gebruik metrische gegevens. Als het GPU-gebruik niet continu hoog is, kan de pijplijn voor gegevensinvoer het knelpunt zijn.

Pijplijn voor gegevensinvoer optimaliseren

Het gebruik van GPU's kan de actieve snelheid voor modeldeductie efficiënt optimaliseren. Naarmate GPU's en andere accelerators sneller worden, is het belangrijk dat de pijplijn voor gegevensinvoer de vraag bijhoudt. De gegevensinvoerpijplijn leest de gegevens in Spark DataFrames, transformeert deze en laadt deze als invoer voor modeldeductie. Als gegevensinvoer het knelpunt is, volgen hier enkele tips om de I/O-doorvoer te verhogen:

• Stel het maximumaantal records per batch in. Een groter aantal records kan de I/O-overhead verminderen om de UDF-functie aan te roepen zolang de records in het geheugen passen. Als u de batchgrootte wilt instellen, stelt u de volgende configuratie in:

  spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "5000")
  

• Laad de gegevens in batches en haal deze van tevoren op bij het vooraf verwerken van de invoergegevens in de pandas UDF.

  Voor TensorFlow raadt Azure Databricks aan om de tf.data-APIte gebruiken. U kunt de kaart parallel parseren door num_parallel_calls in te stellen in een map functie en prefetch aan te roepen en batch aan te roepen voor prefetching en batching.

  dataset.map(parse_example, num_parallel_calls=num_process).prefetch(prefetch_size).batch(batch_size)
  

  Voor PyTorch raadt Azure Databricks aan om de DataLoader-klassete gebruiken. U kunt batch_size instellen voor batchverwerking en num_workers voor het laden van parallelle gegevens.

  torch.utils.data.DataLoader(images, batch_size=batch_size, num_workers=num_process)
  

Batch-inferentie voorbeelden

De voorbeelden in deze sectie volgen de aanbevolen deep learning-deductiewerkstroom. In deze voorbeelden ziet u hoe u modeldeductie uitvoert met behulp van een vooraf getraind neuraal netwerkmodel (ResNets).

• Modeldeductie met behulp van TensorFlow en TensorRT
• Modeldeductie met behulp van PyTorch

Gestructureerde gegevensextractie en batchdeductie met behulp van Spark UDF

In het volgende voorbeeldnotebook ziet u de ontwikkeling, logboekregistratie en evaluatie van een eenvoudige agent voor gestructureerde gegevensextractie om onbewerkte, ongestructureerde gegevens te transformeren in georganiseerde, bruikbare informatie via geautomatiseerde extractietechnieken. Deze benadering laat zien hoe u aangepaste agents implementeert voor batchdeductie met behulp van de PythonModel klasse van MLflow en het vastgelegde agentmodel als een Spark User-Defined Function (UDF). In dit notebook wordt ook getoond hoe u Mosaic AI Agent Evaluation kunt gebruiken om de nauwkeurigheid te evalueren met behulp van gegevens uit de werkelijkheid.

Gestructureerde gegevensextractie en batchdeductie met behulp van Spark UDF

Notebook krijgen