Интеграция Snowflake ML для аналитики и машинного обучения

Отметим: когда объемы данных превышают 500 ГБ, традиционные ML-пайплайны упираются в узкое место: экспорт в S3, обучение на отдельном кластере, загрузка результатов обратно. Это отнимает часы и создает compliance-риски — данные покидают защищенную среду DWH. В финансах и e-commerce такой подход часто неприемлем. Мы внедряем Snowflake ML для аналитики и машинного обучения, ускоряя пайплайны в 3–5 раз и снижая инфраструктурные затраты на 40–60%.

Snowflake ML исключает перемещение данных, позволяя обучать модели прямо внутри DWH. Это не только сокращает расходы, но и гарантирует соответствие требованиям GDPR и PCI DSS. Для финтех-клиента, обрабатывающего 10 млн транзакций в день, мы сократили время обучения модели с 4 часов до 45 минут, снизив вычислительные затраты более чем на 50%. Платформа включает три ключевых компонента: Snowpark ML для построения пайплайнов, Feature Store для управления признаками и Model Registry для версионирования моделей. Кроме того, Cortex AI предоставляет встроенные LLM-функции — тональный анализ, классификацию, суммаризацию — доступные через SQL. Это снижает инфраструктурную нагрузку и ускоряет вывод ML-продуктов на рынок.

Какие проблемы решает Snowflake ML?

Первая и самая острая — «data movement bottleneck». Экспорт 500+ ГБ данных в SageMaker или Vertex AI занимает часы и требует дополнительных платных хранилищ. Snowflake ML обучает модели прямо в DWH, экономя не только время, но и до 50% инфраструктурных затрат. Вторая проблема — compliance. В финансах и медицине данные не могут покидать аккаунт; Snowflake ML гарантирует, что все операции — обучение, инференс, хранение фичей — выполняются внутри защищённого периметра. Третья — latency инференса. Модели разворачиваются как SQL-функции: latency p99 менее 100 мс, что подходит для real-time scoring транзакций.

Как работает Snowpark ML Pipeline?

from snowflake.ml.modeling.pipeline import Pipeline from snowflake.ml.modeling.preprocessing import StandardScaler, OrdinalEncoder from snowflake.ml.modeling.ensemble import GradientBoostingClassifier from snowflake.ml.modeling.model_selection import cross_validate from snowflake.snowpark import Session session = Session.builder.configs({ "account": "your-account", "user": "ml_user", "password": "...", "role": "ML_ROLE", "warehouse": "ML_WH", "database": "ML_DB", "schema": "FEATURES" }).create() df = session.table("TRAINING_DATA") from snowflake.snowpark import functions as F features_df = df.select( "USER_ID", "LABEL", F.col("AMOUNT").cast("float").alias("AMOUNT"), F.datediff("day", F.col("LAST_TX_DATE"), F.current_date()).alias("DAYS_SINCE_TX"), (F.col("TX_COUNT_30D") / F.col("TX_COUNT_90D")).alias("TX_ACCELERATION"), F.col("MERCHANT_CATEGORY"), F.col("COUNTRY") ) train_df, test_df = features_df.random_split([0.8, 0.2], seed=42) pipeline = Pipeline(steps=[ ("encoder", OrdinalEncoder( input_cols=["MERCHANT_CATEGORY", "COUNTRY"], output_cols=["MERCHANT_CATEGORY_ENC", "COUNTRY_ENC"] )), ("scaler", StandardScaler( input_cols=["AMOUNT", "DAYS_SINCE_TX", "TX_ACCELERATION"], output_cols=["AMOUNT_SCALED", "DAYS_SCALED", "TX_ACCEL_SCALED"] )), ("model", GradientBoostingClassifier( input_cols=["AMOUNT_SCALED", "DAYS_SCALED", "TX_ACCEL_SCALED", "MERCHANT_CATEGORY_ENC", "COUNTRY_ENC"], label_col="LABEL", output_cols=["PREDICTED_LABEL"], n_estimators=200, learning_rate=0.05, max_depth=5 )) ]) fitted_pipeline = pipeline.fit(train_df) predictions = fitted_pipeline.transform(test_df) 

Код выполняется в Snowflake — данные не покидают DWH. На практике такой конвейер на 6 млн транзакций обучается за 15 минут, что в 3 раза быстрее выгрузки в SageMaker.

Как настроить Feature Store?

from snowflake.ml.feature_store import FeatureStore, FeatureView, Entity import snowflake.ml.feature_store as fstore fs = FeatureStore( session=session, database="ML_DB", name="PRODUCTION_FS", default_warehouse="ML_WH" ) user_entity = Entity(name="USER", join_keys=["USER_ID"]) merchant_entity = Entity(name="MERCHANT", join_keys=["MERCHANT_ID"]) fs.register_entity(user_entity) fs.register_entity(merchant_entity) user_feature_view = FeatureView( name="USER_TX_FEATURES", entities=[user_entity], feature_df=session.sql(""" SELECT USER_ID, COUNT(*) OVER (PARTITION BY USER_ID ORDER BY TX_DATE RANGE BETWEEN 30 PRECEDING AND CURRENT ROW) as TX_COUNT_30D, SUM(AMOUNT) OVER (PARTITION BY USER_ID ORDER BY TX_DATE RANGE BETWEEN 30 PRECEDING AND CURRENT ROW) as TX_AMOUNT_30D, AVG(AMOUNT) OVER (PARTITION BY USER_ID ORDER BY TX_DATE RANGE BETWEEN 7 PRECEDING AND CURRENT ROW) as TX_AVG_7D FROM TRANSACTIONS """), refresh_freq="1 day", desc="User transaction features, rolling windows" ) registered_fv = fs.register_feature_view(user_feature_view, version="v1") dataset = fs.generate_dataset( spine_df=session.table("TRAINING_LABELS"), features=[registered_fv], spine_timestamp_col="TX_DATE", name="fraud_training_v1", desc="Fraud detection training set" ) 

Feature Store автоматически обновляет фичи через таски Snowflake, избавляя команду от ручного пересчёта.

from snowflake.ml.registry import Registry registry = Registry(session=session, database_name="ML_DB", schema_name="MODELS") model_ref = registry.log_model( fitted_pipeline, model_name="FRAUD_DETECTION", version_name="v1_0", comment="GBT fraud detection model, trained on 6M transactions", metrics={"test_auc": 0.934, "test_f1": 0.812}, tags={"team": "risk", "env": "production"} ) model_ref.deploy( deployment_name="fraud_scoring", platform="WAREHOUSE", target_method="predict", options={"compute_pool": "ML_COMPUTE_POOL"} ) session.sql(""" SELECT t.TRANSACTION_ID, t.AMOUNT, FRAUD_DETECTION!PREDICT(t.AMOUNT_SCALED, ...) as FRAUD_SCORE FROM TRANSACTIONS t WHERE TX_DATE = CURRENT_DATE() """).show()