PySpark для обработки больших данных: от основ до ML

Введение в экосистему Apache Spark и настройка окружения PySpark

Добро пожаловать в курс по PySpark! Если вы читаете эту статью, значит, вы столкнулись с проблемой: ваши данные стали слишком большими для обработки на одном компьютере с помощью Excel или Pandas, или вы просто хотите освоить один из самых востребованных инструментов в мире Big Data. В этой статье мы разберем, что такое Apache Spark, почему он стал стандартом индустрии и как подготовить ваш компьютер к работе с ним.

Зачем нам нужен Spark?

Представьте, что вам нужно перевезти 100 тонн песка. У вас есть два варианта:

1. Использовать один гигантский самосвал, который может перевезти всё за один раз, но стоит безумных денег и требует специальной дороги.
2. Использовать 100 обычных грузовиков, которые перевезут груз параллельно.

В мире данных первый вариант — это вертикальное масштабирование (покупка суперкомпьютера). Второй вариант — это горизонтальное масштабирование (распределенные вычисления). Apache Spark — это технология, которая позволяет управлять этими «100 грузовиками» так, будто это одна машина.

Традиционные инструменты, такие как библиотека Pandas в Python, загружают все данные в оперативную память (RAM) одной машины. Если у вас 16 ГБ памяти, а файл весит 100 ГБ, Pandas выдаст ошибку MemoryError. Spark же разбивает эти 100 ГБ на мелкие кусочки и обрабатывает их частями или распределяет по нескольким компьютерам.

Что такое Apache Spark?

Apache Spark — это быстрый универсальный движок для крупномасштабной обработки данных. Он был разработан в университете Беркли в 2009 году как более быстрая альтернатива Hadoop MapReduce.

Главная особенность Spark — вычисления в оперативной памяти (In-Memory Computing). В отличие от своих предшественников, которые постоянно сохраняли промежуточные результаты на жесткий диск, Spark старается держать данные в RAM. Это делает его в 100 раз быстрее при определенных задачах.

Экосистема Spark

Spark — это не просто одна библиотека, а целая экосистема инструментов, работающих на одном ядре.

Основные компоненты:

• Spark Core: Базовый движок, отвечающий за распределение задач, работу с памятью и отказоустойчивость.
• Spark SQL: Самый популярный модуль. Позволяет работать с данными как с таблицами (DataFrames) и использовать SQL-запросы.
• Spark Streaming (и Structured Streaming): Обработка данных в реальном времени.
• MLlib: Библиотека машинного обучения (классификация, регрессия, кластеризация).
• GraphX: Обработка графов (социальные сети, маршруты).

Архитектура Spark: Как это работает?

Чтобы писать эффективный код, важно понимать, что происходит «под капотом». Spark использует архитектуру Master-Slave (Ведущий-Ведомый).

Ключевые понятия:

1. Driver (Драйвер): Это «мозг» вашего приложения. Здесь работает ваш Python-код. Драйвер преобразует ваш код в задачи (tasks) и отправляет их на выполнение.
2. Cluster Manager (Менеджер кластера): Это «отдел кадров». Он знает, сколько у нас есть свободных компьютеров и выделяет ресурсы под задачи Драйвера.
3. Executors (Исполнители): Это «рабочие». Процессы, запущенные на рабочих узлах (Worker Nodes). Они выполняют вычисления и хранят данные.

Когда вы запускаете код на PySpark, происходит магия: ваш Python-код транслируется в команды для JVM (Java Virtual Machine), так как сам Spark написан на языке Scala (который работает на Java). Для этого используется библиотека Py4J.

Почему PySpark?

PySpark — это Python-интерфейс для Apache Spark. Он объединяет простоту Python с мощностью Spark.

Почему выбирают PySpark:

• Простота: Python знают почти все аналитики и Data Scientist'ы.
• Библиотеки: Можно легко использовать вместе с NumPy, Scikit-learn и другими библиотеками Python.
• Сообщество: Огромное количество документации и примеров.

Настройка окружения

Для начала работы нам не обязательно иметь кластер из 10 серверов. Spark отлично работает в локальном режиме (Local Mode), используя все ядра вашего процессора.

Шаг 1: Установка Java (JDK)

Поскольку Spark работает на JVM, наличие Java обязательно. Spark обычно требует Java 8, 11 или 17.

Важно: Убедитесь, что путь к Java не содержит пробелов (например, Program Files в Windows может вызывать проблемы, лучше устанавливать в корень диска или использовать короткие пути).

Для проверки наличия Java введите в терминале:

java -version

Если команда не найдена, скачайте и установите OpenJDK 11.

Шаг 2: Установка Python

Рекомендуется использовать Python версии 3.7 и выше. Проверьте версию:

python --version

Шаг 3: Установка PySpark

Самый простой способ установить PySpark для локальной разработки — использовать менеджер пакетов pip. Откройте терминал (или командную строку) и введите:

pip install pyspark

Эта команда скачает сам Spark и необходимые Python-обертки. Размер скачивания может быть около 300 МБ.

