Overview

Prerequisites

What You'll Learn

What You'll Need

What You'll Build

Set Up the Environment

USE ROLE ACCOUNTADMIN;

CREATE WAREHOUSE IF NOT EXISTS REVIEW_WH
  WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE;

CREATE DATABASE IF NOT EXISTS REVIEW_INTEL;
CREATE SCHEMA IF NOT EXISTS REVIEW_INTEL.EVAL;

USE WAREHOUSE REVIEW_WH;
USE SCHEMA REVIEW_INTEL.EVAL;

CREATE OR REPLACE FILE FORMAT JSONL_FORMAT TYPE = 'JSON';
CREATE OR REPLACE STAGE REVIEW_FILES
  FILE_FORMAT = JSONL_FORMAT
  DIRECTORY = (ENABLE = TRUE);

CREATE STAGE IF NOT EXISTS MODEL_FILES;

CREATE OR REPLACE PROCEDURE CHECK_PYTHON()
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.11'
  PACKAGES = ('snowflake-snowpark-python', 'scikit-learn', 'numpy')
  HANDLER = 'run'
AS $$
import numpy, sklearn
def run(session):
    return f"python ok | numpy {numpy.__version__} | sklearn {sklearn.__version__}"
$$;

CALL CHECK_PYTHON();

Load the Data

CREATE OR REPLACE TABLE REVIEWS_ALL (
  ID     STRING,
  SPLIT  STRING,      -- train / valid / test (the dataset's own split)
  TEXT   STRING,
  STARS  NUMBER(1),
  LABEL  STRING       -- negative (1-2 stars) / neutral (3) / positive (4-5)
);

COPY INTO REVIEWS_ALL (ID, SPLIT, TEXT, STARS, LABEL)
FROM (
  SELECT $1:id::STRING, $1:split::STRING, $1:text::STRING, $1:stars::NUMBER, $1:label::STRING
  FROM @REVIEW_FILES
)
FILES = ('reviews_pl.jsonl');

-- One row per review per Claude model, with the real cost and latency of that API call.
CREATE OR REPLACE TABLE CLAUDE_PREDICTIONS (
  ID              STRING,
  MODEL           STRING,
  PRED            STRING,
  COST_USD        FLOAT,
  LATENCY_S       FLOAT,
  IN_EVAL_SAMPLE  BOOLEAN
);

COPY INTO CLAUDE_PREDICTIONS (ID, MODEL, PRED, COST_USD, LATENCY_S, IN_EVAL_SAMPLE)
FROM (
  SELECT $1:id::STRING, $1:model::STRING, $1:pred::STRING, $1:cost_usd::FLOAT,
         $1:latency_s::FLOAT, $1:in_eval_sample::BOOLEAN
  FROM @REVIEW_FILES
)
FILES = ('claude_predictions_pl.jsonl');

SELECT 'reviews: ' || SPLIT AS WHAT, COUNT(*) AS N FROM REVIEWS_ALL GROUP BY SPLIT
UNION ALL
SELECT 'claude: ' || MODEL, COUNT(*) FROM CLAUDE_PREDICTIONS GROUP BY MODEL
ORDER BY WHAT;

aside positive The Claude predictions were produced outside Snowflake with the Anthropic API, using a short prompt that asks for one of the three labels. The script is in the companion repository (src/llm_eval.py) if you want to regenerate them with other models. You do not need an API key to follow this guide.

Train the Model in Snowflake

CREATE OR REPLACE PROCEDURE TRAIN_REVIEW_MODEL()
  RETURNS VARIANT
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.11'
  PACKAGES = ('snowflake-snowpark-python', 'scikit-learn', 'pandas', 'pyarrow', 'numpy', 'joblib')
  HANDLER = 'run'
  EXECUTE AS CALLER
AS $$
import os
import tempfile
import time

import joblib
import pandas as pd
import sklearn
from sklearn.compose import ColumnTransformer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, f1_score
from sklearn.pipeline import Pipeline

MODEL_FILE = "review_sentiment.joblib"
# The UDF body below needs a double-dollar delimiter. It is built from two pieces so
# the SQL parser does not read it as the end of this procedure's own body.
DQ = "$" + "$"


def build(c):
    features = ColumnTransformer([
        ("word", TfidfVectorizer(ngram_range=(1, 2), min_df=2, sublinear_tf=True,
                                 max_features=200_000), "TEXT"),
        ("char", TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 5), min_df=3,
                                 sublinear_tf=True, max_features=300_000), "TEXT"),
    ])
    clf = LogisticRegression(C=c, max_iter=3000, class_weight="balanced")
    return Pipeline([("features", features), ("clf", clf)])


