Snowpark ML Modeling : modèle de développement ML¶

Exemples¶

Examinez les exemples suivants pour vous faire une idée des similitudes entre l’API Snowpark ML Modeling et les bibliothèques de machine learning que vous connaissez peut-être.

import numpy as np
import pandas as pd
import random
import string

from sklearn.datasets import make_regression
from snowflake.ml.modeling.preprocessing import MinMaxScaler, OrdinalEncoder
from snowflake.ml.modeling.pipeline import Pipeline
from snowflake.snowpark import Session

# Create a session with your preferred method
# session =

NUMERICAL_COLS = ["X1", "X2", "X3"]
CATEGORICAL_COLS = ["C1", "C2", "C3"]
FEATURE_COLS = NUMERICAL_COLS + CATEGORICAL_COLS
CATEGORICAL_OUTPUT_COLS = ["C1_OUT", "C2_OUT", "C3_OUT"]
FEATURE_OUTPUT_COLS = ["X1_FEAT_OUT", "X2_FEAT_OUT", "X3_FEAT_OUT", "C1_FEAT_OUT", "C2_FEAT_OUT", "C3_FEAT_OUT"]

# Create a dataset with numerical and categorical features
X, _ = make_regression(
    n_samples=1000,
    n_features=3,
    noise=0.1,
    random_state=0,
)
X = pd.DataFrame(X, columns=NUMERICAL_COLS)

def generate_random_string(length):
    return "".join(random.choices(string.ascii_uppercase, k=length))

categorical_feature_length = 2
categorical_features = {}
for c in CATEGORICAL_COLS:
    categorical_column = [generate_random_string(categorical_feature_length) for _ in range(X.shape[0])]
    categorical_features[c] = categorical_column

X = X.assign(**categorical_features)

features_df = session.create_dataframe(X)

# Fit a pipeline with OrdinalEncoder and MinMaxScaler on Snowflake
pipeline = Pipeline(
    steps=[
        (
            "OE",
            OrdinalEncoder(
                input_cols=CATEGORICAL_COLS,
                output_cols=CATEGORICAL_OUTPUT_COLS,
            )
        ),
        (
            "MMS",
            MinMaxScaler(
                input_cols=NUMERICAL_COLS + CATEGORICAL_OUTPUT_COLS,
                output_cols=FEATURE_OUTPUT_COLS,
            )
        ),
    ]
)

pipeline.fit(features_df)

# Use the pipeline to transform a dataset.
result = pipeline.transform(features_df)

import pandas as pd
from sklearn.datasets import make_classification

from snowflake.ml.modeling.xgboost import XGBClassifier
from snowflake.ml.utils.connection_params import SnowflakeLoginOptions
from snowflake.snowpark import Session

# Create a session with your preferred method
# session =

FEATURE_COLS = ["X1", "X2", "X3", "X4", "X5", "X6"]
LABEL_COLS = ["Y"]
OUTPUT_COLS = ["PREDICTIONS"]

# Set up data.
X, y = make_classification(
    n_samples=40000,
    n_features=6,
    n_informative=4,
    n_redundant=1,
    random_state=0,
    shuffle=True,
)

X = pd.DataFrame(X, columns=FEATURE_COLS)
y = pd.DataFrame(y, columns=LABEL_COLS)

features_pandas = pd.concat([X, y], axis=1)
features_df = session.create_dataframe(features_pandas)

# Train an XGBoost model on snowflake.
xgboost_model = XGBClassifier(
    input_cols=FEATURE_COLS,
    label_cols=LABEL_COLS,
    output_cols=OUTPUT_COLS
)

xgboost_model.fit(features_df)

# Use the model to make predictions.
predictions = xgboost_model.predict(features_df)
predictions[OUTPUT_COLS].show()

DATA_FILE_PATH = "~/Downloads/magic+gamma+telescope/magic04.data"

# Setup
from snowflake.ml.utils.connection_params import SnowflakeLoginOptions
from snowflake.snowpark import Session
import posixpath
import os

##
# Note: Create session https://docs.snowflake.com/en/developer-guide/snowpark/reference/python/latest/api/snowflake.snowpark.Session
##
session = Session.builder.configs(SnowflakeLoginOptions()).create()