Шаг 4: Проверка установки (Hello World)

Давайте напишем нашу первую программу, чтобы убедиться, что всё работает. Мы создадим SparkSession — точку входа в любое приложение Spark.

Создайте файл test_spark.py или запустите Jupyter Notebook и выполните следующий код:

from pyspark.sql import SparkSession

# 1. Создаем сессию Spark
# .master("local[*]") означает использование всех ядер процессора локально
spark = SparkSession.builder \
    .appName("HelloSpark") \
    .master("local[*]") \
    .getOrCreate()

# 2. Создаем простые данные
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
columns = ["Name", "Age"]

# 3. Создаем DataFrame (распределенную таблицу)
df = spark.createDataFrame(data, columns)

# 4. Показываем содержимое
print("Наша первая таблица в Spark:")
df.show()

# 5. Останавливаем сессию
spark.stop()

Если после запуска вы увидели красивую таблицу в консоли, поздравляю! Вы успешно настроили окружение.

+-------+---+
|   Name|Age|
+-------+---+
|  Alice| 25|
|    Bob| 30|
|Charlie| 35|
+-------+---+

Концепция ленивых вычислений (Lazy Evaluation)

Одной из важнейших особенностей Spark является ленивость. Когда вы говорите Spark «прочитай файл» или «отфильтруй данные», он на самом деле ничего не делает с данными мгновенно. Он просто запоминает план действий (DAG — Directed Acyclic Graph).

Вычисления начинаются только тогда, когда вы требуете результат (действие или Action). Примеры действий:

• show() — показать данные на экране.
• count() — посчитать количество строк.
• write() — сохранить данные на диск.

Это позволяет Spark оптимизировать весь процесс обработки перед запуском, отбрасывая ненужные шаги.

Заключение

Мы познакомились с основами Apache Spark, разобрали его архитектуру и настроили локальное окружение. Теперь у вас есть мощный инструмент, способный обрабатывать терабайты данных прямо на вашем ноутбуке (в рамках объема диска и памяти, конечно) или масштабироваться на тысячи серверов без изменения кода.

В следующей статье мы углубимся в структуру данных DataFrame, научимся читать файлы разных форматов и выполнять базовые операции трансформации данных.

Основы работы с данными: DataFrame API, чтение и запись файлов

В предыдущей лекции мы успешно настроили окружение и запустили наш первый «Hello World» на Spark. Теперь пришло время погрузиться в работу с данными по-настоящему. В мире PySpark основным инструментом, с которым вы будете взаимодействовать 90% времени, является DataFrame.

Если вы пришли из мира Python и библиотеки Pandas, концепция DataFrame вам уже знакома. Однако в Spark это понятие имеет свои уникальные особенности, связанные с распределенной природой вычислений. В этой статье мы разберем, что такое DataFrame в Spark, как правильно читать файлы различных форматов (CSV, JSON, Parquet) и как сохранять результаты вашей работы.

Что такое DataFrame в Spark?

DataFrame — это неизменяемая распределенная коллекция данных, организованная в именованные колонки. Проще говоря, это таблица, которая умеет жить сразу на множестве компьютеров.

В отличие от Pandas DataFrame, который должен полностью помещаться в оперативную память одной машины, Spark DataFrame разбивается на части, называемые партициями (partitions). Каждая партиция обрабатывается отдельным исполнителем (Executor) в кластере.

Ключевые характеристики DataFrame:

1. Распределенность: Данные физически находятся на разных узлах кластера.
2. Неизменяемость (Immutability): Вы не можете изменить существующий DataFrame. Любая операция трансформации (например, добавление колонки) создает новый DataFrame.
3. Ленивость (Lazy Evaluation): Как мы обсуждали ранее, вычисления происходят только в момент вызова действия (Action).
4. Наличие схемы: У каждой колонки есть имя и строгий тип данных (String, Integer, Double и т.д.).

Создание DataFrame вручную

Хотя в реальных задачах мы обычно читаем данные из файлов, для тестов и обучения полезно уметь создавать DataFrame из списка данных.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Инициализация сессии
spark = SparkSession.builder.appName("DataFrameBasics").getOrCreate()

# Данные
data = [
    ("Анна", "Менеджер", 45000),
    ("Борис", "Разработчик", 80000),
    ("Виктор", "Аналитик", 60000)
]

# Определение схемы (рекомендуемый способ)
schema = StructType([
    StructField("Name", StringType(), True),
    StructField("Role", StringType(), True),
    StructField("Salary", IntegerType(), True)
])

# Создание DataFrame
df = spark.createDataFrame(data, schema=schema)

df.show()

Использование явной схемы (StructType) — это хорошая практика. Она позволяет избежать ошибок при автоматическом определении типов и ускоряет процесс создания DataFrame.

Чтение файлов: Reader API

В реальных проектах Big Data данные хранятся в файлах. Spark предоставляет унифицированный интерфейс для чтения данных — DataFrameReader. Общий синтаксис выглядит так:

