Model Training

Train machine learning models directly inside Snowflake using the Snowpark ML library, built-in ML functions, and popular framework integrations — with hyperparameter tuning and cross-validation strategies for the exam.

The snowflake-ml-python Library

The snowflake-ml-python package (also called Snowpark ML) is Snowflake's official ML library. It provides scikit-learn-compatible APIs that train models on Snowflake's compute, keeping data inside the platform.

Snowpark ML Modeling API

Snowpark ML provides wrappers for popular ML algorithms that accept Snowpark DataFrames directly and train on Snowflake's distributed compute.

from snowflake.ml.modeling.ensemble import RandomForestClassifier
from snowflake.ml.modeling.xgboost import XGBClassifier
from snowflake.ml.modeling.linear_model import LogisticRegression

# Define feature and label columns
FEATURE_COLS = ["SPEND_SCALED", "VISITS_SCALED", "DAYS_SINCE_LAST", "AGE"]
LABEL_COL = "CHURNED"

# Train a Random Forest using Snowpark ML
rf_model = RandomForestClassifier(
    input_cols=FEATURE_COLS,
    label_cols=[LABEL_COL],
    output_cols=["PREDICTED_CHURN"],
    n_estimators=100,
    max_depth=10,
    random_state=42
)

# fit() trains on Snowflake's compute
rf_model.fit(df_train)

# predict() returns a Snowpark DataFrame
df_predictions = rf_model.predict(df_test)

Supported Algorithms

Built-in ML Functions

Snowflake also provides SQL-level ML functions that require no Python code. These are useful for quick analyses and are tested on the exam.

Forecasting

-- Create a time-series forecast model
CREATE OR REPLACE SNOWFLAKE.ML.FORECAST sales_forecast(
    INPUT_DATA => SYSTEM$REFERENCE('TABLE', 'DAILY_SALES'),
    TIMESTAMP_COLNAME => 'SALE_DATE',
    TARGET_COLNAME => 'REVENUE',
    SERIES_COLNAME => 'PRODUCT_ID'
);

-- Generate forecasts
CALL sales_forecast!FORECAST(
    FORECASTING_PERIODS => 30,
    CONFIG_OBJECT => {'prediction_interval': 0.95}
);

Anomaly Detection

-- Create an anomaly detection model
CREATE OR REPLACE SNOWFLAKE.ML.ANOMALY_DETECTION transaction_monitor(
    INPUT_DATA => SYSTEM$REFERENCE('TABLE', 'TRANSACTIONS'),
    TIMESTAMP_COLNAME => 'TXN_DATE',
    TARGET_COLNAME => 'AMOUNT',
    LABEL_COLNAME => ''  -- unsupervised
);

-- Detect anomalies
CALL transaction_monitor!DETECT_ANOMALIES(
    INPUT_DATA => SYSTEM$REFERENCE('TABLE', 'NEW_TRANSACTIONS'),
    TIMESTAMP_COLNAME => 'TXN_DATE',
    TARGET_COLNAME => 'AMOUNT',
    CONFIG_OBJECT => {'prediction_interval': 0.99}
);

Classification (Cortex ML)

-- Create a classification model
CREATE OR REPLACE SNOWFLAKE.ML.CLASSIFICATION churn_classifier(
    INPUT_DATA => SYSTEM$REFERENCE('TABLE', 'TRAINING_DATA'),
    TARGET_COLNAME => 'CHURNED'
);

-- Predict on new data
SELECT *, churn_classifier!PREDICT(
    INPUT_DATA => OBJECT_CONSTRUCT(*)
) AS PREDICTION
FROM NEW_CUSTOMERS;

Training with External Frameworks

For models that require specific frameworks, you can train inside stored procedures with full access to scikit-learn, XGBoost, LightGBM, and PyTorch.

from snowflake.snowpark.functions import sproc
from snowflake.snowpark.types import StringType

