Інтеграція Apache Spark MLlib для обробки великих даних

Ви завантажили 500 ГБ даних, запустили .fit() і через годину отримали OutOfMemoryError. Pandas та scikit-learn упираються в RAM однієї машини. Рішення — розподілене навчання на Spark MLlib. Ми займаємося розподіленим ML на Spark більше 5 років, виконали 20+ проєктів з датасетами до 10 ТБ. Нижче — практичний досвід, який допоможе уникнути типових помилок і швидко запустити навчання на кластері.

Проблеми масштабування та feature engineering

Spark MLlib розподіляє обчислення на кластер, обробляючи дані будь-якого обсягу. Типовий поріг — датасети >100 GB або >100 млн рядків, де pipeline на sklearn перестає працювати. Сотні ознак, категоріальні змінні з мільйонами унікальних значень — StringIndexer і OneHotEncoder в Spark справляються без вивантаження в пам'ять. CrossValidator на Spark запускає фолди паралельно, прискорюючи підбір гіперпараметрів у 3–5 разів. Вбудовані інструменти для роботи з пропусками (Imputer) та масштабування (StandardScaler) дозволяють будувати пайплайни без перемикання контексту.

Порівняння Spark MLlib з альтернативами

Порівняємо з підходами на pandas+sklearn та Dask. Spark виграє за рахунок нативної підтримки розподілених DataFrames, оптимізованих під shuffle, вбудованих алгоритмів (GBT, RandomForest, KMeans) та інтеграції з MLflow для трекінгу експериментів.

Spark MLlib на 80% швидший при вдвічі меншому споживанні ресурсів на executor.

Як налаштувати Spark MLlib для оптимальної продуктивності?

Ці налаштування скорочують час CV з 6 до 1.5 годин на кластері з 10 executor.

Чому Spark MLlib швидший за sklearn на великих даних?

Spark MLlib використовує розподілені обчислення та оптимізовані алгоритми для роботи з даними, що не поміщаються в пам'ять. На відміну від sklearn, який завантажує все в RAM, Spark обробляє дані частинами на кластері. Це дозволяє досягти лінійної масштабованості при додаванні вузлів.

Реалізація пайплайну: від конфігурації до деплою

Конфігурація кластера та підготовка даних

Використовуємо PySpark 3.4+, MLflow 2.x, ONNX для інференсу. Конфігурація кластера підбирається під задачу: executor memory від 4 до 16 GB, кількість executor кратна числу партицій. Нижче — робочий пайплайн для бінарної класифікації з градієнтним бустингом.

from pyspark.sql import SparkSession from pyspark.ml import Pipeline from pyspark.ml.feature import (VectorAssembler, StringIndexer, StandardScaler, Imputer) from pyspark.ml.classification import GBTClassifier, RandomForestClassifier from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml.tuning import CrossValidator, ParamGridBuilder spark = SparkSession.builder \ .appName("ML Pipeline") \ .config("spark.executor.memory", "8g") \ .config("spark.executor.core", "4") \ .config("spark.executor.instances", "10") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.ml.param.maxParallelism", "4") \ .getOrCreate() # Загрузка данных df = spark.read.parquet("s3://data/training/*.parquet") df = df.repartition(200) # Оптимальное число партиций # Feature engineering numeric_cols = ['amount', 'age', 'days_since_last_tx', 'tx_count_30d'] categorical_cols = ['category', 'country', 'device_type'] # Imputer для числовых imputer = Imputer( inputCols=numeric_cols, outputCols=[f"{c}_imputed" for c in numeric_cols], strategy="median" ) # Кодирование категориальных indexers = [ StringIndexer(inputCol=col, outputCol=f"{col}_idx", handleInvalid="keep") for col in categorical_cols ] # Сборка вектора признаков all_feature_cols = ( [f"{c}_imputed" for c in numeric_cols] + [f"{c}_idx" for c in categorical_cols] ) assembler = VectorAssembler( inputCols=all_feature_cols, outputCol="features_raw", handleInvalid="keep" ) scaler = StandardScaler( inputCol="features_raw", outputCol="features", withMean=True, withStd=True ) # Модель gbt = GBTClassifier( labelCol="label", featuresCol="features", maxIter=100, maxDepth=5, stepSize=0.05, subsamplingRate=0.8, seed=42 ) # Pipeline pipeline = Pipeline(stages=[ imputer, *indexers, assembler, scaler, gbt ]) # Train/test split train_df, test_df = df.randomSplit([0.8, 0.2], seed=42) # Обучение model = pipeline.fit(train_df) predictions = model.transform(test_df) # Оценка evaluator = BinaryClassificationEvaluator( labelCol="label", rawPredictionCol="rawPrediction", metricName="areaUnderROC" ) auc = evaluator.evaluate(predictions) print(f"Test AUC: {auc:.4f}") 