spark.read.format("формат").option("ключ", "значение").load("путь/к/файлу")

Рассмотрим самые популярные форматы.

1. Чтение CSV

CSV (Comma-Separated Values) — самый распространенный, но не самый эффективный формат. При чтении CSV часто возникают проблемы с типами данных и заголовками.

# Чтение CSV файла
df_csv = spark.read.format("csv") \
    .option("header", "true") \       # Первая строка содержит заголовки
    .option("inferSchema", "true") \  # Попытаться угадать типы данных
    .option("sep", ";") \             # Указать разделитель (если не запятая)
    .load("data/employees.csv")

df_csv.printSchema() # Вывод структуры таблицы

Важно: Опция inferSchema заставляет Spark прочитать файл дважды: первый раз, чтобы понять типы данных, и второй раз — чтобы загрузить данные. На больших объемах (терабайты) это очень дорого. Лучше указывать схему вручную через .schema(my_schema).

2. Чтение JSON

JSON часто используется для полуструктурированных данных (логи, ответы API).

# Чтение JSON
df_json = spark.read.format("json") \
    .option("multiline", "true") \    # Если один объект JSON занимает несколько строк
    .load("data/logs.json")

3. Чтение Parquet

Parquet — это «золотой стандарт» в мире Big Data. Это бинарный колоночный формат хранения данных.

Преимущества Parquet:

• Сжатие: Файлы весят значительно меньше, чем CSV.
• Скорость: Spark читает только нужные колонки, игнорируя остальные.
• Сохранение схемы: Типы данных хранятся внутри файла, их не нужно угадывать.

# Чтение Parquet (формат по умолчанию в Spark)
df_parquet = spark.read.load("data/users.parquet")
# Или явно:
df_parquet = spark.read.parquet("data/users.parquet")

Базовые операции с DataFrame

После загрузки данных мы можем манипулировать ими. Рассмотрим основные методы.

Выбор колонок: select

Метод select позволяет выбрать подмножество колонок или применить к ним выражения.

# Выбрать только имя и зарплату
df.select("Name", "Salary").show()

# Можно использовать выражения (например, увеличить зарплату)
from pyspark.sql.functions import col

# col("Name") позволяет обращаться к колонке как к объекту
df.select(
    col("Name"),
    (col("Salary") * 1.1).alias("New_Salary") # alias задает новое имя
).show()

Фильтрация данных: filter или where

Эти методы работают идентично SQL-оператору WHERE.

# Оставить только аналитиков с зарплатой выше 50000
df.filter((col("Role") == "Аналитик") & (col("Salary") > 50000)).show()

Обратите внимание на синтаксис: условия объединяются через & (И) или | (ИЛИ), и каждое условие должно быть в круглых скобках.

Добавление новых колонок: withColumn

Это один из самых часто используемых методов. Он возвращает новый DataFrame с добавленной или измененной колонкой.

from pyspark.sql.functions import lit

# Добавить колонку с константным значением
df_new = df.withColumn("Company", lit("TechCorp"))

# Создать колонку на основе другой
df_bonus = df.withColumn("Bonus", col("Salary") * 0.2)

Переименование и удаление: withColumnRenamed и drop

# Переименовать колонку
df_renamed = df.withColumnRenamed("Name", "Full_Name")

# Удалить колонку
df_final = df.drop("Salary")

Запись данных: Writer API

После обработки данные нужно сохранить. Для этого используется DataFrameWriter.

Синтаксис:

df.write.format("формат").mode("режим").save("путь")

Режимы записи (mode)

Это критически важный параметр, определяющий поведение, если файл или папка уже существуют.

• append: Добавить данные к существующим.
• overwrite: Удалить существующие данные и записать новые.
• error (или errorIfExists): Выдать ошибку, если данные уже есть (по умолчанию).
• ignore: Если данные есть, ничего не делать и не выдавать ошибку.

Партиционирование при записи

При сохранении больших данных часто используется партиционирование — разбиение данных на папки по значению определенной колонки. Это ускоряет последующее чтение.

# Сохранить данные в формате Parquet, разбив по ролям
df.write \
    .mode("overwrite") \
    .partitionBy("Role") \
    .parquet("output/employees_by_role")

В результате на диске будет создана структура папок:

output/employees_by_role/
    Role=Аналитик/
        part-0001.parquet
    Role=Разработчик/
        part-0002.parquet

Когда вы в следующий раз будете читать эту папку, Spark сможет считывать только нужную папку Role=Аналитик, если вы укажете соответствующий фильтр. Это называется Partition Pruning (отсечение партиций).

Заключение

Сегодня мы изучили фундамент работы с PySpark: DataFrame API. Мы научились создавать таблицы, читать «грязные» CSV и эффективные Parquet файлы, выполнять базовые трансформации и сохранять результаты с учетом партиционирования.

Эти операции — это кирпичики, из которых строятся самые сложные пайплайны обработки данных. В следующей статье мы перейдем к более продвинутым техникам: группировкам, агрегациям и оконным функциям, которые позволят нам извлекать из данных глубокие инсайты.

