Интеграция 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 — мы настроим пайплайн под ваш датасет. Свяжитесь для консультации — поможем выбрать оптимальную архитектуру и запустить первый пайплайн за неделю.

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

ETL для аналитики и ETL для ML — разные задачи. В аналитике важна агрегация, в ML — индивидуальные записи с историей. В аналитике train/val/test split не нужен, в ML — критичен. В аналитике skew данных мешает интерпретации, в ML — напрямую влияет на качество модели.

Инструменты. Apache Spark (Wikipedia) для больших объёмов (10GB+): PySpark с DataFrames, оптимизации через partitioning и caching. dbt для трансформаций поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версионируется, тестируется. Pandas + Polars для объёмов до нескольких GB — Polars в 5-10x быстрее Pandas на типичных трансформациях.

Temporal splits. Для ML важно, что split по времени, а не случайный. Если данные временные (транзакции, события пользователей), случайный split даёт data leakage: модель видит «будущие» данные при обучении. Правило: train на периоде T1-T2, validation на T2-T3 (с gap для предотвращения leakage), test на T3-T4. Неправильный split может стоить 10–15% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. Delta Lake или Apache Iceberg — форматы с ACID-транзакциями, Change Data Capture, time travel.

Как избежать training-serving skew с помощью Feature Store

Feature Store решает проблему рассинхронизации между обучением и инференсом. Самая коварная ошибка в ML-инфраструктуре — training-serving skew: признак считается по-разному в обучении и в продакшене. Модель учится на «правильных» данных, а инференс получает другие.

Feast (open source) — офлайн store на Parquet/Delta в S3 для обучения, онлайн store на Redis для low-latency инференса (<10ms). Feature definitions как Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition, используется везде. Нет расхождений.

Потоковые признаки. Когда признак должен обновляться в реальном времени (количество транзакций за последние 10 минут), нужна потоковая обработка. Apache Kafka + Apache Flink или Kafka Streams для вычисления признаков в реальном времени → запись в онлайн store. Сложнее, дороже, нужно только когда staleness признаков критична для качества.

Разметка данных: как не потратить бюджет впустую

Разметка — самая трудоёмкая и недооцениваемая часть ML-проекта. Плохо размеченные данные не исправит никакая архитектура.

Label Studio — open source, поддерживает разметку изображений (bounding box, polygon, segmentation), текста (NER, классификация), аудио, видео. Поднимается за 10 минут через Docker. Для небольших команд — первый выбор.

Оценка качества разметки. Inter-annotator agreement — насколько согласны разметчики между собой. Cohen's Kappa > 0.8 — хорошо, 0.6-0.8 — приемлемо, < 0.6 — задача неоднозначна или инструкция плохая. Пересечение разметок (10-20% примеров размечают два независимых аннотатора) — обязательная практика.

Active learning. Не размечать случайные примеры, а выбирать те, на которых модель наиболее неуверена (low confidence, high uncertainty). Позволяет добиться того же качества при 50-70% объёма разметки. Modals, Prodigy, Label Studio поддерживают active learning workflows. На одном из проектов для NLP мы сократили бюджет на разметку в 2,5 раза за счёт active learning.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для CV: рендеринг в Blender/Unity с реалистичными текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Риск: модель обучается на distribution синтетических данных, а не реальных — нужна осторожность и проверка на реальном holdout.

Качество данных: валидация и мониторинг

Great Expectations — de facto стандарт для data validation в ML-пайплайнах. Expectations — это декларативные утверждения о данных: «колонка age содержит значения от 0 до 120», «колонка user_id не содержит null», «распределение amount не отклоняется более чем на 20% от baseline». Запускается в пайплайне, при провале — блокирует прохождение.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation с type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель ожидает данные за последние N дней. ETL упал, данные не обновились — модель использует устаревшие признаки. Мониторинг свежести данных: timestamp последней записи в каждой таблице, алерт при задержке > порога.

Дедупликация. Дубликаты в обучающей выборке завышают метрики (одни и те же примеры в train и val) и искажают веса модели. MinHash LSH для приближённой дедупликации больших датасетов. Для точной — хэш по нормализованному контенту.

Инструменты валидации: сравнение

Хранилища и форматы

Для большинства ML-проектов на старте: Parquet в S3 + DVC для версионирования. Delta Lake или Iceberg — когда появляется потребность в инкрементальных обновлениях или time travel.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

• Аудит существующих данных и пайплайнов (1 неделя).
• Проектирование архитектуры: выбор инструментов, форматов, способов разметки.
• Реализация ETL/ELT пайплайна с валидацией и мониторингом.
• Документация кода и процессов (model card, data card).
• Обучение вашей команды работе с пайплайном.
• SLA на сопровождение и поддержку.

Как мы строим пайплайн: пошагово

1. Аудит существующих данных. Профилирование: ydata-profiling (бывший pandas-profiling) генерирует HTML-репорт со статистиками, дистрибуциями, корреляциями, missing values за минуты.
2. Проектирование пайплайна. Определяем источники данных, частоту обновления, требования к latency признаков, объёмы.
3. Реализация и тестирование. Unit-тесты на трансформации, integration-тесты на пайплайн, data validation через Great Expectations.
4. Деплой и мониторинг. Алерты на freshness, quality checks, аномалии в объёмах данных.

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

Свяжитесь с нами для бесплатного аудита ваших данных — оценим текущий пайплайн и предложим roadmap. Закажите построение ML-пайплайна под ключ.