PySparkbig dataData Engineering

PySpark для аналитика: основы на понятных примерах

2026-07-11 11 мин

pandas ломается, когда таблица перестаёт помещаться в оперативку одной машины — и на практике это происходит раньше, чем данные достигают объёма RAM, потому что многие операции (merge, groupby, pivot) под капотом делают копии. Как только вы читаете файл на десятки гигабайт и видите MemoryError или машина уходит в своп, пора брать инструмент, который считает данные кусками сразу на нескольких машинах. Это Apache Spark, а PySpark — его Python-интерфейс. Ниже разберу без академизма: когда он реально нужен, как устроен DataFrame API, почему Spark ленивый, что такое partitions и shuffle, и где всё это спрашивают на собеседованиях.

Когда pandas упирается в память и пора звать Spark?

Простое правило: если данные помещаются в память одной машины с запасом — оставайтесь на pandas. Он проще, быстрее на малых объёмах и не требует кластера. Spark нужен, когда одного из двух не хватает: памяти или процессора.

Ориентиры (всё зависит от железа и характера операций):

Важный нюанс: pandas держит весь датафрейм в RAM целиком. Spark читает и обрабатывает данные партициями (кусками), поэтому может прожевать датасет больше, чем суммарная память кластера — просто не держит всё сразу. Плата за это — накладные расходы. На маленьких данных Spark будет медленнее pandas: время уходит на запуск JVM, планирование и пересылку данных между узлами. Не тащите Spark туда, где хватает pandas — это классическая ошибка джуна.

# pandas: всё в памяти одной машины
import pandas as pd
orders = pd.read_csv("orders.csv")        # 40 ГБ? MemoryError
big = orders.groupby("user_id")["amount"].sum()

Что такое Spark и как он устроен под капотом?

Spark — это движок распределённых вычислений. Вместо одной машины у вас кластер: один driver (координатор, где выполняется ваш код и живёт SparkSession) и несколько executors (рабочие процессы, которые реально считают куски данных). Driver разбивает задачу на части, раскидывает по executors, собирает результат.

Данные внутри Spark разбиты на partitions — независимые куски, каждый обрабатывается отдельной задачей на отдельном ядре. Именно партиционирование даёт параллелизм: 200 партиций на 50 ядрах считаются пачками по 50 штук.

Стартовая точка любого кода на PySpark — сессия:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

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

orders = spark.read.parquet("s3a://data/orders/")   # ничего ещё не прочитано

Обратите внимание: в комментарии написано «ничего ещё не прочитано». Это не оговорка — Spark действительно ленивый, и это ключевая вещь для понимания.

Как выглядит DataFrame API: select, filter, groupBy, join?

DataFrame в Spark — это таблица со схемой (колонки и типы), концептуально как в pandas или как таблица в базе. Но API ближе к SQL: вы описываете, что хотите получить, цепочкой методов.

Выбор колонок и фильтрация:

result = (
    orders
    .select("user_id", "amount", "created_at")
    .filter(F.col("amount") > 1000)
)

Агрегация — groupBy + agg:

by_user = (
    orders
    .groupBy("user_id")
    .agg(
        F.sum("amount").alias("revenue"),
        F.count("*").alias("orders_cnt"),
    )
)

Джойн двух таблиц:

users = spark.read.parquet("s3a://data/users/")

enriched = orders.join(users, on="user_id", how="left")

Новые колонки — withColumn:

orders = orders.withColumn(
    "is_big", F.when(F.col("amount") > 5000, 1).otherwise(0)
)

Синтаксис почти самодокументируемый: select, filter, groupBy, join делают ровно то, что написано. Если вы уверенно пишете такие цепочки на SQL, переучиваться на DataFrame API — вопрос пары дней. Погонять базовые агрегации и джойны без Spark, чтобы набить руку на самой логике, удобно в SQL-тренажёре, а разбор оконных функций и группировок есть в SQL-справочнике.

Кстати, если DataFrame API кажется многословным, у Spark есть чистый SQL-режим:

orders.createOrReplaceTempView("orders")
spark.sql("""
    SELECT user_id, SUM(amount) AS revenue
    FROM orders
    WHERE amount > 1000
    GROUP BY user_id
""")

Результат тот же — под капотом оба варианта компилируются в один план выполнения.

Почему Spark ничего не считает, пока вы не попросите?

Потому что операции делятся на два типа: трансформации и действия. Трансформация (select, filter, groupBy, join, withColumn) ничего не выполняет — она только дописывает шаг в план. Действие (show, count, collect, write) запускает реальное вычисление всей накопленной цепочки.

Это называется lazy evaluation, ленивые вычисления. Когда вы пишете пять трансформаций подряд, Spark не бегает к данным пять раз. Он копит план и, дождавшись действия, оптимизирует его целиком: переставляет фильтры ближе к чтению (чтобы прочитать меньше), выкидывает неиспользуемые колонки, склеивает шаги. За это отвечает оптимизатор Catalyst.