session.sql("""
CREATE OR REPLACE TABLE Gamma_Telescope_Data(
    F_LENGTH FLOAT,
    F_WIDTH FLOAT,
    F_SIZE FLOAT,
    F_CONC FLOAT,
    F_CONC1 FLOAT,
    F_ASYM FLOAT,
    F_M3_LONG FLOAT,
    F_M3_TRANS FLOAT,
    F_ALPHA FLOAT,
    F_DIST FLOAT,
    CLASS VARCHAR(10))
""").collect()
session.sql("CREATE OR REPLACE STAGE SNOWPARK_ML_TEST_DATA_STAGE").collect()
session.file.put(
    DATA_FILE_PATH,
    "SNOWPARK_ML_TEST_DATA_STAGE/magic04.data",
    auto_compress=False,
    overwrite=True,
)

session.sql("""
COPY INTO Gamma_Telescope_Data FROM @SNOWPARK_ML_TEST_DATA_STAGE/magic04.data
FILE_FORMAT = (TYPE = 'CSV' field_optionally_enclosed_by='"',SKIP_HEADER = 0);
""").collect()

session.sql("select * from Gamma_Telescope_Data limit 5").collect()

from snowflake.ml.utils.connection_params import SnowflakeLoginOptions
from snowflake.snowpark import Session, DataFrame

from snowflake.ml.modeling.preprocessing import StandardScaler
from snowflake.ml.modeling.impute import SimpleImputer
from snowflake.ml.modeling.pipeline import Pipeline
from snowflake.ml.modeling.xgboost import XGBClassifier

from snowflake.ml.modeling.metrics import accuracy_score

##
# Note: Create session https://docs.snowflake.com/en/developer-guide/snowpark/reference/python/latest/api/snowflake.snowpark.Session
##
session = Session.builder.configs(SnowflakeLoginOptions()).create()

# Step 1: Create train and test dataframes
all_data = session.sql("select *, IFF(CLASS = 'g', 1.0, 0.0) as LABEL from Gamma_Telescope_Data").drop("CLASS")
train_data, test_data = all_data.random_split(weights=[0.9, 0.1], seed=0)

# Step 2: Construct training pipeline with preprocessing and modeling steps
FEATURE_COLS = [c for c in train_data.columns if c != "LABEL"]
LABEL_COLS = ["LABEL"]

pipeline = Pipeline(steps = [
    ("impute", SimpleImputer(input_cols=FEATURE_COLS, output_cols=FEATURE_COLS)),
    ("scaler", StandardScaler(input_cols=FEATURE_COLS, output_cols=FEATURE_COLS)),
    ("model", XGBClassifier(input_cols=FEATURE_COLS, label_cols=LABEL_COLS))
])

# Step 3: Train
pipeline.fit(train_data)

# Step 4: Eval
predict_on_training_data = pipeline.predict(train_data)
training_accuracy = accuracy_score(df=predict_on_training_data, y_true_col_names=["LABEL"], y_pred_col_names=["OUTPUT_LABEL"])

predict_on_test_data = pipeline.predict(test_data)
eval_accuracy = accuracy_score(df=predict_on_test_data, y_true_col_names=["LABEL"], y_pred_col_names=["OUTPUT_LABEL"])

print(f"Training accuracy: {training_accuracy} \nEval accuracy: {eval_accuracy}")

from snowflake.snowpark import Session, DataFrame
from snowflake.ml.utils.connection_params import SnowflakeLoginOptions

from sklearn.datasets import make_classification
from snowflake.snowpark import Session, DataFrame
from snowflake.ml.modeling.xgboost import XGBClassifier
from snowflake.ml.modeling.model_selection.grid_search_cv import GridSearchCV

FEATURE_COLS = ["X1", "X2", "X3", "X4", "X5", "X6"]
LABEL_COLS = ["Y"]
OUTPUT_COLS = ["PREDICTIONS"]

# Create a session using your favorite login option.
# In this example we use a session builder with `SnowflakeLoginOptions`.
session = Session.builder.configs(SnowflakeLoginOptions()).create()