Продвинутые трансформации и Spark SQL: агрегации, джойны и оконные функции

В предыдущих статьях мы научились настраивать окружение PySpark, читать файлы и выполнять базовые операции с DataFrame, такие как фильтрация и выбор колонок. Однако настоящая сила аналитики больших данных раскрывается, когда мы начинаем комбинировать данные из разных источников, группировать их для получения статистики и анализировать тренды внутри групп.

В этой статье мы перейдем на следующий уровень и разберем три кита обработки данных: агрегации, соединения (Joins) и оконные функции. Также мы посмотрим, как использовать ваши знания SQL прямо внутри PySpark.

Агрегация данных

Агрегация — это процесс объединения нескольких строк в одну на основе какого-либо критерия для вычисления сводных показателей (суммы, среднего, максимума и т.д.).

Группировка с groupBy

Основной метод для агрегации в PySpark — это groupBy(). Он работает аналогично GROUP BY в SQL. После вызова этого метода DataFrame превращается в объект GroupedData, к которому нужно применить агрегирующую функцию.

Предположим, у нас есть данные о продажах:

data = [
    ("Алексей", "Электроника", 1000),
    ("Алексей", "Книги", 500),
    ("Мария", "Электроника", 2000),
    ("Мария", "Одежда", 1500),
    ("Иван", "Книги", 300)
]
columns = ["Salesperson", "Category", "Amount"]
df = spark.createDataFrame(data, columns)

Чтобы узнать общую сумму продаж для каждого продавца:

# Сумма продаж по продавцам
df.groupBy("Salesperson").sum("Amount").show()

Множественные агрегации с agg

Часто нам нужно посчитать сразу несколько метрик: например, сумму продаж, средний чек и количество сделок. Для этого используется метод agg().

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

$$S = \sum_{i=1}^{n} x_i$$

где $S$ — итоговая сумма, $n$ — количество элементов в группе, а $x_i$ — значение $i$-го элемента.

В коде это реализуется следующим образом:

from pyspark.sql.functions import sum, avg, count, max

# Считаем сумму, среднее и максимум за один проход
df.groupBy("Category").agg(
    sum("Amount").alias("Total_Sales"),
    avg("Amount").alias("Avg_Sales"),
    max("Amount").alias("Max_Sale"),
    count("Amount").alias("Tx_Count")
).show()

Использование agg предпочтительнее цепочки вызовов, так как Spark оптимизирует вычисления и проходит по данным минимальное количество раз.

Соединение данных (Joins)

В реальных проектах данные редко лежат в одной таблице. Обычно у нас есть таблица пользователей, таблица заказов, справочник товаров и так далее. Чтобы собрать всё воедино, используются Join (соединения).

Типы соединений в Spark

Синтаксис метода join выглядит так: df1.join(df2, on="key_column", how="type")

Рассмотрим основные типы параметра how:

1. inner (по умолчанию): Оставляет только те строки, ключи которых есть в обеих таблицах.
2. left (или left_outer): Оставляет все строки из левой таблицы и подтягивает совпадения из правой. Если совпадения нет, в колонках правой таблицы будет null.
3. right (или right_outer): Наоборот — все строки из правой таблицы.
4. outer (или full): Оставляет строки, если ключ есть хотя бы в одной из таблиц.

Пример Join

# Таблица сотрудников
employees = spark.createDataFrame([
    (1, "Олег", 10), (2, "Анна", 20), (3, "Денис", 10)
], ["Id", "Name", "DeptId"])

# Таблица департаментов
departments = spark.createDataFrame([
    (10, "IT"), (20, "HR"), (30, "Marketing")
], ["DeptId", "DeptName"])

# Inner Join
employees.join(departments, "DeptId", "inner").show()

Важно: Если имена колонок, по которым идет соединение, различаются (например, emp_dept_id и dept_id), синтаксис меняется: df1.join(df2, df1.emp_dept_id == df2.dept_id, "inner")

Spark SQL: Используем мощь SQL

Одной из киллер-фич Spark является модуль Spark SQL. Он позволяет писать запросы на чистом SQL к вашим DataFrame. Это особенно полезно, если вы мигрируете логику из традиционных баз данных или если аналитикам привычнее писать SQL-код.

Чтобы обращаться к DataFrame через SQL, его нужно зарегистрировать как временное представление (Temp View).

# 1. Регистрируем DataFrame как таблицу "sales"
df.createOrReplaceTempView("sales")

# 2. Пишем SQL запрос
sql_result = spark.sql("""
    SELECT Category, sum(Amount) as Total
    FROM sales
    GROUP BY Category
    HAVING sum(Amount) > 1000
""")

sql_result.show()

Этот код выполняется тем же движком Catalyst Optimizer, что и обычный код на DataFrame API, поэтому производительность будет идентичной.

Оконные функции (Window Functions)

