Snowpark ML Modeling: desenvolvimento de modelo de ML¶

Exemplos¶

Revise os exemplos a seguir para ter uma ideia das semelhanças da API do Snowpark ML Modeling com as bibliotecas de aprendizado de máquina com as quais você talvez esteja familiarizado.

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

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

num_categorical_cols, categorical_feature_length = 3, 2
categorical_features = []
for _ in range(num_categorical_cols):
    categorical_column = [generate_random_string(categorical_feature_length) for _ in range(X.shape[0])]
    categorical_features.append(categorical_column)
X = np.column_stack((X, *categorical_features))
X = pd.DataFrame(X, columns=FEATURE_COLS)

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 do Snowpark ML Modeling¶

Todas as classes do Snowpark ML Modeling e de pré-processamento estão no namespace snowflake.ml.modeling. Os módulos do Snowpark ML têm o mesmo nome que os módulos correspondentes do namespace sklearn. Por exemplo, o módulo Snowpark ML correspondente a sklearn.calibration é snowflake.ml.modeling.calibration. O módulos xgboost e lightgbm correspondem a snowflake.ml.modeling.xgboost e snowflake.ml.modeling.lightgbm, respectivamente.

A API do Snowpark ML Modeling fornece wrappers para classes scikit-learn, xgboost e lightgbm subjacentes, a maioria dos quais são executados como procedimentos armazenados (executando em um único nó de warehouse) no warehouse virtual. Nem todas as classes do scikit-learn são suportadas no Snowpark ML. Consulte a Referência da API do Snowpark ML para obter uma lista das classes disponíveis atualmente.

Algumas classes (incluindo classes de pré-processamento e métricas) suportam a execução distribuída e podem fornecer benefícios significativos de desempenho em comparação com a execução local das mesmas operações. Para obter mais informações, consulte Pré-processamento distribuído e Otimização de hiperparâmetros distribuídos. A tabela abaixo lista as classes específicas que oferecem suporte à execução distribuída.

Diferenças gerais da API¶

O Snowpark ML Modeling inclui pré-processamento de dados, transformação e algoritmos de previsão baseados em scikit-learn, xgboost e lightgbm. As classes Snowpark Python são substituições para as classes correspondentes dos pacotes originais, com assinaturas semelhantes. No entanto, estas APIs são projetadas para funcionar com o Snowpark DataFrames em vez das matrizes NumPy.

Embora a API do Snowpark ML Modeling seja semelhante ao scikit-learn, existem algumas diferenças importantes. Esta seção explica como chamar os métodos __init__ (construtor), fit e predict para as classes estimador e transformador fornecidas no Snowpark ML.

• O construtor de todas as classes do Snowpark ML Python aceitam cinco parâmetros adicionais — input_cols, output_cols, sample_weight_col, label_cols e drop_input_cols — além dos parâmetros aceitos pelas classes equivalentes em scikit-learn, xgboost ou lightgbm. São cadeias de caracteres, ou sequências de cadeias de caracteres, que especificam os nomes das colunas de entrada, colunas de saída, coluna de peso de amostra e colunas de rótulo em um DataFrame do Snowpark ou Pandas.

• Os métodos fit e predict no Snowpark ML aceitam um único DataFrame em vez de matrizes separadas que representam dados de entrada, rótulos e pesos. Com o Snowpark ML, você especifica os nomes das colunas a serem usadas para essas finalidades ao instanciar a classe; esses nomes são usados para encontrar as colunas necessárias no DataFrame que você passa para fit ou predict. Consulte fit e predict.

• Os métodos transform e predict no Snowpark ML retornam um DataFrame contendo todas as colunas do DataFrame passado para o método, com a saída da previsão armazenada em colunas adicionais. (Você pode transformar no local especificando os nomes das colunas de entrada que correspondem aos nomes das colunas de saída ou descartar as colunas de entrada passando drop_input_cols = True.) Os equivalentes scikit-learn, xgboost e lightgbm retornam matrizes contendo apenas os resultados.

• Os transformadores Snowpark Python não têm um método fit_transform. No entanto, como no scikit-learn, a validação de parâmetros é realizada apenas no método fit, então você deve chamar fit em algum momento antes de transform, mesmo quando o transformador não tem nenhum encaixe. fit retorna o transformador, então as chamadas de método podem ser encadeadas; por exemplo, Binarizer(threshold=0.5).fit(df).transform(df).

• Os transformadores Snowpark ML não têm um método inverse_transform. Este método é desnecessário com o Snowpark ML porque a representação original permanece disponível nas colunas de entrada do DataFrame de entrada, que são preservadas, a menos que você execute explicitamente uma transformação no local especificando os mesmos nomes para as colunas de entrada e saída.

Você pode converter qualquer objeto de modelagem do Snowpark ML no objeto scikit-learn, xgboost ou lightgbm correspondente, permitindo usar todos os métodos e atributos do tipo subjacente. Consulte Recuperação do modelo subjacente.

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é-processamento distribuído¶

Muitas das funções de pré-processamento e transformação de dados no Snowpark ML são implementadas usando o mecanismo de execução distribuída do Snowflake, que oferece benefícios significativos de desempenho em comparação com a execução de nó único (ou seja, procedimentos armazenados). Para descobrir quais funções oferecem suporte à execução distribuída, consulte Classes do Snowpark ML Modeling.