# Set up data.
def set_up_data(session: Session, n_samples: int) -> DataFrame:
    X, y = make_classification(
        n_samples=n_samples,
        n_features=6,
        n_informative=2,
        n_redundant=0,
        random_state=0,
        shuffle=True,
    )

    X = pd.DataFrame(X, columns=FEATURE_COLS)
    y = pd.DataFrame(y, columns=LABEL_COLS)

    features_pandas = pd.concat([X, y], axis=1)
    features_pandas.head()

    features_df = session.create_dataframe(features_pandas)
    return features_df

features_df = set_up_data(session, 10**4)

# Create a warehouse to use for the tuning job.
session.sql(
    """
CREATE or replace warehouse HYPERPARAM_WH
    WITH WAREHOUSE_SIZE = 'X-SMALL'
    WAREHOUSE_TYPE = 'Standard'
    AUTO_SUSPEND = 60
    AUTO_RESUME = TRUE
    INITIALLY_SUSPENDED = FALSE;"""
).collect()
session.use_warehouse("HYPERPARAM_WH")

# Tune an XGB Classifier model using sklearn GridSearchCV.
DISTRIBUTIONS = dict(
    n_estimators=[10, 50],
    learning_rate=[0.01, 0.1, 0.2],
)
estimator = XGBClassifier()
grid_search_cv = GridSearchCV(estimator=estimator, param_grid=DISTRIBUTIONS, input_cols=FEATURE_COLS, label_cols=LABEL_COLS, output_cols=OUTPUT_COLS)

grid_search_cv.fit(features_df)

# Use the best model to make predictions.
predictions = grid_search_cv.predict(features_df)
predictions[OUTPUT_COLS].show()

# Retrieve sklearn model, and print the best score
sklearn_grid_search_cv = grid_search_cv.to_sklearn()
print(sklearn_grid_search_cv.best_score_)

large_features_df = set_up_data(session, 10**6)

# Scale up the warehouse for a faster fit. This takes 2m15s to run on an L warehouse versus 4m5s on a XS warehouse.
session.sql(f"ALTER WAREHOUSE {session.get_current_warehouse()} SET WAREHOUSE_SIZE='LARGE'").collect()

grid_search_cv.fit(large_features_df)
print(grid_search_cv.to_sklearn().best_score_)

Classes Snowpark ML Modeling¶

Toutes les classes Snowpark ML Modeling et les classes de prétraitement se trouvent dans l’espace de noms snowflake.ml.modeling. Les modules Snowpark ML portent le même nom que les modules correspondants de l’espace de noms sklearn. Par exemple, le module Snowpark ML correspondant à sklearn.calibration est snowflake.ml.modeling.calibration. Les modules xgboost et lightgbm correspondent respectivement à snowflake.ml.modeling.xgboost et snowflake.ml.modeling.lightgbm.

L’API Snowpark ML Modeling fournit des wrappers pour les classes scikit-learn, xgboost et lightgbm sous-jacentes, la majorité d’entre elles étant exécutées sous forme de procédures stockées (sur un seul nœud d’entrepôt) dans l’entrepôt virtuel. Toutes les classes de scikit-learn ne sont pas prises en charge dans Snowpark ML. Voir la Référence API Snowpark ML pour une liste des classes actuellement disponibles.

Certaines classes (notamment les classes métriques et de prétraitement) prennent en charge l’exécution distribuée et peuvent offrir des avantages significatifs en termes de performances par rapport à l’exécution locale des mêmes opérations. Pour plus d’informations, voir Prétraitement distribué et Optimisation distribuée des hyperparamètres. Le tableau ci-dessous répertorie les classes spécifiques qui prennent en charge l’exécution distribuée.

Différences générales entre les API¶

Snowpark ML Modeling comprend des algorithmes de prétraitement, de transformation et de prédiction des données basés sur scikit-learn, xgboost et lightgbm. Les classes Python de Snowpark remplacent les classes correspondantes des paquets d’origine, avec des signatures similaires. Cependant, ces APIs sont conçues pour fonctionner avec des DataFrames Snowpark plutôt qu’avec des tableaux NumPy.