Гіперпараметрична оптимізація

# Cross-validation на кластере param_grid = ParamGridBuilder() \ .addGrid(gbt.maxDepth, [4, 6, 8]) \ .addGrid(gbt.maxIter, [50, 100]) \ .addGrid(gbt.stepSize, [0.05, 0.1]) \ .build() cv = CrossValidator( estimator=pipeline, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=3, parallelism=4, # Параллельный запуск фолдов seed=42 ) cv_model = cv.fit(train_df) best_model = cv_model.bestModel print(f"Best params: {cv_model.bestModel.stages[-1].extractParamMap()}") 

Feature Importance та інтерпретація

# Извлечение feature importance gbt_model = best_model.stages[-1] importance = gbt_model.featureImportances # Маппинг на имена признаков feature_names = all_feature_cols importance_df = spark.createDataFrame( [(name, float(imp)) for name, imp in zip(feature_names, importance.toArray())], ["feature", "importance"] ).orderBy("importance", ascending=False) importance_df.show(20) # SHAP через pandas на выборке sample_pandas = predictions.sample(fraction=0.01).toPandas() # ... далее стандартный TreeExplainer 

Збереження та деплой моделі

import mlflow import mlflow.spark # Логирование в MLflow with mlflow.start_run(): mlflow.log_param("max_depth", gbt.getMaxDepth()) mlflow.log_param("max_iter", gbt.getMaxIter()) mlflow.log_metric("auc", auc) # Сохранение Spark модели mlflow.spark.log_model(best_model, "spark_model") # Экспорт в ONNX для быстрого инференса from onnxmltools import convert_sparkml onnx_model = convert_sparkml(best_model, "GBT Model", test_df.limit(5)) mlflow.onnx.log_model(onnx_model, "onnx_model") # Загрузка для предсказаний loaded_model = mlflow.spark.load_model("runs:/RUN_ID/spark_model") batch_predictions = loaded_model.transform(new_data_df) 

Що входить в роботу

При замовленні інтеграції Spark MLlib ви отримуєте:

• Документація пайплайну та конфігурацій (включаючи обґрунтування вибору алгоритмів та параметрів).
• Налаштований MLflow tracking server для відтворюваності експериментів.
• Docker-образ для інференсу з ONNX Runtime.
• Навчання команди замовника (1 день) — робота з PySpark, MLflow, оптимізація.
• Підтримка 2 тижні після деплою — виправлення інцидентів, консультації.

Процес роботи, строки та гарантії

1. Аналітика: вивчаємо дані, визначаємо ознаки, цільову змінну, метрики.
2. Проектування: вибираємо алгоритм, конфігурацію кластера, пайплайн.
3. Реалізація: пишемо код, налаштовуємо MLflow, запускаємо пробне навчання.
4. Тестування: валідація на відкладеній вибірці, A/B тест у стейджингу.
5. Деплой: упаковка в MLflow Model, сервінг через REST API або ONNX.

Строки під ключ — від 2 до 4 тижнів залежно від складності даних та алгоритму. Вартість розраховується індивідуально після аудиту вашого датасету та інфраструктури. Отримайте консультацію — оцінимо проєкт безкоштовно.

Багаторічний досвід у промисловому ML на Spark, десятки завершених проєктів з датасетами від 100 GB до 10 TB. Сертифіковані спеціалісти з Spark та MLflow. Гарантуємо відтворюваність результатів.

Замовте інтеграцію Spark MLlib — ми налаштуємо пайплайн під ваш датасет. Зв'яжіться для консультації — допоможемо вибрати оптимальну архітектуру та запустити перший пайплайн за тиждень.