Інтеграція Snowflake ML для аналітики та машинного навчання

Зазначимо: коли обсяги даних перевищують 500 ГБ, традиційні ML-пайплайни впираються у вузьке місце: експорт у S3, навчання на окремому кластері, завантаження результатів назад. Це забирає години та створює комплаєнс-ризики — дані залишають захищене середовище 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-продуктів на ринок.

Перша та найгостріша проблема — «data movement bottleneck». Експорт 500+ ГБ даних у SageMaker або Vertex AI займає години та потребує додаткових платних сховищ. Snowflake ML навчає моделі прямо в DWH, економлячи не лише час, але й до 50% інфраструктурних витрат. Друга проблема — комплаєнс: у фінансах та медицині дані не можуть залишати акаунт; 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()