Apache Spark

1. Что такое Spark

Spark делит задачу на партиции, распределяет их по исполнителям (executors) и обрабатывает параллельно. Компоненты:

• Spark Core — планировщик задач, RDD.
• Spark SQL — DataFrame и SQL-запросы.
• MLlib — машинное обучение на кластере.
• Structured Streaming — потоковая обработка.

2. Установка и запуск

pip install pyspark

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName('demo').master('local[*]') \
    .getOrCreate()

local[*] запускает на одном узле со всеми ядрами — для обучения достаточно.

3. Создание DataFrame

df = spark.read.csv('sales.csv', header=True, inferSchema=True)
df2 = spark.read.json('events/*.json')
df3 = spark.read.parquet('data.parquet')

df.show(5)
df.printSchema()

В отличие от Pandas, DataFrame ленивый: вычисления запускаются при действий (show, count, write).

4. Колонки и трансформации

from pyspark.sql import functions as F

df = df.withColumn('revenue', F.col('price') * F.col('qty'))
df = df.filter(F.col('revenue') > 100)
df = df.select('product', 'revenue')
df = df.dropDuplicates(['user_id'])
df.show()

5. Агрегации

total = df.groupBy('product').agg(
    F.sum('revenue').alias('total'),
    F.count('*').alias('n')
).orderBy(F.col('total').desc())
total.show()

Работает как GROUP BY из SQL, только по кластеру.

6. Spark SQL

df.createOrReplaceTempView('sales')

results = spark.sql('''
    SELECT product, SUM(revenue) AS total
    FROM sales WHERE revenue > 100
    GROUP BY product
''')

Пишите SQL прямо в Spark, если привычнее — движок тот же.

7. RDD vs DataFrame

RDD — низкоуровневое API (Java/Scala корни). DataFrame построен поверх и оптимизирован Catalyst-планировщиком: всегда выбирай DataFrame, если не нужна специфика RDD.

8. Интеграции и коннекторы

• Kafka: потоковое чтение spark.readStream.format('kafka').
• ClickHouse, JDBC, S3, Parquet/Delta.
• Hive Metastore — каталог таблиц (см. Hive).

9. Потоковая обработка

stream_df = spark.readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', 'localhost:9092') \
    .option('subscribe', 'events') \
    .load()

query = stream_df \
    .selectExpr('CAST(value AS STRING)') \
    .writeStream.outputMode('append').format('console') \
    .start()

query.awaitTermination(timeout=30)

10. Кластер

• Driver — планировщик, executors — рабочие.
• Ресурсы: executor cores/memory, shuffle partitions.
• Деплой: Standalone, YARN, Kubernetes (K8s).

11. Лучшие практики

• Избегай collect() больших данных в драйвер — вернётся в память и взорвётся.
• Кэшируй повторно используемые DataFrame: df.cache().
• Пиши в Parquet с разделением (partitionBy) — ускоряет чтения дальше.

12. RDD, DataFrame и Datasets

Spark предоставляет три API: низкоуровневые RDD, современные DataFrame и типизированные Datasets. На практике 95% работы идёт через DataFrame — оно оптимизируется планировщиком (Catalyst) автоматически.

df = spark.read.parquet("events.parquet")
(# планировщик сам сдвинет фильтры вниз и объединит чтения
df.filter(col("region") == "RU")
  .groupBy("product")
  .agg(sum("revenue").alias("total"))
  .show())

13. Оптимизация задач: партиции и broadcast

Классические причины медленных джобов — неравномерные партиции (shuffle-перекос), слишком большие партиции и большие таблицы, которые копируются в каждый shuffle.

# добавить соль против перекоса ключа
df.withColumn("salt", (rand() * 10).cast("int"))
  .groupBy("key", "salt").agg(...)

# маленькая справочная таблица — в broadcast
from pyspark.sql.functions import broadcast
joined = df.join(broadcast(lookup), "id", "left")

• spark.sql.shuffle.partitions — число партиций shuffle (обычно 200);
• для cache()/persist() выбирай StorageLevel по памяти и диску;
• контролируй узкие места через Spark UI → SQL → View Timeline.

14. Обработка ошибок и контрольные точки

Долгие ETL-джобы нужно делать устойчивыми к сбоям: правильно завершать работу, логировать и уметь возобновляться с места остановки.

try:
    (df.write.mode("overwrite")
       .partitionBy("date")
       .parquet("warehouse/events"))
except Exception as e:
    log.error(f"Запись упала: {e}")
    spark.stop()
    raise

• контрольные точки (checkpoint()) обрезают line age — обязательны в стриминге;
• пиши в staging-каталог, затем атомарно переименовывай в финальный;
• мониторь через Grafana/Prometheus: затраченное время, размер shuffles, ошибки.

15. Spark в продакшене: практический сценарий

Соберём типовой конвейер «сырые события → витрина для BI» на одном кластере.

1. чтение сырья из Kafka или S3;
2. очистка: фильтры, дедупликация, приведение типов;
3. партиционирование по дате и запись в Parquet;
4. агрегация витрины: продуктовые метрики по дням;
5. передача в ClickHouse/PostgreSQL для дашбордов;
6. расписание через cron или Airflow с повторными попытками.