O gráfico abaixo mostra números de desempenho ilustrativos em grandes conjuntos de dados públicos, executados em um warehouse médio otimizado para Snowpark, comparando o scikit-learn executado em procedimentos armazenados com as implementações distribuídas do Snowpark ML. Em muitos cenários, seu código pode ser executado de 25 a 50 vezes mais rápido ao usar Snowpark ML Modeling.

Otimização de hiperparâmetros distribuídos¶

O ajuste de hiperparâmetros é parte integrante do fluxo de trabalho da ciência de dados. A biblioteca Snowpark ML fornece implementações distribuídas do scikit-learn GridSearchCV e RandomizedSearchCV APIs para permitir o ajuste eficiente de hiperparâmetros em warehouses de nó único e de vários nós.

import snowflake.ml.modeling.parameters.disable_distributed_hpo

Para warehouses de nó único (XS), a capacidade total do nó é utilizada por padrão usando a estrutura de multiprocessamento joblib do scikit-learn.

Para warehouses com vários nós, as operações fit dentro do trabalho de ajuste de validação cruzada são distribuídas entre os nós. Nenhuma alteração de código é necessária para aumentar a escala. Os ajustes do estimador são executados em paralelo em todos os núcleos disponíveis em todos os nós do warehouse.

Como ilustração, considere o conjunto de dados habitacionais da Califórnia fornecido com a biblioteca scikit-learn. Os dados incluem 20.640 linhas de dados com as seguintes informações:

• MedInc: renda mediana no grupo de quarteirões

• HouseAge: idade média da residência no grupo de quarteirões

• AveRooms: número médio de cômodos por domicílio

• AveBedrms: número médio de quartos por domicílio

• População: a população do grupo de quarteirões

• AveOccup: número médio de membros da família

• Latitude e Longitude

O alvo do conjunto de dados é a renda média, expressa em centenas de milhares de dólares.

Neste exemplo, fazemos a validação cruzada de pesquisa em grade em um regressor aleatório para encontrar a melhor combinação de hiperparâmetros para prever a renda mediana.

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)

Este exemplo é executado em pouco mais de 7 minutos em um warehouse Medium (nó único) otimizado para Snowpark e leva apenas 3 minutos para ser executado em um warehouse X-Large.

Como implantar e executar seu modelo¶

O resultado do treinamento de um modelo é um objeto de modelo Python do Snowpark ML. Você pode usar o modelo treinado para fazer previsões chamando o método predict do modelo. Isso cria uma função temporária definida pelo usuário para executar o modelo em seu warehouse virtual Snowflake. Esta função é excluída automaticamente no fim de sua sessão do Snowpark ML (por exemplo, quando seu script termina ou quando você fecha seu Jupyter Notebook).

Para manter a função definida pelo usuário após o término da sessão, você pode criá-la manualmente. Consulte o Guia de início rápido sobre o tópico para obter mais informações.

O registro de modelo do Snowpark ML, um recurso futuro, também oferece suporte a modelos persistentes e facilita sua localização e implantação. Consulte o Guia de início rápido para obter uma introdução ao registro do modelo junto com outros recursos do Snowpark ML.

Pipeline para múltiplas transformações¶

Com o scikit-learn, é comum executar uma série de transformações usando um pipeline. Os pipelines do scikit-learn não funcionam com classes do Snowpark ML, portanto, o Snowpark ML fornece uma versão Snowpark Python de sklearn.pipeline.Pipeline para executar uma série de transformações. Esta classe está no pacote snowflake.ml.modeling.pipeline e funciona da mesma forma que a versão scikit-learn.

Recuperação do modelo subjacente¶

Os modelos do Snowpark ML podem ser “desencapsulados”, ou seja, convertidos em modelos subjacentes de terceiros, com os seguintes métodos (dependendo da biblioteca):

• to_sklearn

• to_xgboost

• to_lightgbm

Todos os atributos e métodos do modelo subjacente podem então ser acessados e executados localmente no estimador. Por exemplo, no exemplo GridSearchCV, convertemos o estimador de pesquisa em grade em um objeto scikit-learn para recuperar a melhor pontuação.

best_score = grid_search_cv.to_sklearn().best_score_

Limitações conhecidas¶

• Os estimadores e transformadores do Snowpark ML atualmente não oferecem suporte às entradas esparsas ou respostas esparsas. Se você tiver dados esparsos, converta-os em um formato denso antes de passá-los para os estimadores ou transformadores do Snowpark ML.

• O pacote Snowpark ML atualmente não oferece suporte a tipos de dados de matriz. Qualquer operação em estimadores e transformadores que produzam uma matriz como resultado falha.

• Não há garantia de que a ordem das linhas nos dados de resultado corresponda à ordem das linhas nos dados de entrada.

Solução de problemas¶

import logging, sys

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

Leitura adicional¶

Consulte a documentação das bibliotecas originais para obter informações completas sobre sua funcionalidade.

• Scikit-Learn

• XGBoost

• LightGBM

Aviso¶

Algumas partes deste documento são derivadas da documentação do Scikit-learn, que é licenciada sob a licença BSD-3 «Nova» ou «Revisada» e Copyright © 2007-2023 The scikit-learn developers. Todos os direitos reservados.

Algumas partes deste documento são derivadas da documentação XGboost, que é coberta pela licença Apache 2.0, janeiro de 2004 e Copyright © 2019. Todos os direitos reservados.

Algumas partes deste documento são derivadas da documentação LightGBM, que é licenciado para MIT e Copyright © Microsoft Corp. Todos os direitos reservados.