Это, пожалуй, самый сложный, но и самый мощный инструмент в арсенале инженера данных. В отличие от groupBy, который схлопывает строки в одну, оконные функции сохраняют исходное количество строк, но добавляют к каждой строке агрегированную информацию по группе («окну»).

Примеры задач для оконных функций:

• Присвоить ранг сотрудникам по зарплате внутри каждого департамента.
• Посчитать скользящее среднее за последние 3 дня.
• Найти разницу продаж текущего дня с предыдущим.

Структура окна

Для работы с окнами нужно импортировать модуль Window. Окно определяется через спецификацию:

1. partitionBy: Как разбиваем данные на группы (аналог GROUP BY).
2. orderBy: Как сортируем данные внутри группы.
3. rowsBetween (опционально): Границы окна (например, «текущая строка и одна предыдущая»).

Пример: Ранжирование

Допустим, мы хотим узнать, кто продал больше всех в каждой категории, и присвоить им места (1, 2, 3...).

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, desc

# Определяем окно:
# Разбиваем по Категории, сортируем по Сумме убывающе
windowSpec = Window.partitionBy("Category").orderBy(desc("Amount"))

# Применяем функцию row_number() поверх окна
df.withColumn("Rank", row_number().over(windowSpec)).show()

Результат:

Salesperson	Category	Amount	Rank
Алексей	Электроника	1000	2
Мария	Электроника	2000	1
Иван	Книги	300	2
Алексей	Книги	500	1

Lag и Lead

Функции lag (предыдущий) и lead (следующий) позволяют обращаться к данным соседних строк без джойнов с самой собой.

from pyspark.sql.functions import lag

# Окно по времени
timeWindow = Window.orderBy("Date")

# Сравнение с предыдущим днем
df.withColumn("Prev_Day_Sales", lag("Amount", 1).over(timeWindow)) \
  .withColumn("Diff", col("Amount") - col("Prev_Day_Sales"))

Заключение

Мы разобрали продвинутые методы трансформации данных в PySpark. Агрегации позволяют нам видеть общую картину, джойны — обогащать данные, а оконные функции — проводить глубокий аналитический анализ рядов и рейтингов.

Владение этими инструментами переводит вас из разряда новичков в категорию уверенных пользователей Spark. В следующей части курса мы поговорим о том, как делать всё это не только правильно, но и быстро: мы обсудим оптимизацию производительности и кэширование.

Машинное обучение на больших данных с использованием библиотеки MLlib

Мы прошли долгий путь: настроили Spark, научились читать данные, очищать их и проводить сложную аналитику с помощью SQL и оконных функций. Теперь, когда наши данные чисты и структурированы, мы подходим к одной из самых захватывающих тем — машинному обучению (Machine Learning, ML).

В экосистеме Python стандартом де-факто для ML является библиотека Scikit-learn. Она прекрасна, удобна и имеет огромную поддержку. Но у неё есть один фатальный недостаток в контексте Big Data: она работает на одной машине. Если ваш датасет весит 500 ГБ, Scikit-learn просто «упадет» с ошибкой памяти.

Здесь на сцену выходит Spark MLlib (Machine Learning Library). Это библиотека, которая позволяет обучать модели на кластере, распределяя вычисления между множеством серверов.

Особенности MLlib: DataFrame API

Исторически в Spark было две библиотеки для ML:

1. spark.mllib — старая версия, работающая поверх RDD.
2. spark.ml — новая версия, работающая поверх DataFrame.

В этом курсе и в современной разработке мы используем только вторую (spark.ml). Она позволяет использовать все преимущества оптимизатора Catalyst и удобство работы с таблицами, к которому мы привыкли в предыдущих уроках.

Ключевые концепции: Transformer, Estimator и Pipeline

Работа с MLlib строится вокруг трех главных абстракций. Понимание этих терминов критически важно для написания любого ML-кода в Spark.

1. Transformer (Трансформер)

Это алгоритм, который преобразует один DataFrame в другой. Обычно это происходит путем добавления одной или нескольких новых колонок.

• Метод: transform()
• Пример: Вы хотите превратить текстовую колонку «Пол» (М/Ж) в числовую (0/1). Трансформер берет таблицу, читает колонку «Пол» и добавляет колонку «Пол_Индекс».
• Суть: Данные $\rightarrow$ Преобразованные данные.

2. Estimator (Оценщик или Эстиматор)

Это алгоритм, который нужно обучить на данных, чтобы получить модель (которая, в свою очередь, станет Трансформером).

• Метод: fit()
• Пример: Алгоритм Логистической Регрессии. Ему нужно «посмотреть» на ваши данные, найти закономерности и подобрать коэффициенты. После вызова fit() он возвращает обученную модель.
• Суть: Данные $\rightarrow$ Модель.

3. Pipeline (Конвейер)

В реальных задачах процесс ML состоит из множества шагов: заполнить пропуски, закодировать текст, собрать векторы, обучить модель. Pipeline позволяет объединить все эти шаги в одну цепочку. Если вы вызовете fit у Pipeline, он последовательно запустит обработку данных и обучение.

