Snowflake Snowpark: Eine umfassende Einführung

Was ist Snowflake Snowpark?

Klassisches Machine Learning holt die Daten aus Datenbanken dorthin, wo die Modelle laufen. Mit dem aktuellen KI-Boom und den schieren Datenmengen wird dieser Ansatz zunehmend unpraktisch.

Terabytes an Daten müssen für Bereinigung, Analyse und Modelltraining aus der Datenbank in Client-Anwendungen verschoben werden. Diese scheinbar harmlose Hin- und Rückfahrt verschwendet wertvolle Ressourcen. Deshalb setzen immer mehr Unternehmen auf In-Database-Technologien, um Datenbewegungen zu minimieren und Abläufe reibungslos im Datenbankumfeld auszuführen.

Eine der besten In-Database-Technologien am Markt ist Snowpark von Snowflake Cloud. Snowpark ist ein Set aus Bibliotheken und Runtimes, mit dem du Programmiersprachen sicher in Snowflake Cloud ausführen kannst, um Datenpipelines und Machine-Learning-Modelle in derselben Umgebung wie deine Snowflake-Datenbanken zu entwickeln.

In diesem Tutorial besprechen wir die Grundlagen von Snowpark und wie du es in deinen Projekten einsetzt. Wir setzen SQL- und Snowflake-Kenntnisse voraus—falls du diese zuerst auffrischen möchtest, starte mit unserem Lernpfad SQL Grundlagen oder lies dieses Snowflake-Tutorial für Einsteiger.

Los geht’s!

Warum Snowflake Snowpark?

Snowflake Snowpark ist ein Set aus Bibliotheken und Runtimes, mit dem du Programmiersprachen wie Python, Java und Scala sicher direkt in Snowflakes Cloud-Plattform zur Datenverarbeitung verwenden kannst.

So entfällt die Notwendigkeit, Daten zur Verarbeitung aus Snowflake herauszubewegen—das steigert Effizienz und Sicherheit. Hier sind die wichtigsten Vorteile:

Daten in Snowflake verarbeiten: Schreibe Code in deiner bevorzugten Sprache, um deine SQL-Datenbanken in Snowflake zu manipulieren und zu analysieren, und führe ihn in Snowflakes sicherer Umgebung aus. Exporte in andere Umgebungen entfallen.
Bessere Performance: Durch die Verarbeitung direkt in Snowflake nutzt Snowpark die elastische, serverlose Architektur der Plattform für effiziente Rechenleistung.
Weniger Kosten und Technik-Overhead: Da Snowflake die Ressourcen bereitstellt, musst du keine separaten Plattformen für Compute und Storage managen.
Arbeite mit dem, was du kennst—wo du willst: Die APIs von Snowpark erlauben Verbindungen zu deinen SQL-Datenbanken aus jeder Umgebung wie Jupyter oder VS Code, um Datenpipelines und ML-Apps zu bauen. Das Beste: Du kombinierst deine Lieblingsbibliotheken wie Pandas, Scikit-learn, XGBoost etc. mit den Snowpark-Frameworks.

Kurz gesagt: Snowpark ist ein mächtiger und dennoch einfacher Weg für Entwickler, Datenpipelines, Machine-Learning-Lösungen und datengetriebene Anwendungen direkt in Snowflake Cloud zu bauen.

Erste Schritte mit Snowpark

Unser Ziel in diesem Tutorial ist es, ein hyperparametertuntes Modell zu trainieren, das auf einer Tabelle einer Snowflake-Datenbank basiert—und das mit Snowpark. Dafür starten wir mit:

Erstellen einer virtuellen Umgebung mit den benötigten Bibliotheken.
Import von Beispieldaten nach Snowflake.
Erstellen einer Snowpark-Session zur Verbindung mit Snowflake.
Laden der importierten Daten in die Session.

Virtuelle Umgebung erstellen

Für dieses Tutorial verwenden wir eine neue conda-Umgebung:

$ conda create -n snowpark python==3.10 -y
$ conda activate snowparkBereitgestellt von

Hinweis: In einer frisch installierten conda-Umgebung musst du vor dem Aktivieren von snowpark einmal $conda init ausführen.

Nach der Aktivierung installieren wir folgende Bibliotheken:

$ pip install snowflake-snowpark-python #The Snowpark API
$ pip install pandas pyarrow numpy matplotlib seabornBereitgestellt von