df = spark.read.parquet("s3a://data/events/")   # трансформация — ничего не считается
df = df.filter(F.col("event") == "purchase")     # трансформация
df = df.select("user_id", "ts")                  # трансформация
df.show(10)                                       # ДЕЙСТВИЕ — вот теперь Spark работает

Практическое следствие, на котором спотыкаются новички: пока вы не вызвали действие, ошибок можно не увидеть. Вы написали кривой фильтр, запустили ячейку — тишина, всё «отработало» мгновенно. А MemoryError или AnalysisException прилетает на show() через десять ячеек ниже. Так что держите в голове, где именно у вас действие — там и происходит вся работа.

Мини-шпаргалка:

Отдельно про collect() и toPandas() — они стягивают ВЕСЬ результат на driver, в память одной машины. На большом датафрейме это тот самый MemoryError, от которого вы убегали. Стягивайте на driver только заведомо маленькие агрегаты.

Что такое partitions и почему shuffle такой дорогой?

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

Shuffle — это перераспределение данных по сети между executors. Он неизбежен, когда результат зависит от строк, которые лежат на разных машинах. Классические виновники: groupBy, join, distinct, orderBy. Чтобы сгруппировать по user_id, Spark должен собрать все строки одного пользователя в одну партицию — а они изначально разбросаны по всему кластеру.

Почему это дорого:

Отдельная боль — перекос (data skew). Если у одного user_id миллион строк, а у остальных по десять, то одна партиция окажется гигантской, и весь джоб будет ждать эту единственную задачу. Так бывает с «мусорными» ключами вроде user_id = 0, NULL, дефолтных значений или служебных аккаунтов, через которые проходит половина событий.

Практический вывод для аналитика: фильтруйте раньше, чем группируете и джойните. Сначала filter, потом groupBy — Catalyst и сам попробует так сделать, но помогать ему стоит. Чем меньше данных доедет до shuffle, тем дешевле джоб.

# лучше: сузили данные ДО тяжёлой перетасовки
(
    events
    .filter(F.col("event") == "purchase")   # режем объём заранее
    .groupBy("user_id")                       # shuffle уже по меньшим данным
    .agg(F.sum("amount"))
)

Чем PySpark отличается от SQL и pandas по синтаксису?

На практике вы будете думать на одном языке, а писать на трёх. Полезно держать в голове соответствия. Возьмём одну задачу — выручка по пользователям с фильтром — в трёх вариантах.

-- SQL
SELECT user_id, SUM(amount) AS revenue
FROM orders
WHERE amount > 1000
GROUP BY user_id
# pandas
(orders[orders["amount"] > 1000]
    .groupby("user_id")["amount"].sum()
    .reset_index(name="revenue"))
# PySpark
(orders
    .filter(F.col("amount") > 1000)
    .groupBy("user_id")
    .agg(F.sum("amount").alias("revenue")))

Что стоит запомнить про различия:

Если pandas у вас пока шаткий, есть смысл сначала закрепить его отдельно — в Python-тренажёре и по Python-справочнику. PySpark ляжет гораздо легче на уверенный pandas плюс уверенный SQL.

Где PySpark встречается на собеседованиях Data Engineer?

На собесах DE и на стыке DE/аналитики PySpark — почти обязательный пункт, как только в вакансии есть слова «большие данные» или «пайплайны». Что спрашивают чаще всего (набор типовой, конкретный вес зависит от компании):

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

С чего начать, чтобы не утонуть?

Не начинайте с настройки кластера — это отпугивает. PySpark ставится как обычный пакет и поднимает локальный «кластер» из ядер вашей машины: pip install pyspark, дальше SparkSession.builder.getOrCreate(). Этого достаточно, чтобы прочувствовать DataFrame API, ленивость и разницу между show() и count() на игрушечных данных.

Дальше — по нарастающей. Сначала перепишите на PySpark пару своих обычных pandas-скриптов: те же filter, groupBy, join, только на новом API. Потом разберитесь, где в вашем коде происходит shuffle и можно ли его сократить. И только когда синтаксис перестанет мешать думать, лезьте в тюнинг партиций и broadcast-join — на собесе про них спросят, но фундамент важнее.

Порядок освоения, который реально работает: уверенный SQL → уверенный pandas → PySpark как надстройка над обоими. Если первые два шага крепкие, третий занимает недели, а не месяцы. А базовые метрики, которые вы будете считать этими джойнами и группировками, — те же самые: DAU, retention, выручка на пользователя.

Если хотите один маршрут вместо десятка вкладок — в Pro-подписке собраны SQL- и Python-тренажёры, разборы кейсов и конспекты (включая ноутбук по PySpark), плюс AI-собеседования без лимита. Но и бесплатной части хватит, чтобы начать переходить с pandas на Spark осознанно.

Отработай SQL на практике
545 SQL-задач с автопроверкой в браузере — первые открыты без регистрации.
SQL-тренажёр →

Похожие статьи