Подготовка данных: Feature Engineering

Главное отличие MLlib от Scikit-learn заключается в формате подачи данных. Scikit-learn принимает на вход матрицу (список списков), где каждая колонка — это признак (feature).

Spark MLlib требует, чтобы все признаки (features) находились в ОДНОЙ колонке в виде вектора.

Для этого используется специальный трансформер — VectorAssembler.

Пример подготовки данных

Допустим, мы хотим предсказать, купит ли клиент подписку (Target), основываясь на его возрасте (Age), доходе (Income) и количестве посещений сайта (Visits).

from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler

spark = SparkSession.builder.appName("MLlibIntro").getOrCreate()

# 1. Создаем тренировочные данные
data = [
    (25, 30000, 10, 0), # Не купил
    (30, 50000, 20, 1), # Купил
    (35, 80000, 5, 1),  # Купил
    (22, 25000, 8, 0)   # Не купил
]
columns = ["Age", "Income", "Visits", "Target"]
df = spark.createDataFrame(data, columns)

# 2. Собираем признаки в один вектор
# inputCols - список колонок, которые станут признаками
# outputCol - имя новой колонки с вектором
assembler = VectorAssembler(
    inputCols=["Age", "Income", "Visits"],
    outputCol="features"
)

# Трансформируем данные
output = assembler.transform(df)

output.select("features", "Target").show(truncate=False)

Результат будет выглядеть так:

+--------------------+------+
|features            |Target|
+--------------------+------+
|[25.0,30000.0,10.0] |0     |
|[30.0,50000.0,20.0] |1     |
|[35.0,80000.0,5.0]  |1     |
|[22.0,25000.0,8.0]  |0     |
+--------------------+------+

Теперь колонка features готова для подачи в алгоритм машинного обучения.

Обучение модели: Логистическая регрессия

Давайте решим задачу классификации. Мы будем использовать Логистическую регрессию. Несмотря на название «регрессия», это алгоритм для классификации (разделения на классы, например, 0 и 1).

Математически модель предсказывает вероятность принадлежности к классу 1 с помощью сигмоидной функции:

$$P(y=1|x) = \frac{1}{1 + e^{-z}}$$

Где:

• $P(y=1|x)$ — вероятность того, что целевая переменная $y$ равна 1 при заданных признаках $x$.
• $e$ — основание натурального логарифма (примерно 2.718).
• $z$ — линейная комбинация признаков и весов (уравнение прямой или гиперплоскости).

В Spark это делается очень просто:

from pyspark.ml.classification import LogisticRegression

# Инициализируем алгоритм
# featuresCol - колонка с вектором признаков (созданная VectorAssembler)
# labelCol - колонка с правильным ответом
lr = LogisticRegression(featuresCol="features", labelCol="Target")

# Обучаем модель (Estimator превращается в Model)
model = lr.fit(output)

# Смотрим на результаты обучения (коэффициенты)
print(f"Коэффициенты: {model.coefficients}")
print(f"Свободный член (Intercept): {model.intercept}")

Полный цикл с использованием Pipeline

Теперь соберем всё вместе в правильный пайплайн, разделив данные на обучающую и тестовую выборки. Это «золотой стандарт» написания кода на PySpark.

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# 1. Подготовка данных (как раньше)
data = [
    (25, 30000, 0), (30, 50000, 1), (35, 80000, 1), (22, 25000, 0),
    (40, 100000, 1), (19, 15000, 0), (50, 40000, 0), (28, 60000, 1)
]
df = spark.createDataFrame(data, ["Age", "Income", "Target"])

# 2. Разделение на Train (80%) и Test (20%)
train_data, test_data = df.randomSplit([0.8, 0.2], seed=42)

# 3. Определение этапов Pipeline
# Этап 1: Сборка векторов
assembler = VectorAssembler(inputCols=["Age", "Income"], outputCol="features")

# Этап 2: Модель
lr = LogisticRegression(featuresCol="features", labelCol="Target")

# 4. Создание Pipeline
pipeline = Pipeline(stages=[assembler, lr])

# 5. Обучение (Fit) на тренировочных данных
# Pipeline.fit() запускает fit() для всех Estimator'ов внутри
model = pipeline.fit(train_data)

# 6. Предсказание (Transform) на тестовых данных
# PipelineModel.transform() запускает transform() для всех этапов
predictions = model.transform(test_data)

# Смотрим, что получилось
predictions.select("Age", "Income", "Target", "prediction", "probability").show()

В результате в DataFrame predictions появятся новые колонки:

• rawPrediction: «сырая» оценка модели.
• probability: вероятность для каждого класса (например, [0.1, 0.9] означает 90% уверенности в классе 1).
• prediction: итоговый класс (0.0 или 1.0).

Оценка качества модели

Чтобы понять, насколько хорошо работает наша модель, нам нужны метрики. Для бинарной классификации часто используют AUC-ROC (Area Under ROC Curve).