Wenn du Jupyter verwendest, installiere zusätzlich ipykernel und führe folgenden Befehl aus, damit die genutzte Umgebung als Jupyter-Kernel verfügbar ist:

$ pip install ipykernel
$ ipython kernel install --user --name=snowparkBereitgestellt von

Importieren wir nun allgemeine Bibliotheken, die wir unterwegs benötigen:

import warnings
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
warnings.filterwarnings("ignore")Bereitgestellt von

Beispieldaten in Snowflake importieren

Wir verwenden das Diamonds-Dataset von Seaborn—du kannst die CSV-Datei direkt von meinem GitHub herunterladen.

Sobald du dich in der Snowflake-App angemeldet hast, folge dem GIF unten, um das Dataset in eine neue Datenbanktabelle zu laden (erstelle ein neues Konto, falls du noch keines hast):

In der Praxis importierst du nur selten CSV-Dateien als Tabellen in deine Snowflake-Datenbanken. Meist arbeitest du mit bestehenden Datenbanken, für die dir die Datenbank-Admins deines Unternehmens unterschiedliche Zugriffsrechte vergeben.

Eine Snowpark-Session erstellen

Jetzt müssen wir eine Verbindung zwischen der Snowpark-API und der Snowflake-Cloud herstellen, um unsere Datenbank zu abzufragen. Dafür sind folgende Snowflake-Zugangsdaten erforderlich: Account-Name, Benutzername und Passwort.

Im Bild unten siehst du, wie du Benutzername und Account-Namen im Snowflake-Dashboard findest:

Erstelle eine separate Datei namens config.py, die ein einzelnes Dictionary credentials im folgenden Format enthält:

credentials = (
   {
       "account": "3-4",  # Combine 3 and 4 with a hyphen
       "username": "bexgboost",  # Your username in lowercase
       "password": "your_password",  # Your Snowflake password
   },
)Bereitgestellt von

Aus Sicherheitsgründen speichern wir Zugangsdaten in einer separaten Datei. Wenn du diese Datei in .gitignore aufnimmst, vermeidest du, dass deine Snowflake-Credentials versehentlich veröffentlicht werden.

Nun importieren wir das credentials-Dictionary und die Klasse Session von Snowpark:

from config import credentials
from snowflake.snowpark import SessionBereitgestellt von

Für die Verbindung nutzen wir die Methode Session.builder.configs.create():

connection_parameters = {
   "account": credentials["account"],
   "user": credentials["username"],
   "password": credentials["password"],
}

new_session = Session.builder.configs(connection_parameters).create()
new_session.get_current_user()Bereitgestellt von

'"bexgboost"'Bereitgestellt von

Wenn dein Benutzername ausgegeben wird, ist deine erste Session erfolgreich erstellt!

Importierte Daten in die Session laden

Das Objekt new_session hat Zugriff auf alles, was deinem Account in Snowflake gehört (und wofür du Berechtigungen hast).

Bevor wir Daten laden, sagen wir der Session, dass wir die zuvor erstellte Datenbank test_db verwenden:

new_session.sql("USE DATABASE test_db;").collect()Bereitgestellt von

[Row(status='Statement executed successfully.')]Bereitgestellt von

Dazu verwenden wir die Methode .sql, mit der wir jede Snowflake-SQL-kompatible Abfrage ausführen können.

Nun können wir die Tabelle diamonds aus test_db über die table-Methode des Session-Objekts laden:

diamonds_df = new_session.table("diamonds")
diamonds_df.show(5)Bereitgestellt von

Die Methode .show() entspricht grob DataFrame.describe() in Pandas und zeigt uns, dass die Verbindung zur Tabelle steht.

Genau—wir haben nur eine Verbindung zur Tabelle hergestellt. Das Objekt diamonds_df hält keine Daten, was seine Größe belegt:

import sys
sys.getsizeof(diamonds_df)Bereitgestellt von

48Bereitgestellt von

Es sind nur 48 Bytes, obwohl es eigentlich über 3 MB sein müssten. Schauen wir uns kurz an, warum das so ist.

Snowpark DataFrames verstehen

Bisher haben wir Folgendes erledigt:

Eine virtuelle Umgebung mit relevanten Bibliotheken erstellt.
Beispieldaten in Snowflake importiert.
Eine Snowpark-Session zur Verbindung mit Snowflake erstellt.

Jetzt sprechen wir über Snowpark DataFrames!

