ML con Snowflake: Feature Store y Model Registry

La stack de ML en Snowflake

Snowflake ofrece un stack completo de MLOps sin salir de la plataforma:

DATOS → Feature Store → Entrenamiento → Model Registry → Inferencia
         (features)     (Snowpark ML)    (registro)      (SPCS o UDF)

Feature Store

El Feature Store permite crear y compartir features (variables) calculadas para entrenamiento y predicción.

from snowflake.ml.feature_store import (
    FeatureStore, FeatureView, Entity, CreationMode
)

# Inicializar el Feature Store
fs = FeatureStore(
    session=session,
    database="ML_DB",
    name="PROD_FEATURE_STORE",
    default_warehouse="ML_WH",
    creation_mode=CreationMode.CREATE_IF_NOT_EXIST
)

# Definir una entidad (el "key" de las features)
cliente_entity = Entity(name="CLIENTE", join_keys=["cliente_id"])
fs.register_entity(cliente_entity)

# Crear features de comportamiento
features_sql = session.sql("""
    SELECT
        cliente_id,
        COUNT(*) AS compras_30d,
        SUM(monto) AS gasto_30d,
        AVG(monto) AS ticket_promedio_30d,
        MAX(fecha) AS ultima_compra
    FROM ventas
    WHERE fecha >= DATEADD('day', -30, CURRENT_DATE())
    GROUP BY cliente_id
""")

fv = FeatureView(
    name="FEATURES_COMPORTAMIENTO",
    entities=[cliente_entity],
    feature_df=features_sql,
    refresh_freq="1 day",
    desc="Features de comportamiento de compra últimos 30 días"
)

fv = fs.register_feature_view(feature_view=fv, version="V1")

Entrenamiento con Snowpark ML

from snowflake.ml.modeling.ensemble import GradientBoostingClassifier
from snowflake.ml.modeling.preprocessing import StandardScaler
from snowflake.ml.modeling.pipeline import Pipeline

# Obtener training data desde el Feature Store
training_data = fs.retrieve_feature_values(
    spine_df=session.table("clientes_etiquetados"),
    features=[fv],
    spine_timestamp_col="fecha_etiqueta"
)

# Pipeline de ML
pipeline = Pipeline([
    ("scaler", StandardScaler(
        input_cols=["compras_30d", "gasto_30d", "ticket_promedio_30d"],
        output_cols=["compras_scaled", "gasto_scaled", "ticket_scaled"]
    )),
    ("model", GradientBoostingClassifier(
        input_cols=["compras_scaled", "gasto_scaled", "ticket_scaled"],
        label_cols=["churn_label"],
        output_cols=["prediccion"]
    ))
])

pipeline.fit(training_data)

Model Registry

from snowflake.ml.registry import Registry

registry = Registry(session=session, database_name="ML_DB", schema_name="MODELS")

# Registrar el modelo
model_version = registry.log_model(
    model=pipeline,
    model_name="modelo_churn_clientes",
    version_name="V1_0",
    comment="Gradient Boosting para predicción de churn",
    metrics={"accuracy": 0.87, "auc": 0.92},
    tags={"team": "data_science", "proyecto": "retencion"}
)

# Inferencia desde SQL
registry.deploy(
    model_name="modelo_churn_clientes",
    version_name="V1_0",
    platform="WAREHOUSE"
)

-- Predicción en batch desde SQL
SELECT
    c.cliente_id,
    c.nombre,
    modelo_churn_clientes!PREDICT(
        compras_30d, gasto_30d, ticket_promedio_30d
    ) AS prediccion_churn
FROM clientes c
JOIN features_comportamiento f ON c.cliente_id = f.cliente_id
WHERE prediccion_churn > 0.7;  -- clientes con >70% probabilidad de churn

Próximos pasos

Combina el ML predictivo con Cortex Agents para crear sistemas de decisión automáticos.