spark-submit \
  --master yarn \
  --num-executors 8 \
  --executor-memory 4g \
  --conf spark.sql.shuffle.partitions=400 \
  etl_events.py

16. Catalyst-оптимизация: планы и explain

Spark строит план выполнения, который проходит через оптимизатор Catalyst: правило переписывает ваш DataFrame в оптимальный набор операций. Важно уметь читать план и находить «дорогие» шаги.

df = spark.read.parquet("events/")
query = (df.filter(col("type") == "click")
          .groupBy("day").count())

query.explain(True)      # физический план
query.explain("cost")    # с оценкой стоимости

• explain(True) показывает логический и физический план;
• ищите Exchange (shuffle) — самый дорогой шаг;
• predicate pushdown: фильтры «опускаются» к источнику данных;
• проекция: лишние колонки режутся раньше;
• понимание плана — ключ к ускорению джобов.

17. Структурированный стриминг: детали

Структурированный стриминг обрабатывает данные в микро-батчах или continuously: события как бесконечная таблица, к которой применяются ваши трансформации. Ключ — надёжность и водяные знаки.

from pyspark.sql.functions import window

events = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "clicks")
    .load())

windowed = (events
    .groupBy(window("timestamp", "1 minute"),
             "user_id")
    .count())

query = (windowed.writeStream
    .outputMode("update")
    .format("console")
    .start())

• окна (tumbling/sliding) реально востребованны в аналитике;
• withWatermark — защита от «бесконечной» памяти;
• outputMode: append, update, complete — выбор по типу результатов;
• checkpointLocation обязателен для восстановления после сбоя.

18. Форматы файлов: Parquet, Avro, ORC

Выбор формата определяет скорость, объём и совместимость. Parquet — колоночный стандарт аналитики, Avro — строчный для потоков, ORC — «сорт» для Hive-хранилищ.

• для данных с редким чтением колонок — Parquet/ORC;
• для строчных обновлений и потоков — Avro;
• spark.read.parquet("…"), .format("avro"), .orc;
• компрессия: snappy (скорость), gzip (размер).

19. Spark UI и дебаг джобов

Spark UI (порт 4040) — главный инструмент диагностики: стадии, задачи, shuffle, кэшированные RDD, SQL-планы. Научиться читать его — половина успеха в производительных ETL.

1. откройте вкладку SQL: увидите реальный план и счётчики, где Spark тратит время;
2. Jobs: сколько «лишних» действий запускается из-за повторных действий;
3. Stage: длительность, число задач, распределение (скью на задачу);
4. Storage: что в кэше и не «раздулся» ли;
5. Executors: память и GC каждого воркера.

Признаки проблем:
⌛ Long Job → неэффективный план или мало узлов
⚙️ Large Shuffle → перекос по ключу
📈 High GC → мало памяти на кэш или партиции

20. Память и настройка исполнителей

Частые проблемы Spark — GC-паузы и вытеснение кэша. Управление ядрами, памятью и сериализацией — ключ к стабильной работе джобов на больших данных.

# пример конфигурации
spark.executor.memory      8g
spark.executor.cores       4
spark.memory.offHeap.enabled  true
spark.sql.adaptive.enabled    true

• размер исполнителя подбирайте под ресурсы узла: память и ядра должны «уживаться»;
• spark.sql.adaptive (AQE) динамически перепланирует задачи;
• Kryo-сериализация сокращает объём shuffle и нагрузку на GC;
• кэшируйте только многократно используемые данные.

21. Delta Lake: транзакции поверх данных

Delta Lake добавляет к Spark-таблицам свойства базы данных: транзакции, версии, обновления. Он устраняет главную боль озёр данных — частично записанные файлы и несогласованные чтения.

# upsert через MERGE
MERGE INTO events t USING updates s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.value = s.value
WHEN NOT MATCHED THEN INSERT *

• ACID-транзакции на файлах Parquet: чтение никогда не видит «полузапись»;
• time travel: чтение исторических версий таблицы без бэкапов;
• upsert и delete стандартными операциями SQL;
• streaming и batch работают с одним и тем же слоем данных.

22. MLlib: машинное обучение на Spark

MLlib — библиотека машинного обучения для распределённых объёмов: подготовка признаков, обучение моделей и оценка выполняются на кластере. Это стандарт для ML с миллионами строк.

• VectorAssembler и StringIndexer готовят признаки для моделей;
• алгоритмы: регрессии, деревья, кластеризация k-means, рекомендательные (ALS);
• Pipeline объединяет подготовку и обучение в один конвейер;
• конвейеры сохраняются в файл и переиспользуются для предсказаний.

23. Мониторинг джобов и тонкая настройка

Spark UI и журналы дают полную картину каждого джоба: стадии, шuffle, память, время. Умение читать эти данные быстрее находит узкое место, чем догадки.

1. смотрите Jobs: время каждой стадии и её длительность;
2. проверяйте Stage Details: shuffle read/write и баланс партиций;
3. анализируйте Executors: использование памяти и параллельность;
4. ищите перекос: пара партиций «тянет» весь джоб.

Что дальше

Данные для Spark часто приходят из Kafka; храни и анализируй результаты в ClickHouse, смотри в Pandas на малых объёмах.

← Вернуться к курсам