Snowpark DataFrames arbeiten lazy

Pandas DataFrames nutzen deinen RAM, sie leben also auf deiner Maschine. Snowpark DataFrames dagegen leben in der Cloud-Plattform von Snowflake, auch wenn du ihre Repräsentation in deiner Entwicklungsumgebung siehst. Die Daten bleiben in der Cloud und müssen für Operationen nicht lokal heruntergeladen werden.

Ein weiterer wichtiger Punkt: Snowpark DataFrames arbeiten lazy. Das heißt, sie führen keine Operationen auf den Daten aus, bis du sie explizit dazu aufforderst.

Stattdessen bauen sie eine logische Darstellung der gewünschten Operationen auf, die dann in optimierte SQL-Queries übersetzt werden, die Snowflake ausführt.

Im Vergleich dazu führt Pandas Operationen sofort aus—also eager.

Leistungstechnisch ist Lazy Evaluation in der Regel deutlich schneller als Eager Execution. Da lazy DataFrames die leistungsfähigen elastischen Compute-Ressourcen von Snowflake nutzen, können sie Operationen bis auf Datenbankebene pushen und wesentlich schneller ausführen.

Um den Unterschied zwischen Lazy Evaluation und Eager Execution greifbar zu machen, stell dir unsere Daten in Snowflake als riesige Bibliothek vor. Es gibt zwei Szenarien:

Eager Execution (Pandas DataFrames): Wir holen alle benötigten Bücher an unseren Schreibtisch (lokale Maschine) und analysieren sie einzeln.
Lazy Evaluation (Snowpark DataFrames): Wir sagen der Bibliothekarin (Snowpark), welche Bücher wir brauchen und welche Analysen wir wollen, und die Bibliothekarin beschafft und analysiert die relevanten Informationen effizient.

Wann du Snowpark DataFrames verwenden solltest

Da Snowpark in Pandas DataFrames konvertieren kann: Wann nutzt man was?

pandas_diamonds = diamonds_df.to_pandas()
pandas_diamonds.head()Bereitgestellt von

Die Antwort hängt stark von der Datengröße ab.

Ist dein Dataset klein (wie das Diamonds-Dataset), kannst du es problemlos lokal herunterladen und mit Pandas verarbeiten (genau das macht DataFrame.to_pandas()).

Bei heutigen Datengrößen gerät das aber schnell aus dem Ruder, und du wartest im Zweifel Stunden auf den Download. Und denk daran: Es ist ein Hin- und Rückweg—neue Ergebnisse, die du behalten willst, müssen wieder zurück in die Datenbank. Genau deshalb lohnt es sich, die Snowpark-DataFrame-API zu beherrschen.

Alternativ erlaubt dir die Methode Session.sql(), jede mit Snowflake kompatible SQL-Anweisung auszuführen (siehe Beispiel unten). Ich empfehle dir aber klar, die Snowpark-DataFrame-API zu meistern—sie bringt für über 100 SQL-Funktionen eigene Helfer mit.

result = new_session.sql(
   """
   SELECT PRICE, CUT FROM DIAMONDS LIMIT 10
"""
)

type(result)
Bereitgestellt von

snowflake.snowpark.dataframe.DataFrameBereitgestellt von

result.show()Bereitgestellt von

Transformationsfunktionen für Snowpark DataFrames

Du wirst viel Zeit mit den Transformationsfunktionen von Snowpark verbringen. Sie sind im folgenden Submodul verfügbar:

# Die gängige Abkürzung ist F
import snowflake.snowpark.functions as FBereitgestellt von

Alle Funktionen unter F (auflistbar mit dir(F)) beschreiben, wie ein DataFrame transformiert werden soll. Ein Beispiel:

F.upper(F.col("CUT"))Bereitgestellt von

Hier erzeugt die Funktion col eine Referenz auf die angegebene Spalte und gibt ein Column-Objekt zurück. Danach wandelt .upper() die Werte der Spalte CUT in Großbuchstaben um:

diamonds_df.show(5)Bereitgestellt von

Schauen wir ins Dataset, sehen wir, dass die Werte in CUT immer noch Kleinbuchstaben enthalten. Warum? Nun, der Ausdruck F.upper(F.col("CUT")) entspricht folgender SQL-Abfrage:

SELECT UPPER(CUT);Bereitgestellt von

Was fehlt?