# Оценщик для бинарной классификации
evaluator = BinaryClassificationEvaluator(
    rawPredictionCol="rawPrediction",
    labelCol="Target",
    metricName="areaUnderROC"
)

auc = evaluator.evaluate(predictions)
print(f"Area Under ROC: {auc}")

Если auc равен 1.0 — модель идеальна. Если 0.5 — модель гадает случайно.

Работа с категориальными признаками

В примере выше мы использовали только числа (Возраст, Доход). Но что делать, если у нас есть текст, например, «Город» (Москва, СПб, Казань)? Алгоритмы ML понимают только числа.

Для этого в Spark есть специальные трансформеры:

1. StringIndexer: Превращает строки в числовые индексы (Москва $\rightarrow$ 0, СПб $\rightarrow$ 1).
2. OneHotEncoder: Превращает индексы в бинарные векторы (0 $\rightarrow$ [1, 0, 0]).

Их нужно просто добавить в stages вашего Pipeline перед VectorAssembler.

Заключение

Сегодня мы сделали огромный шаг вперед. Мы разобрали основы библиотеки MLlib, поняли разницу между Transformer и Estimator, научились собирать признаки с помощью VectorAssembler и обучили нашу первую распределенную модель классификации.

Spark MLlib позволяет вам масштабировать этот процесс. Код, который мы написали для 8 строк данных, без изменений сработает и для 8 миллиардов строк — просто добавьте больше серверов в кластер.

В следующих материалах курса мы можем углубиться в настройку гиперпараметров и сохранение моделей для использования в продакшене.

Оптимизация производительности, тюнинг приложений и лучшие практики разработки

Поздравляю! Вы прошли путь от установки Spark до создания моделей машинного обучения. Ваш код работает, данные обрабатываются, и модели обучаются. Но в мире Big Data «код работает» — это только половина успеха. Вторая половина — «код работает быстро и не падает с ошибкой OutOfMemory».

Часто бывает так: на тестовом файле в 100 МБ всё летает, а при запуске на реальном терабайтном датасете приложение зависает на часы или падает. В этой статье мы разберем, почему так происходит, что такое Shuffle, как бороться с перекосом данных (Skew) и какие лучшие практики помогут вам писать эффективный PySpark-код.

Понимание механизма выполнения: Jobs, Stages и Tasks

Чтобы оптимизировать Spark, нужно понимать, как он думает. Как мы обсуждали ранее, Spark ленив. Он строит план выполнения (DAG), который превращается в физические действия только при вызове Action (например, count() или write()).

Весь процесс обработки делится на иерархию:

1. Job (Работа): Создается при вызове Action.
2. Stage (Этап): Job делится на этапы. Границы этапов определяются операциями, требующими перемещения данных между серверами (Shuffle).
3. Task (Задача): Самая мелкая единица работы. Один этап состоит из множества задач, каждая из которых обрабатывает одну партицию данных.

Главный враг производительности — это Shuffle (перемешивание). Это процесс, когда данные пересылаются по сети между узлами кластера для группировки или соединения. Это дорого, долго и забивает сеть.

Кэширование данных: cache() и persist()

Одна из самых частых ошибок новичков — непонимание того, когда нужно кэшировать данные. Поскольку Spark вычисляет всё «на лету», если вы используете один и тот же DataFrame в двух разных действиях, Spark может прочитать и обработать исходный файл дважды.

Чтобы избежать этого, используются методы cache() или persist().

Когда использовать кэширование?

• Если вы используете один и тот же DataFrame несколько раз (например, в алгоритмах машинного обучения или при ветвлении логики).
• После тяжелых трансформаций, чтобы не пересчитывать их заново при сбое.

В чем разница?

• cache(): Сохраняет данные в оперативной памяти (Storage Level: MEMORY_AND_DISK). Если памяти не хватит, часть данных не будет закэширована.
• persist(level): Позволяет гибко выбрать уровень хранения (только диск, только память, память + диск и т.д.).

Лучшая практика: Не кэшируйте всё подряд. Кэширование занимает оперативную память, которой может не хватить для самих вычислений. Всегда делайте unpersist(), когда данные больше не нужны.

Проблема перекоса данных (Data Skew)

Data Skew — это ситуация, когда данные распределены по партициям неравномерно. Представьте, что вы распределили 100 коробок между 10 грузчиками. Но одному грузчику досталось 91 коробка, а остальным девяти — по одной. В итоге 9 человек закончат работу за минуту и будут ждать, пока десятый будет работать час.

В Spark это выглядит так: 199 задач завершились за 5 секунд, а одна задача висит 30 минут.

Как обнаружить?

Зайдите в Spark UI (веб-интерфейс мониторинга). Если вы видите, что в этапе (Stage) максимальное время выполнения задачи (Max Duration) в разы превышает медианное, у вас перекос.

Как лечить? Техника «Salting» (Подсаливание)

Если перекос возникает при Join или GroupBy по популярному ключу (например, null или «Москва»), можно искусственно разбить этот ключ.

Математически это можно представить так. Мы добавляем к ключу случайное число (соль).