Bien que l’API de Snowpark ML Modeling soit similaire à celle de scikit-learn, il existe quelques différences clés. Cette section explique comment appeler les méthodes __init__ (constructeur), fit, et predict pour les classes d’estimateurs et de transformateurs fournies dans Snowpark ML.

• Le constructeur de toutes les classes Python de Snowpark ML accepte cinq paramètres supplémentaires (input_cols, output_cols, sample_weight_col, label_cols, et drop_input_cols) en plus des paramètres acceptés par les classes équivalentes dans scikit-learn, xgboost ou lightgbm. Il s’agit de chaînes ou de séquences de chaînes qui spécifient les noms des colonnes d’entrée, des colonnes de sortie, de la colonne de poids de l’échantillon et des colonnes de balises dans un DataFrame Snowpark ou Pandas.

• Les méthodes fit et predict de Snowpark ML acceptent un seul DataFrame plutôt que des tableaux séparés représentant les données d’entrée, les balises et les poids. Avec Snowpark ML, vous spécifiez les noms des colonnes à utiliser à ces fins lorsque vous instanciez la classe ; ces noms sont ensuite utilisés pour trouver les colonnes requises dans le DataFrame que vous transmettez à fit ou predict. Consultez fit et predict.

• Les méthodes transform et predict de Snowpark ML renvoient un DataFrame contenant toutes les colonnes du DataFrame transmis à la méthode, avec le résultat de la prédiction stocké dans des colonnes supplémentaires. (Vous pouvez effectuer une transformation sur place en spécifiant les noms de colonne d’entrée correspondant aux noms de colonne de sortie, ou supprimer les colonnes d’entrée en transmettant drop_input_cols = True.) Les équivalents scikit-learn, xgboost et lightgbm renvoient des tableaux contenant uniquement les résultats.

• Les transformateurs Snowpark Python n’ont pas de méthode fit_transform. Cependant, comme avec scikit-learn, la validation des paramètres n’est effectuée que dans la méthode fit . Vous devez donc appeler fit à un moment donné avant transform, même si le transformateur n’effectue aucun ajustement. fit renvoie le transformateur, de sorte que les appels de méthode peuvent être chaînés, par exemple Binarizer(threshold=0.5).fit(df).transform(df).

• Les transformateurs Snowpark ML n’ont pas de méthode inverse_transform. Cette méthode n’est pas nécessaire avec Snowpark ML, car la représentation d’origine reste disponible dans les colonnes d’entrée du DataFrame d’entrée, qui sont préservées à moins que vous n’effectuiez explicitement une transformation sur place en spécifiant les mêmes noms pour les colonnes d’entrée et de sortie.

Vous pouvez convertir n’importe quel objet de modélisation Snowpark ML en objet scikit-learn, xgboost ou lightgbm correspondant, ce qui vous permet d’utiliser l’ensemble des méthodes et attributs du type sous-jacent. Voir Récupération du modèle sous-jacent.

from snowflake.ml.modeling.preprocessing import MinMaxScaler
from snowflake.snowpark import Session

# Snowflake identifiers are not case sensitive by default.
# These column names will be automatically updated to ["COLUMN_1", "COLUMN_2", "COLUMN_3"] by the Snowpark DataFrame.
schema = ["column_1", "column_2", "column_3"]
df = session.create_dataframe([[1, 2, 3]], schema = schema)
df.show()

--------------------------------------
|"COLUMN_1"  |"COLUMN_2"  |"COLUMN_3"|
--------------------------------------
|1           |2          |3          |
--------------------------------------

# Identify the column names using the Snowflake identifier.
input_cols = ["COLUMN_1", "COLUMN_2", "COLUMN_3"]
mms = MinMaxScaler(input_cols=input_cols)
mms.fit(df)

# To maintain lower case column names, include a double quote within the string.
schema = ['"column_1"', '"column_2"', '"column_3"']
df = session.create_dataframe([[1, 2, 3]], schema = schema)
df.show()

----------------------------------------
|'"column_1"'|'"column_2"'|'"column_3"'|
----------------------------------------
|1           |2           |3           |
----------------------------------------

# Since no conversion took place, the schema labels can be used as the column identifiers.
mms = MinMaxScaler(input_cols=schema)
mms.fit(df)

