Apache Spark: основы

Декабрь 2012 года: команда AmpLab UC Berkeley публикует Spark 0.6. Задача - обработать 1 петабайт данных на 100 нодах. Hadoop MapReduce справляется за 72 часа. Spark - за 23 минуты. Секрет: данные в памяти, не на диске. Lineage вместо репликации. Lazy evaluation вместо немедленного выполнения. Сегодня Apache Spark обрабатывает exabytes данных ежедневно в Alibaba, Netflix, Apple и тысячах других компаний.

• **Netflix**: Spark обрабатывает 500+ миллиардов событий в день для A/B тестирования и рекомендаций
• **Alibaba**: Spark SQL заменил Hive на крупнейшей в мире e-commerce платформе - 10x ускорение ETL
• **Apple**: Spark ML pipeline для Siri NLU - обработка петабайт логов для тренировки моделей

Матей Захария (2009)

В 2009 году PhD студент Матей Захария начал работу над Spark как исследовательским проектом в AmpLab UC Berkeley. Проблема Hadoop MapReduce - каждый шаг pipeline записывает результат на HDFS. Iterative алгоритмы (ML, граф) делают десятки roundtrip'ов диска. Матей предложил концепцию RDD: хранить данные в памяти, восстанавливать через lineage при отказах. В 2010 вышла статья 'Spark: Cluster Computing with Working Sets'. В 2013 проект передан Apache Software Foundation. Матей стал CEO Databricks - компании с оценкой $43 миллиарда на 2024 год. Apache Spark стал de facto стандартом для большие данные обработки, вытеснив Hadoop MapReduce из большинства production систем

RDD: Resilient Distributed Dataset

RDD (Resilient Distributed Dataset) - фундаментальная абстракция Spark. Неизменяемая распределённая коллекция объектов, разбитая на партиции по нодам кластера. **Resilient**: при отказе ноды Spark восстанавливает данные через lineage граф (цепочку трансформаций), не репликацию. Это ключевое отличие от Hadoop MapReduce, который всегда записывает на диск.

RDD хранит **lineage** - граф всех трансформаций от источника. При потере партиции Spark не читает из репликации (как в HDFS), а перевычисляет только утраченные данные по lineage. Для длинных цепочек трансформаций стоит использовать `persist()` - кешировать промежуточный результат.

RDD работает с Java/Python объектами без schema - нет оптимизации через Catalyst. В 2025 году прямое использование RDD оправдано только для нестандартных операций, недоступных в DataFrame API. В остальных случаях DataFrame на 2-10x быстрее.

DataFrame и Catalyst Optimizer

DataFrame - структурированная абстракция поверх RDD с именованными типизированными колонками (schema). Выполнение проходит через **Catalyst Optimizer**: анализ -> логический план -> оптимизация (predicate pushdown, column pruning) -> физический план -> codegen. Одинаковый DataFrame код работает через Python API с производительностью JVM кода.

**Predicate pushdown**: фильтры по колонкам Parquet выполняются при чтении, не в памяти. **Column pruning**: читаются только нужные колонки. **Join reordering**: Catalyst выбирает порядок join'ов по статистике. Эти оптимизации недоступны при работе с RDD напрямую.

Трансформации: lazy evaluation и DAG

Трансформации в Spark **ленивые** (lazy): вызов `filter()`, `map()`, `groupBy()` не запускает вычисления - только строит DAG (Directed Acyclic Graph) операций. Spark откладывает выполнение до момента вызова action. Это позволяет Catalyst оптимизировать весь план перед выполнением.

**Narrow трансформации** (filter, map, select): каждая партиция обрабатывается независимо - нет сетевого трафика. **Wide трансформации** (groupBy, join, repartition): данные перемещаются между нодами (shuffle) - дорогая операция. Оптимизация: минимизировать shuffle, использовать broadcast join для малых таблиц.

Actions: запуск выполнения DAG

Action - операция, которая запускает выполнение накопленного DAG трансформаций и возвращает результат в driver или записывает на хранилище. В отличие от трансформаций, каждый action запускает новый Spark job. Понимание этого критично для оптимизации: лишние actions - лишние job'ы.

Правило: **один логический pipeline - один action записи**. Частая ошибка: `count()` + `write()` вычисляют DAG дважды. Решение: `write()` один раз, потом проверять результат через метаданные хранилища. Если нужны несколько результатов - вычислить один раз, кешировать, применить несколько actions.

Ключевые идеи

• **RDD** - неизменяемая распределённая коллекция с lineage для восстановления без репликации данных
• **DataFrame** + Catalyst Optimizer: predicate pushdown, column pruning, codegen дают 2-10x ускорение vs RDD
• **Трансформации** ленивые (narrow/wide): narrow не требуют shuffle, wide перемещают данные между нодами
• **Actions** запускают DAG выполнение: лишние actions = лишние job'ы, кешировать перед несколькими actions

Вопросы для размышления

• Когда lineage восстановление RDD менее надёжно, чем репликация HDFS, и как `persist()` решает эту проблему?
• Как Catalyst Optimizer определяет, использовать broadcast join или sort-merge join, и когда ручная настройка необходима?
• Почему repartition(N) создаёт shuffle, а coalesce(N) нет - и когда каждый из них предпочтительнее?

Связанные уроки

• bd-03
• bd-05
• ds-02-cap-theorem
• ml-54-distributed-training
• bd-06