@sproc(return_type=StringType(),
       input_types=[StringType()],
       packages=["snowflake-ml-python", "xgboost", "scikit-learn", "joblib"])
def train_xgboost(session, training_table):
    import xgboost as xgb
    from sklearn.model_selection import cross_val_score
    import joblib

    # Read data into pandas (runs inside Snowflake compute)
    df = session.table(training_table).to_pandas()
    X = df.drop("TARGET", axis=1)
    y = df["TARGET"]

    # Train with cross-validation
    model = xgb.XGBClassifier(
        n_estimators=200,
        max_depth=6,
        learning_rate=0.1,
        eval_metric="logloss"
    )
    scores = cross_val_score(model, X, y, cv=5, scoring="roc_auc")
    model.fit(X, y)

    # Save to stage
    joblib.dump(model, "/tmp/xgb_model.pkl")
    session.file.put("/tmp/xgb_model.pkl", "@ML_MODELS/xgboost/")

    return f"Model trained. CV AUC: {scores.mean():.4f} (+/- {scores.std():.4f})"

Snowpark ML Pipelines

Just like scikit-learn's Pipeline, Snowpark ML supports chaining preprocessing and modeling steps.

from snowflake.ml.modeling.pipeline import Pipeline
from snowflake.ml.modeling.preprocessing import StandardScaler, OneHotEncoder
from snowflake.ml.modeling.ensemble import RandomForestClassifier

# Build a pipeline
pipeline = Pipeline(steps=[
    ("scaler", StandardScaler(
        input_cols=["SPEND", "VISITS"],
        output_cols=["SPEND_SCALED", "VISITS_SCALED"]
    )),
    ("encoder", OneHotEncoder(
        input_cols=["CATEGORY"],
        output_cols=["CATEGORY_ENCODED"]
    )),
    ("classifier", RandomForestClassifier(
        input_cols=["SPEND_SCALED", "VISITS_SCALED", "CATEGORY_ENCODED"],
        label_cols=["CHURNED"],
        output_cols=["PREDICTION"]
    ))
])

# Fit and predict
pipeline.fit(df_train)
df_results = pipeline.predict(df_test)

Model Evaluation

Snowpark ML provides evaluation metrics that work directly with Snowpark DataFrames.

from snowflake.ml.modeling.metrics import (
    accuracy_score, precision_score, recall_score,
    f1_score, roc_auc_score, mean_squared_error, log_loss
)

# Classification metrics
accuracy = accuracy_score(df=df_predictions, y_true_col_names="CHURNED", y_pred_col_names="PREDICTION")
precision = precision_score(df=df_predictions, y_true_col_names="CHURNED", y_pred_col_names="PREDICTION")
recall = recall_score(df=df_predictions, y_true_col_names="CHURNED", y_pred_col_names="PREDICTION")
f1 = f1_score(df=df_predictions, y_true_col_names="CHURNED", y_pred_col_names="PREDICTION")

Hyperparameter Tuning

Snowpark ML supports grid search and random search for hyperparameter optimization.

from snowflake.ml.modeling.model_selection import GridSearchCV, RandomizedSearchCV

# Grid search
grid_search = GridSearchCV(
    estimator=RandomForestClassifier(
        input_cols=FEATURE_COLS,
        label_cols=[LABEL_COL],
        output_cols=["PREDICTION"]
    ),
    param_grid={
        "n_estimators": [50, 100, 200],
        "max_depth": [5, 10, 15],
        "min_samples_split": [2, 5, 10]
    },
    scoring="roc_auc",
    cv=5,
    input_cols=FEATURE_COLS,
    label_cols=[LABEL_COL],
    output_cols=["PREDICTION"]
)

grid_search.fit(df_train)
best_params = grid_search.to_sklearn().best_params_
best_score = grid_search.to_sklearn().best_score_

Practice Questions

Question 1

Question 2

Question 3

Question 4