def run(session):
    df = session.table("REVIEWS_ALL").select("ID", "SPLIT", "TEXT", "LABEL").to_pandas()
    train, valid, test = (df[df.SPLIT == s] for s in ("train", "valid", "test"))

    # Pick the regularization strength on the validation split.
    search = {}
    for c in (0.5, 1.0, 2.0, 4.0, 8.0):
        model = build(c).fit(train[["TEXT"]], train.LABEL)
        search[c] = f1_score(valid.LABEL, model.predict(valid[["TEXT"]]), average="macro")
    best_c = max(search, key=search.get)

    # Final model on train + valid, scored once on test.
    fit_df = pd.concat([train, valid])
    model = build(best_c).fit(fit_df[["TEXT"]], fit_df.LABEL)
    test_pred = model.predict(test[["TEXT"]])
    result = {
        "best_C": best_c,
        "valid_macro_f1_by_C": {str(k): round(v, 4) for k, v in search.items()},
        "test_accuracy": round(accuracy_score(test.LABEL, test_pred), 4),
        "test_macro_f1": round(f1_score(test.LABEL, test_pred, average="macro"), 4),
        "classes": list(model.classes_),
        "sklearn_version": sklearn.__version__,
    }

    # TfidfVectorizer keeps every pruned n-gram in stop_words_ (millions of character
    # n-grams here). It is not used for prediction, and dropping it keeps the file small.
    for _, vectorizer, _ in model.named_steps["features"].transformers_:
        if hasattr(vectorizer, "stop_words_"):
            delattr(vectorizer, "stop_words_")

    # Save the model to a stage.
    path = os.path.join(tempfile.gettempdir(), MODEL_FILE)
    joblib.dump(model, path)
    result["model_file_mb"] = round(os.path.getsize(path) / 1e6, 1)
    session.file.put(path, "@MODEL_FILES", auto_compress=False, overwrite=True)

    # Serve it as a SQL function. The model is loaded once per Python process.
    udf_sql = """
CREATE OR REPLACE FUNCTION PREDICT_REVIEW_PROBA(TEXT STRING)
  RETURNS ARRAY
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.11'
  PACKAGES = ('{SKLEARN}', 'pandas', 'joblib')
  IMPORTS = ('@MODEL_FILES/{MODEL_FILE}')
  HANDLER = 'predict'
  COMMENT = 'Probabilities for [negative, neutral, positive]'
AS {DQ}
import os
import sys

import joblib
import pandas as pd

_model = joblib.load(os.path.join(sys._xoptions["snowflake_import_directory"], "{MODEL_FILE}"))

def predict(text):
    proba = _model.predict_proba(pd.DataFrame({{"TEXT": [text or ""]}}))[0]
    return [float(p) for p in proba]
{DQ}
"""
    # Pin the sklearn version used for training (pickled models are version-sensitive).
    try:
        session.sql(udf_sql.format(SKLEARN=f"scikit-learn=={sklearn.__version__}",
                                   MODEL_FILE=MODEL_FILE, DQ=DQ)).collect()
        result["serving_sklearn"] = f"pinned {sklearn.__version__}"
    except Exception:
        session.sql(udf_sql.format(SKLEARN="scikit-learn", MODEL_FILE=MODEL_FILE, DQ=DQ)).collect()
        result["serving_sklearn"] = "unpinned (pin was not available)"

    # Label every review with one SQL query through the served function.
    t0 = time.time()
    session.sql("""
CREATE OR REPLACE TABLE SNOWFLAKE_MODEL_PREDICTIONS AS
SELECT ID, P[0]::FLOAT AS P_NEGATIVE, P[1]::FLOAT AS P_NEUTRAL, P[2]::FLOAT AS P_POSITIVE
FROM (SELECT ID, PREDICT_REVIEW_PROBA(TEXT) AS P FROM REVIEWS_ALL)
""").collect()
    result["inference_rows"] = session.table("SNOWFLAKE_MODEL_PREDICTIONS").count()
    result["inference_seconds"] = round(time.time() - t0, 1)
    return result
$$;

CALL TRAIN_REVIEW_MODEL();

SELECT PREDICT_REVIEW_PROBA('Produkt zgodny z opisem, szybka wysyłka, polecam!') AS P_NEG_NEU_POS;

aside positive Why classes_ matters. scikit-learn sorts class labels alphabetically, so the order is negative, neutral, positive. The procedure returns classes in its result so you can confirm the order the SQL code below relies on.

Measure the Model in SQL

CREATE OR REPLACE VIEW SF_MODEL AS
SELECT ID, P_NEGATIVE, P_NEUTRAL, P_POSITIVE,
       GREATEST(P_NEGATIVE, P_NEUTRAL, P_POSITIVE) AS CONFIDENCE,
       CASE GREATEST(P_NEGATIVE, P_NEUTRAL, P_POSITIVE)
            WHEN P_NEGATIVE THEN 'negative'
            WHEN P_NEUTRAL  THEN 'neutral'
            ELSE 'positive' END AS PRED
FROM SNOWFLAKE_MODEL_PREDICTIONS;