Klar, die FROM-Klausel, die die Tabelle angibt, auf die die Transformation angewandt werden soll!

Wir müssen den Ausdruck F.upper(F.col("CUT")) also mit dem Objekt diamonds_df kombinieren. Das geht im Wesentlichen auf drei Arten:

Für Ausdrücke, die Spalten transformieren oder auswählen: in .select() übergeben
Für Ausdrücke, die Zeilen auswählen: in .filter() übergeben
Für Ausdrücke, die neue Spalten erstellen: in .with_column() übergeben

Da unser Ausdruck eine Spalte transformiert, übergeben wir ihn an .select():

our_expression = F.upper(F.col("CUT"))
diamonds_df.select(our_expression).show()Bereitgestellt von

Die Werte sind erfolgreich geändert!

Schreiben wir nun einen Ausdruck, der Zeilen filtert:

diamonds_df.filter(F.col("PRICE") > 10000).count()Bereitgestellt von

5222Bereitgestellt von

Es gibt also über 5.000 Diamanten im Dataset, die mehr als 10.000 $ kosten. Erstellen wir auch noch eine neue Spalte:

(
diamonds_df.with_column("PRICE_SQUARED", F.col("PRICE") ** 2)
.select("PRICE", "PRICE_SQUARED")
.show(5)
)Bereitgestellt von

Oben siehst du, dass auch Method Chaining möglich ist—wir haben eine neue Spalte namens „PRICE_SQUARED“. Du merkst: Die Funktionen der Snowpark DataFrames sind denen von Pandas sehr ähnlich.

Kurzum: Suchst du das Snowpark-Pendant zu einer Pandas-Funktion, findest du es entweder unter F oder als Methode der Snowpark DataFrames. Du kannst das mit dir(F) oder dir(diamonds_df) prüfen oder im Snowpark-Python-Referenzhandbuch nachlesen.

EDA mit Snowpark DataFrames

Ein großer Teil der explorativen Datenanalyse (EDA) sind Visualisierungen. Zum Glück brauchen wir dafür Snowpark nicht—wir können bei unserem modernen Python-Stack bleiben.

Die Idee: Da EDA vor allem allgemeine Trends und Einsichten aufzeigt, reicht oft eine Stichprobe statt des gesamten Datasets. Eine repräsentative Stichprobe können wir herunterladen, in einen Pandas DataFrame konvertieren und mit Matplotlib oder Seaborn visualisieren.

Starten wir mit der Konvertierung:

sample = diamonds_df.sample(0.25).to_pandas()
sample.head()Bereitgestellt von

Oben laden wir 25% des Datasets als Pandas DataFrame herunter. Konvertieren wir die Spaltennamen in Kleinbuchstaben:

sample.columns = [col.lower() for col in sample.columns]
sample.columnsBereitgestellt von

Index(['carat', 'cut', 'color', 'clarity', 'depth', 'table', 'price', 'x', 'y', 'z'], dtype='object')Bereitgestellt von

Jetzt kannst du EDA wie gewohnt durchführen. Unten einige Ideen—beginnen wir mit einem Scatterplot für den Zusammenhang zwischen Preis und Karat:

#Remember we imported seaborn as sns
sns.scatterplot(data=sample, x="price", y="carat", s=1)
plt.title("Diamond prices vs. carat")
plt.show()Bereitgestellt von

Als Nächstes zeichnen wir eine Korrelationsmatrix, um die Zusammenhänge zwischen allen numerischen Merkmalen zu sehen:

corr_matrix = sample.corr(numeric_only=True)
sns.heatmap(corr_matrix, center=0, square=True, annot=True
plt.title("The correlation matrix of Diamond numeric features")
plt.show()Bereitgestellt von

Wir können auch ein Balkendiagramm der Diamant-Schliffkategorien erstellen:

cut_counts = sample.groupby("cut")["cut"].count()
cut_counts.plot(kind="bar")
plt.title("Category counts of diamond cuts")
plt.xlabel("Diamond cut")
plt.show()Bereitgestellt von

Um zu bestätigen, dass die Plots auf dem Gesamtdatensatz ähnlich aussehen, führen wir die SQL-Variante des obigen Balkendiagramms aus.

Zuerst schreiben wir eine SQL-Abfrage, die das Diamonds-Dataset nach CUT gruppiert:

result = new_session.sql(
   """
   SELECT cut, COUNT(*) AS count
     FROM diamonds
    GROUP BY cut;
"""
)
Bereitgestellt von

Konvertieren und laden wir das Ergebnis als Pandas DataFrame herunter:

result_pd = result.to_pandas()
result_pdBereitgestellt von

Diesmal verwenden wir plt.bar(), um den Graphen zu zeichnen:

plt.bar(x=result_pd["CUT"], height=result_pd["COUNT"])
plt.show()Bereitgestellt von

Die Reihenfolge der Balken unterscheidet sich, aber nebeneinander betrachtet vermitteln beide Diagramme dieselbe Aussage.

Ein Machine-Learning-Modell in Snowpark trainieren

Jetzt lernen wir, wie man in Snowpark Machine-Learning-Modelle trainiert!

Wir beginnen mit der Datenbereinigung.

Daten in Snowpark bereinigen

Stellen wir zuerst sicher, dass alle Spalten den korrekten Datentyp haben:

list(diamonds_df.schema)Bereitgestellt von

[StructField('CARAT', DecimalType(38, 2), nullable=True),
StructField('CUT', StringType(16777216), nullable=True),
StructField('COLOR', StringType(16777216), nullable=True),
StructField('CLARITY', StringType(16777216), nullable=True),
StructField('DEPTH', DecimalType(38, 1), nullable=True),
StructField('"table"', DecimalType(38, 1), nullable=True),
StructField('PRICE', LongType(), nullable=True),
StructField('X', DecimalType(38, 2), nullable=True),
StructField('Y', DecimalType(38, 2), nullable=True),
StructField('Z', DecimalType(38, 2), nullable=True)]Bereitgestellt von

Die Spalte table hat einen unglücklichen Namen, daher benennen wir sie in TABLE_ um (mit Unterstrich am Ende, damit es nicht mit einem SQL-Schlüsselwort kollidiert):

diamonds_df = diamonds_df.with_column_renamed('"table"', "TABLE_")
diamonds_df.columnsBereitgestellt von

['CARAT', 'CUT', 'COLOR', 'CLARITY', 'DEPTH', 'TABLE_', 'PRICE', 'X', 'Y', 'Z']Bereitgestellt von

Wir müssen die Datentypen der numerischen Merkmale von DecimalType auf DoubleType umstellen, da Dezimalzahlen in Snowpark derzeit noch nicht unterstützt werden.

from snowflake.snowpark.types import DoubleType

numeric_features = ["CARAT", "X", "Y", "Z", "DEPTH", "TABLE_"]
for col in numeric_features:
   diamonds_df = diamonds_df.with_column(col, diamonds_df[col].cast(DoubleType()))

list(diamonds_df.select(*numeric_features))Bereitgestellt von

[Column("CARAT"),
Column("X"),
Column("Y"),
Column("Z"),
Column("DEPTH"),
Column("TABLE_")]Bereitgestellt von

Im Snippet oben verwenden wir erneut .with_column(). Neu ist .cast(), eine Methode der Snowpark DataFrames.

Snowpark verlangt außerdem, dass alle Textmerkmale vor dem Encoding in Großbuchstaben stehen und keine Leerzeichen enthalten. Aktuell verletzt das Merkmal CUT diese Anforderung. Wir beheben das mit Funktions-Transformationen:

import snowflake.snowpark.functions as F

def remove_space_and_upper(df):
   df = df.with_column("CUT", F.upper(F.replace(F.col("CUT"), " ", "_")))

   return df

diamonds_df = remove_space_and_upper(diamonds_df)
Bereitgestellt von

Unsere Daten sind nun bereinigt. Wir können sie als neue Tabelle in Snowflake speichern:

diamonds_df.write.mode("overwrite").save_as_table("diamonds_cleaned")Bereitgestellt von

Stelle den Modus overwrite ein, denn möglicherweise passt du die Bereinigung später an und speicherst erneut.

Daten für das Modeling in Snowpark vorbereiten

In diesem Abschnitt kümmern wir uns um verbleibende Vorverarbeitungsschritte, die das Training behindern könnten. Laden wir die bereinigte Tabelle:

clean_df = new_session.table("diamonds_cleaned")
clean_df.show()Bereitgestellt von

Sehr gut!

Als Nächstes importieren wir die Submodule preprocessing und pipeline aus dem Namespace snowflake.ml.

import snowflake.ml.modeling.preprocessing as snowml
from snowflake.ml.modeling.pipeline import PipelineBereitgestellt von

Während die DataFrame-API von Snowpark unter snowflake.snowpark liegt, findest du die Snowpark-ML-Module unter snowflake.ml. Das ist anfangs ungewohnt—nimm dir Zeit, dich daran zu gewöhnen.

Im folgenden Code:

Listen wir die kategorialen Merkmale auf, die encodiert werden müssen.
Definieren wir Ausgabespalten, da Snowpark encodierte Features als neue Spalten hinzufügt statt die alten zu ersetzen.
Geben wir die korrekte Ordnung ordinaler Kategorien an; im Diamonds-Dataset sind alle Kategorien geordnet—nach rechts werden sie teurer.

# List all the features for processing
cat_cols = ["CUT", "COLOR", "CLARITY"]
cat_cols_encoded = ["CUT_OE", "COLOR_OE", "CLARITY_OE"]

# We already have numeric_features

# List the correct ordering of categorical features
categories = {
   "CUT": np.array(["IDEAL", "PREMIUM", "VERY_GOOD", "GOOD", "FAIR"]),
   "CLARITY": np.array(
       ["IF", "VVS1", "VVS2", "VS1", "VS2", "SI1", "SI2", "I1", "I2", "I3"]
   ),
   "COLOR": np.array(["D", "E", "F", "G", "H", "I", "J"]),
}Bereitgestellt von

Jetzt bauen wir eine Pipeline, ähnlich zu Scikit-learn:

# Build the pipeline
preprocessing_pipeline = Pipeline(
   steps=[
       (
           "OE",
           snowml.OrdinalEncoder(
               input_cols=cat_cols, output_cols=cat_cols_encoded, categories=categories
           ),
       ),
       (
           "SS",
           snowml.StandardScaler(
               input_cols=numeric_features, output_cols=numeric_features
           ),
       ),
   ]
)
Bereitgestellt von

Die Klasse Pipeline akzeptiert eine Liste von Schritten, jeweils eine Preprocessing-Klasse. Hier encodieren wir mit OrdinalEncoder kategoriale Features als 0, 1, 2, … und normalisieren mit StandardScaler die numerischen Features.

Zum Schluss speichern wir die Pipeline lokal mit joblib und testen sie auf dem gesamten Dataset:

import joblib
PIPELINE_FILE = "pipeline.joblib"

# Pickle locally first
joblib.dump(preprocessing_pipeline, PIPELINE_FILE)

transformed_diamonds_df = preprocessing_pipeline.fit(clean_df).transform(clean_df)

# transformed_diamonds_df.show() - commented because of long outputBereitgestellt von

Der Code läuft fehlerfrei—unsere Pipeline funktioniert. Zeit, ein Modell zu trainieren.

Ein XGBoost-Modell in Snowpark trainieren

Zum Trainieren listen wir noch einmal die Feature- und Zielspalten auf:

# Define the columns again
cat_cols_encoded = ["CUT_OE", "COLOR_OE", "CLARITY_OE"]
numeric_features = ["CARAT", "X", "Y", "Z", "DEPTH", "TABLE_"]

label_cols = ["PRICE"]  # Must be a list
output_cols = ["PRICE_PREDICTED"]  # Required in snowparkBereitgestellt von

Wir splitten die Daten mit DataFrame.random_split() und geben Gewichte für Trainings- und Testanteil an:

# Split the data
train_df, test_df = diamonds_df.random_split(weights=[0.8, 0.2], seed=42)Bereitgestellt von

Dann laden wir die Pipeline und führen Fit-Transform auf Train und nur Transform auf Test aus.

# Load the pre-processing pipeline locally
pipeline = joblib.load("pipeline.joblib")

# Apply it to both dataframes
train_df_transformed = pipeline.fit(train_df).transform(train_df)
test_df_transformed = pipeline.transform(test_df)Bereitgestellt von

Achte darauf, beim Testset nur .transform() aufzurufen, um Data Leakage zu vermeiden.

Als Nächstes initialisieren wir einen XGBoost-Regressor aus dem ml.modeling.xgboost-Submodul:

# Snowpark has models from scikit-learn and lightgbm too
from snowflake.ml.modeling.xgboost import XGBRegressor

# Initialize
regressor = XGBRegressor(
   input_cols=cat_cols_encoded + numeric_features,
   label_cols=label_cols,
   output_cols=output_cols,
)Bereitgestellt von

Die Klasse XGBRegressor benötigt alle Eingabespalten, die Zielspalten und die Namen der Ausgabespalten. Danach starten wir mit .fit() das Training:

# Train
regressor.fit(train_df_transformed)Bereitgestellt von

Das Training kann auf einem kostenlosen Snowflake-Account ohne GPU etwas dauern.

Nach dem Training generieren wir Vorhersagen:

# Predict

train_preds = regressor.predict(train_df_transformed)
test_preds = regressor.predict(test_df_transformed)Bereitgestellt von

Werfen wir einen Blick auf die Vorhersagen:

train_preds.select("PRICE", "PRICE_PREDICTED").show(5)Bereitgestellt von

Jetzt messen wir die Qualität unseres ersten Modells. Snowpark bringt dutzende Metriken aus Scikit-learn unter dem Submodul ml.modeling.metrics mit, das wir als M importieren:

import snowflake.ml.modeling.metrics as MBereitgestellt von

Dann berechnen wir zweimal mean_squared_error, um die Root Mean Squared Error (RMSE) für Train und Test zu messen (squared=False liefert RMSE statt MSE):

rmse_train = M.mean_squared_error(
   df=train_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

rmse_test = M.mean_squared_error(
   df=test_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

print(f"Train RMSE score for XGBRegressor: {rmse_train:.4f}")
print(f"Test RMSE score for XGBRegressor: {rmse_test:.4f}")Bereitgestellt von

Train RMSE score for XGBRegressor: 371.9664
Test RMSE score for XGBRegressor: 542.1566Bereitgestellt von

Unser Modell liegt im Schnitt um 542 $ daneben und überfittet möglicherweise, denn die Differenz zwischen Train- und Test-RMSE ist groß.

Lass uns die Hyperparameter optimieren, um das zu adressieren.

Hyperparameter-Tuning in Snowpark

Aktuell bietet Snowpark zwei Klassen für Hyperparameter-Tuning:

GridSearchCV: Exhaustive Search aller Hyperparameter-Kombinationen mit Cross-Validation.
RandomizedSearchCV: Zufällige Suche über definierte Verteilungen mit Cross-Validation.

XGBoost hat rund ein Dutzend Hyperparameter, die die Performance verbessern können (mehr dazu im Tutorial XGBoost in Python). Daher bietet sich eine Randomized Search an. Persönlich würde ich Optuna für Bayes’sche Optimierung bevorzugen—das steht in Snowpark derzeit noch nicht zur Verfügung.

Definieren wir nun die Suche:

from snowflake.ml.modeling.model_selection import RandomizedSearchCV

rscv = RandomizedSearchCV(
   estimator=XGBRegressor(),
   param_distributions={
       "n_estimators": [2000],
       "max_depth": list(range(3, 13)),
       "learning_rate": np.linspace(0.1, 0.5, num=10),
   },
   n_jobs=-1,
   scoring="neg_mean_squared_error",
   input_cols=numeric_features + cat_cols_encoded,
   label_cols=label_cols,
   output_cols=output_cols,
   n_iter=10,
)

rscv.fit(train_df_transformed)
Bereitgestellt von

Wir fixieren die Anzahl der Bäume auf 2000 und tunen nur max_depth und learning_rate. Der Standard für n_iter ist 10, d. h. es werden 10 zufällige Kombinationen getestet. Für genauere Ergebnisse sollte dieser Wert höher sein.

Messen wir nun die Performance des besten Modells. Der Code ist derselbe wie zuvor, nur verwenden wir statt regressor das Objekt rscv:

# Predict
train_preds = rscv.predict(train_df_transformed)
test_preds = rscv.predict(test_df_transformed)
rmse_train = M.mean_squared_error(
   df=train_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

optimal_rmse_test = M.mean_squared_error(
   df=test_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

print(f"Train RMSE score for optimal model: {rmse_train:.4f}")
print(f"Test RMSE score for optimal model: {rmse_test:.4f}")Bereitgestellt von

Train RMSE score for optimal model: 224.2405
Test RMSE score for optimal model: 572.0517Bereitgestellt von

Der Trainings-RMSE ist gesunken, der Test-RMSE jedoch gestiegen. Wir überfitten also weiterhin und sollten die Suche ausweiten und zusätzliche Parameter einbeziehen, um die Entscheidungsbäume von XGBoost zu beschneiden.

Diesen Teil übergebe ich an dich.

Das beste Modell in Snowpark speichern

Wenn wir die Session jetzt schließen, geht unser getuntes Modell verloren. Zum Speichern nutzen wir das native Model Registry von Snowpark—einen virtuellen Speicher für Modelle samt Metadaten.

Das Registry ist als Klasse Registry verfügbar und benötigt den Namen der aktuellen Datenbank und des Schemas der Session. Außerdem braucht es einen Projektnamen, damit es keine Konflikte mit Modellen anderer Projekte gibt:

# Set up for Registry
from snowflake.ml.registry import Registry

# Get the current db and schema name
db_name = new_session.get_current_database()
schema_name = new_session.get_current_schema()

# Define global model name for the project
model_name = "diamond_prices_regression"

# Initialize a registry to log models
registry = Registry(session=new_session, database_name=db_name, schema_name=schema_name)Bereitgestellt von

Zum Speichern verwenden wir die Methode Registry.log_model():

# Get sample data to pass into registry for schema
sample = train_df.select(cat_cols_encoded + numeric_features).limit(50)

# Log the first model
v0 = registry.log_model(
   model_name=model_name,
   version_name="v0",
   model=regressor,
   sample_input_data=sample,
)Bereitgestellt von

Wir können mit .set_metric() eine Metrik für das Modell hinterlegen (beliebig viele). Zudem fügen wir einen Kommentar hinzu:

# Add the models RMSE score
v0.set_metric(metric_name="RMSE", value=rmse_test)

# Add a description
v0.comment = "The first model to predict diamond prices"Bereitgestellt von

Dasselbe machen wir für das beste Modell aus der Randomized Search. Beim Setzen der Metrik nicht vergessen, den Wert auf optimal_rmse_test zu setzen.

# Log the optimal model
v1 = registry.log_model(
   model_name=model_name,
   version_name="v1",
   model=regressor,
   sample_input_data=sample,
)

# Add the models RMSE score
v1.set_metric(metric_name="RMSE", value=optimal_rmse_test)

# Add a description
v1.comment = "Optimal model found with RandomizedSearchCV"Bereitgestellt von

Um zu prüfen, ob die Modelle gespeichert sind, rufe .show_models() auf:

# Confirm the models are added
registry.show_models()Bereitgestellt von

Mehr zur Verwaltung des Registrys findest du in dieser Dokumentation.

Inference in Snowpark

Für Inference wählst du üblicherweise das beste Modell aus deinem Registry. Das geht, indem du das Ergebnis von .show_models() filterst oder das Modell direkt über sein Version-Tag holst, wie unten:

# Doing inference with the optimal model
optimal_version = registry.get_model(model_name).version("v1")
results = optimal_version.run(test_df, function_name="predict")

results.columnsBereitgestellt von

Die Methode get_model() gibt ein generisches model-Objekt zurück, das sich von XGBRegressor unterscheidet. Deshalb musst du für Inference .run() mit dem Funktionsnamen aufrufen.

Fazit

In diesem Tutorial haben wir ein End-to-End-Vorgehen zum Trainieren von Machine-Learning-Modellen in Snowflake Snowpark behandelt. Stark gemacht!

Wir sind mit einem unbereinigten Roh-Dataset gestartet und bei einem getunten XGBoost-Regressor gelandet. Alles lief in Snowflake Cloud—wir mussten keine lokalen Ressourcen einsetzen.

Wir haben gesehen, wie effektiv solche In-Database-Workflows bei großen Datasets sind—und dass Snowflake Snowpark eines der besten Tools ist, um genau das zu ermöglichen.

Wenn du mehr über Snowpark oder Snowflake lernen willst, schau dir diese Ressourcen an:

Danke fürs Lesen!

Autor

Bexruz (Bex) Tuychiev

Ich bin Content-Creator im Bereich Data Science mit über zwei Jahren Erfahrung und zähle zu den größten Stimmen auf Medium. Ich schreibe gern ausführliche Artikel über KI und ML – mit einer Prise Sarkasmus, damit das Ganze nicht zu trocken wird. Bisher habe ich über 130 Artikel veröffentlicht und einen DataCamp-Kurs produziert, ein weiterer ist in Arbeit. Meine Inhalte wurden von über 5 Millionen Menschen gelesen, 20.000 davon folgen mir auf Medium und LinkedIn.

Themen

Big Data Künstliche Intelligenz