$$K_{new} = K_{orig} + (rand() \pmod N)$$

Где:

• $K_{new}$ — новый ключ с добавленной солью.
• $K_{orig}$ — исходный ключ, вызывающий перекос.
• $rand()$ — функция генерации случайного числа.
• $\pmod N$ — операция взятия остатка от деления на $N$ (количество частей, на которые мы хотим разбить ключ).

После агрегации по $K_{new}$ мы делаем повторную агрегацию, убирая соль, чтобы получить верный результат.

Оптимизация соединений (Joins)

Соединение таблиц — самая тяжелая операция. Spark поддерживает несколько стратегий, но вам нужно знать две основные.

1. Sort Merge Join (SMJ)

Стандартный метод для соединения двух больших таблиц. Данные сортируются и перемешиваются (Shuffle) по ключам. Это надежно, но медленно.

2. Broadcast Hash Join (BHJ)

Это «секретное оружие» оптимизации. Если одна из таблиц маленькая (справочник), Spark может отправить её копию на каждый узел кластера.

Преимущества:

• Нет Shuffle большой таблицы.
• Огромный прирост скорости.

Spark пытается делать это автоматически, если размер таблицы меньше spark.sql.autoBroadcastJoinThreshold (по умолчанию 10 МБ). Вы можете форсировать это в коде:

from pyspark.sql.functions import broadcast

# Принудительно используем Broadcast Join для маленькой таблицы df_small
big_df.join(broadcast(small_df), "key").show()

Проблема маленьких файлов (Small Files Problem)

В Big Data большие файлы — это хорошо, а много маленьких — плохо. Открытие каждого файла занимает время. Если у вас 10 000 файлов по 1 КБ, чтение будет идти вечность.

Это часто случается, если вы сохраняете DataFrame с большим количеством партиций.

Решение: coalesce() и repartition()

Перед записью данных на диск контролируйте количество файлов.

• repartition(n): Полное перемешивание данных (Shuffle) для создания $n$ равных партиций. Используйте, чтобы увеличить параллелизм или равномерно распределить данные.
• coalesce(n): Уменьшение количества партиций без Shuffle (объединяет соседние). Используйте перед сохранением, чтобы уменьшить число файлов.

# Плохо: сохранит 200 маленьких файлов (по умолчанию shuffle partitions = 200)
df.write.parquet("/data/output")

# Хорошо: склеит данные в 5 файлов перед записью
df.coalesce(5).write.parquet("/data/output")

User Defined Functions (UDF): Убийцы производительности

Python — прекрасный язык, но Spark написан на Scala (JVM). Когда вы используете стандартную Python-функцию как UDF, происходит следующее:

1. Spark берет данные из JVM.
2. Сериализует их (превращает в байты).
3. Передает в процесс Python.
4. Python выполняет функцию.
5. Результат сериализуется и возвращается в JVM.

Эти накладные расходы могут замедлить код в 10-100 раз.

Как избежать?

1. Используйте встроенные функции Spark SQL: В модуле pyspark.sql.functions есть почти всё, что нужно (работа со строками, датами, математика). Они работают внутри JVM и максимально оптимизированы.
2. Pandas UDF (Vectorized UDF): Если встроенных функций не хватает, используйте Pandas UDF. Они используют формат Apache Arrow для передачи данных пачками (векторами), что значительно быстрее обычной построчной обработки.

Форматы файлов

Выбор формата хранения критически влияет на скорость чтения.

• CSV/JSON: Текстовые форматы. Медленные, тяжелые, не хранят схему. Используйте только для обмена данными с внешними системами.
• Parquet/ORC: Бинарные колоночные форматы. Поддерживают сжатие, хранят схему и позволяют Spark читать только нужные колонки (Column Pruning). Всегда используйте их для промежуточного хранения.

Чек-лист лучших практик

Подводя итог, вот краткий список правил для написания быстрого кода на PySpark:

1. Избегайте collect(): Никогда не вызывайте collect() на больших данных. Это пытается скачать всё на драйвер и убивает его.
2. Фильтруйте рано: Применяйте filter() или where() как можно раньше, до джойнов. Чем меньше данных идет дальше, тем быстрее.
3. Используйте Broadcast Join: Для соединения большой таблицы с маленькой.
4. Предпочитайте встроенные функции: Избегайте Python UDF, если это возможно.
5. Следите за партициями: Используйте coalesce() перед записью, чтобы не плодить мелкие файлы.
6. Кэшируйте с умом: Только если DataFrame используется многократно.
7. Смотрите в UI: Если задача висит, Spark UI подскажет, где именно (Skew, Shuffle, Input/Output).

Оптимизация — это искусство баланса между ресурсами кластера (CPU, RAM, Сеть) и временем выполнения. Следуя этим советам, вы сможете создавать надежные и быстрые пайплайны обработки данных.

На этом наш курс по основам PySpark завершен. Вы прошли путь от новичка до инженера, способного решать задачи Big Data и ML. Удачи в ваших проектах!