from sklearn.tree import DecisionTreeClassifier

model = DecisionTreeClassifier()

model.fit(
    X=df[feature_column_names], y=df[label_column_names], sample_weight=df[weight_column_name]
)

model.fit(df)

prediction_results = model.predict(X=df[feature_column_names])

prediction_results = model.predict(df)[output_column_names]

Prétraitement distribué¶

De nombreuses fonctions de prétraitement et de transformation de données dans Snowpark ML sont mises en œuvre via le moteur d’exécution distribuée de Snowflake, qui apporte des avantages significatifs en termes de performances par rapport à l’exécution sur un seul nœud (c’est-à-dire, les procédures stockées). Pour savoir quelles fonctions prennent en charge l’exécution distribuée, voir Classes Snowpark ML Modeling.

Le graphique ci-dessous illustre les chiffres des performances de grands ensembles de données publics, exécutés dans un entrepôt moyen optimisé par Snowpark, en comparant scikit-learn exécuté dans des procédures stockées aux implémentations distribuées de Snowpark ML. Dans de nombreux scénarios, votre code peut s’exécuter 25 à 50 fois plus vite via Snowpark ML Modeling.

Optimisation distribuée des hyperparamètres¶

Le réglage des hyperparamètres fait partie intégrante du flux de travail de la science des données. L’API Snowpark ML fournit des implémentations distribuées des APIs GridSearchCV et RandomizedSearchCV scikit-learn pour permettre le réglage efficace des hyperparamètres dans des entrepôts à un ou plusieurs nœuds.

import snowflake.ml.modeling.parameters.disable_distributed_hpo

Pour les entrepôts à un seul nœud (XS), la capacité totale du nœud est utilisée par défaut à l’aide du cadre de multitraitement joblib de scikit-learn.

Pour les entrepôts à plusieurs nœuds, les opérations fit de votre tâche de réglage de validation croisée sont réparties entre les nœuds. Aucune modification de code n’est nécessaire pour monter en échelle. Les ajustements de l’estimateur sont exécutés en parallèle sur tous les cœurs disponibles sur tous les nœuds de l’entrepôt.

À titre d’illustration, considérons l”ensemble de données sur le logement en Californie fourni avec la bibliothèque scikit-learn. Il comprend 20 640 lignes de données fournissant les informations suivantes :

• MedInc : Revenu moyen du groupe de pâtés de maisons

• HouseAge : Âge moyen du foyer du groupe de pâtés de maisons

• AveRooms : Nombre moyen de pièces par foyer

• AveBedrms : Nombre moyen de chambres à coucher par foyer

• Population : Population du groupe de pâtés de maisons

• AveOccup : Nombre moyen de membres du foyer

• Latitude et Longitude

La cible de l’ensemble de données est le revenu moyen, exprimé en centaines de milliers de dollars.

Dans cet exemple, nous effectuons une validation croisée via une recherche par grille sur une régression de forêt aléatoire afin de trouver la meilleure combinaison d’hyperparamètres pour prédire le revenu moyen.

from snowflake.ml.modeling.ensemble.random_forest_regressor import RandomForestRegressor
from snowflake.ml.modeling.model_selection.grid_search_cv import GridSearchCV
from sklearn import datasets

def load_housing_data() -> DataFrame:
    input_df_pandas = datasets.fetch_california_housing(as_frame=True).frame
    # Set the columns to be upper case for consistency with Snowflake identifiers.
    input_df_pandas.columns = [c.upper() for c in input_df_pandas.columns]
    input_df = session.create_dataframe(input_df_pandas)

    return input_df

input_df = load_housing_data()

# Use all the columns besides the median value as the features
input_cols = [c for c in input_df.columns if not c.startswith("MEDHOUSEVAL")]
# Set the target median value as the only label columns
label_cols = [c for c in input_df.columns if c.startswith("MEDHOUSEVAL")]


DISTRIBUTIONS = dict(
            max_depth=[80, 90, 100, 110],
            min_samples_leaf=[1,3,10],
            min_samples_split=[1.0, 3,10],
            n_estimators=[100,200,400]
        )