WITH scored AS (
  SELECT r.LABEL, m.PRED
  FROM REVIEWS_ALL r JOIN SF_MODEL m USING (ID)
  WHERE r.SPLIT = 'test'
),
per_class AS (
  SELECT cls.L,
         SUM(IFF(PRED = cls.L AND LABEL = cls.L, 1, 0))  AS TP,
         SUM(IFF(PRED = cls.L AND LABEL <> cls.L, 1, 0)) AS FP,
         SUM(IFF(PRED <> cls.L AND LABEL = cls.L, 1, 0)) AS FN
  FROM scored
  CROSS JOIN (SELECT COLUMN1 AS L FROM VALUES ('negative'), ('neutral'), ('positive')) cls
  GROUP BY cls.L
)
SELECT ROUND(AVG(2 * TP / NULLIF(2 * TP + FP + FN, 0)), 3) AS MACRO_F1,
       (SELECT ROUND(AVG(IFF(PRED = LABEL, 1, 0)), 3) FROM scored) AS ACCURACY
FROM per_class;

Route the Hard Reviews to an LLM

CREATE OR REPLACE TABLE ROUTER_CURVE AS
WITH thresholds AS (
  SELECT ROUND(0.34 + 0.02 * SEQ4(), 2) AS T FROM TABLE(GENERATOR(ROWCOUNT => 34))
),
joined AS (
  SELECT c.MODEL, r.ID, r.LABEL, m.PRED AS SF_PRED, m.CONFIDENCE, c.PRED AS LLM_PRED, c.COST_USD
  FROM CLAUDE_PREDICTIONS c
  JOIN SF_MODEL m USING (ID)
  JOIN REVIEWS_ALL r USING (ID)
  WHERE c.IN_EVAL_SAMPLE
),
routed AS (
  SELECT t.T, j.MODEL, j.LABEL,
         j.CONFIDENCE < t.T AS TO_LLM,
         IFF(j.CONFIDENCE < t.T, j.LLM_PRED, j.SF_PRED) AS FINAL_PRED,
         IFF(j.CONFIDENCE < t.T, j.COST_USD, 0) AS COST_USD
  FROM thresholds t CROSS JOIN joined j
),
per_class AS (
  SELECT T, MODEL, cls.L AS CLASS,
         SUM(IFF(FINAL_PRED = cls.L AND LABEL = cls.L, 1, 0)) AS TP,
         SUM(IFF(FINAL_PRED = cls.L AND LABEL <> cls.L, 1, 0)) AS FP,
         SUM(IFF(FINAL_PRED IS DISTINCT FROM cls.L AND LABEL = cls.L, 1, 0)) AS FN
  FROM routed
  CROSS JOIN (SELECT COLUMN1 AS L FROM VALUES ('negative'), ('neutral'), ('positive')) cls
  GROUP BY T, MODEL, cls.L
)
SELECT r.T, r.MODEL,
       ROUND(AVG(IFF(r.TO_LLM, 1, 0)), 3)               AS SHARE_TO_LLM,
       ROUND(AVG(IFF(r.FINAL_PRED = r.LABEL, 1, 0)), 3) AS ACCURACY,
       ROUND(f.MACRO_F1, 3)                             AS MACRO_F1,
       ROUND(1000 * SUM(r.COST_USD) / COUNT(*), 4)      AS COST_PER_1K_USD
FROM routed r
JOIN (SELECT T, MODEL, AVG(2 * TP / NULLIF(2 * TP + FP + FN, 0)) AS MACRO_F1
      FROM per_class GROUP BY T, MODEL) f
  ON f.T = r.T AND f.MODEL = r.MODEL
GROUP BY r.T, r.MODEL, f.MACRO_F1;

WITH full_llm AS (
  SELECT MODEL, MACRO_F1 AS FULL_F1, COST_PER_1K_USD AS FULL_COST
  FROM ROUTER_CURVE WHERE T = 1.00
),
best AS (
  SELECT c.*, f.FULL_F1, f.FULL_COST,
         ROW_NUMBER() OVER (PARTITION BY c.MODEL ORDER BY c.COST_PER_1K_USD, c.T) AS RN
  FROM ROUTER_CURVE c JOIN full_llm f USING (MODEL)
  WHERE c.MACRO_F1 >= f.FULL_F1 - 0.01
)
SELECT MODEL, T, SHARE_TO_LLM,
       MACRO_F1, FULL_F1 AS ALL_LLM_MACRO_F1,
       COST_PER_1K_USD, FULL_COST AS ALL_LLM_COST_PER_1K_USD,
       ROUND(100 * (1 - COST_PER_1K_USD / NULLIF(FULL_COST, 0))) AS PCT_CHEAPER
FROM best WHERE RN = 1
ORDER BY MODEL;

aside negative The 300-review sample is small. A difference of 0.01 macro-F1 is within noise, so treat the "within one point" rule as a practical choice, not a precise one. In production, choose T on one sample and confirm it on another.

Trial Account Notes

Clean Up

DROP DATABASE IF EXISTS REVIEW_INTEL;
DROP WAREHOUSE IF EXISTS REVIEW_WH;

Conclusion and Resources

What You Learned

Related Resources