estimator = RandomForestRegressor()
n_folds = 5

clf = GridSearchCV(estimator=estimator, param_grid=DISTRIBUTIONS, cv=n_folds, input_cols=input_cols, label_cols=label_col)
clf.fit(input_df)

Cet exemple s’exécute en un peu plus de 7 minutes sur un entrepôt moyen (Medium) (un seul nœud) optimisé par Snowpark, et ne prend que 3 minutes pour s’exécuter sur un entrepôt X-Large.

Déploiement et exécution de votre modèle¶

Le résultat de l’entraînement d’un modèle est un objet de modèle Python Snowpark ML. Vous pouvez utiliser le modèle entraîné pour faire des prédictions en appelant la méthode predict du modèle. Cela crée une fonction temporaire définie par l’utilisateur pour exécuter le modèle dans votre entrepôt virtuel Snowflake. Cette fonction est automatiquement supprimée à la fin de votre session Snowpark ML (par exemple, lorsque votre script se termine, ou lorsque vous fermez votre Jupyter Notebook).

Pour conserver la fonction définie par l’utilisateur après la fin de votre session, vous pouvez la créer manuellement. Consultez les guide de démarrage rapide sur le thème pour plus d’informations.

Le registre des modèles de Snowpark ML, une fonction à venir, prend également en charge les modèles persistants et facilite leur recherche et leur déploiement. Voir le Guide de démarrage rapide pour une introduction au registre de modèle ainsi que d’autres fonctions de Snowpark ML.

Pipeline pour les transformations multiples¶

Avec scikit-learn, il est courant d’exécuter une série de transformations en utilisant un pipeline. Les pipelines scikit-learn ne fonctionnent pas avec les classes Snowpark ML, c’est pourquoi Snowpark ML fournit une version Snowpark Python de sklearn.pipeline.Pipeline pour exécuter une série de transformations. Cette classe se trouve dans le paquet snowflake.ml.modeling.pipeline et fonctionne de la même manière que la version scikit-learn.

Récupération du modèle sous-jacent¶

Les modèles Snowpark ML peuvent être « décapsulés » (unwrapped), c’est-à-dire convertis en modèles tiers sous-jacents, à l’aide des méthodes suivantes (selon la bibliothèque) :

• to_sklearn

• to_xgboost

• to_lightgbm

L’ensemble des attributs et méthodes du modèle sous-jacent sont ensuite accessibles et peuvent être exécutés localement par rapport à l’estimateur. Par exemple, dans l”exemple GridSearchCV, nous convertissons l’estimateur de recherche par grille en un objet scikit-learn afin de récupérer le meilleur score.

best_score = grid_search_cv.to_sklearn().best_score_

Limites connues¶

• Les estimateurs et transformateurs de Snowpark ML ne prennent actuellement pas en charge les entrées ou les réponses clairsemées. Si vous disposez de données clairsemées, convertissez-les dans un format dense avant de les transmettre aux estimateurs ou transformateurs de Snowpark ML.

• Le paquet Snowpark ML ne prend actuellement pas en charge les types de données matricielles. Toute opération sur les estimateurs et les transformateurs qui produirait une matrice comme résultat échoue.

• Il n’est pas garanti que l’ordre des lignes dans les données de résultat corresponde à l’ordre des lignes dans les données d’entrée.

Dépannage¶

import logging, sys

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)

Autres lectures¶

Consultez la documentation des bibliothèques d’origine pour obtenir des informations complètes sur leurs fonctionnalités.

• Scikit-Learn

• XGBoost

• LightGBM

Reconnaissance¶

Certaines parties de ce document sont dérivées de la documentation Scikit-learn, qui est sous licence BSD-3 « New » ou « Revised » et sous Copyright © 2007-2023 The scikit-learn developers. Tous droits réservés.

Certaines parties de ce document sont dérivées de la documentation de XGboost, qui est couverte par la licence Apache 2.0, janvier 2004 et qui est sous Copyright © 2019. Tous droits réservés.

Certaines parties de ce document sont dérivées de la documentation LightGBM, qui est sous licence MIT et sous copyright © Microsoft Corp. Tous droits réservés.