Трансформации и действия: ленивость как основа Spark

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

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

Два сорта операций

Каждая операция Spark — либо трансформация, либо действие. Различать их обязательно: от этого зависит, когда (и сколько раз) реально побежит вычисление.

Правило-памятка: если операция возвращает новый набор данных — это трансформация (ленивая). Если она возвращает обычное значение (число, список, запись на диск) — это действие (запускает счёт). Цепочка трансформаций строит план; первое же действие приводит план в движение.

map, filter, flatMap — базовые трансформации

Три операции, через которые выражается большинство преобразований. Их смысл легко показать на чистом Python — он буквально совпадает с тем, что делает Spark внутри партиции:

data = ["spark spark rdd", "lazy map filter", "map flatmap reduce"]

# map: одна строка -> один результат (число слов в строке)
mapped = list(map(lambda line: len(line.split()), data))
print("map (число слов в строке):", mapped)

# filter: оставить только подходящие элементы
filtered = list(filter(lambda line: "spark" in line, data))
print("filter (строки со 'spark'):", filtered)

# flatMap: одна строка -> МНОЖЕСТВО результатов, всё в один плоский список
flat = []
for line in data:
    flat.extend(line.split())   # каждую строку «разворачиваем» в слова
print("flatMap (все слова одним списком):", flat)

Вывод:

map (число слов в строке): [3, 3, 3]
filter (строки со 'spark'): ['spark spark rdd']
flatMap (все слова одним списком): ['spark', 'spark', 'rdd', 'lazy', 'map', 'filter', 'map', 'flatmap', 'reduce']

Разница map и flatMap — частый источник путаницы. map сохраняет соответствие «один к одному» (на каждый вход — один выход). flatMap позволяет вернуть на один вход ноль, один или много выходов и «расплющивает» их в единый список. Разбиение строк на слова — классический случай для flatMap.

reduce и reduceByKey

reduce — это действие: оно сворачивает все элементы в одно значение бинарной функцией (например, сумма). А reduceByKey — трансформация над парами «ключ-значение»: она сворачивает значения по каждому ключу отдельно. Это рабочая лошадь агрегаций. Покажем подсчёт частоты слов через свод по ключу:

from functools import reduce

words = ["spark", "rdd", "spark", "map", "rdd", "spark"]

# Превращаем в пары (слово, 1) — как map в Spark
pairs = [(w, 1) for w in words]

# reduceByKey: суммируем значения по каждому ключу
counts = {}
for key, val in pairs:
    counts[key] = counts.get(key, 0) + val
print("reduceByKey -> частоты:", counts)

# reduce как ДЕЙСТВИЕ: свернуть все значения в одно число
total = reduce(lambda a, b: a + b, [v for _, v in pairs])
print("reduce -> всего слов:", total)

Вывод:

reduceByKey -> частоты: {'spark': 3, 'rdd': 2, 'map': 1}
reduce -> всего слов: 6

Важная деталь производительности: reduceByKey сначала сворачивает значения локально внутри каждой партиции (это называют map-side combine) и только потом перемешивает по сети уже сжатые частичные суммы. Поэтому reduceByKey намного эффективнее, чем сгруппировать всё через groupByKey и потом суммировать: groupByKey тащит по сети все исходные значения, reduceByKey — только агрегаты. Это первый практический пример «как уменьшить shuffle».

Почему ленивость — это база, а не оптимизация поверх

Ленивость — не приятное дополнение, а фундамент модели Spark. Откладывая выполнение до действия, Spark получает на руки весь план и может:

• Слить операции в конвейер. map за filter за map внутри одной партиции выполняются за один проход по данным, без создания промежуточных коллекций — операции «склеиваются» в один stage.
• Двинуть фильтры к источнику. Зная, что в конце будет filter, Spark постарается отбросить строки как можно раньше (predicate pushdown), чтобы не тащить лишнее через всю цепочку.
• Не читать ненужное. Если в финале нужны две колонки из ста, Spark прочитает из источника только их (column pruning).

Если бы каждый шаг исполнялся немедленно (как в pandas), ни одну из этих оптимизаций сделать было бы нельзя — данные уже пробежали бы через всю цепочку. Ленивость — это то, что превращает наивную последовательность шагов в эффективный физический план.

Подводные камни

• Несколько действий → несколько пересчётов. Каждое действие запускает план с нуля. Два count() подряд по одной цепочке пересчитают её дважды. Если результат нужен многократно — кэшируйте (следующий урок).
• groupByKey вместо reduceByKey. groupByKey перемешивает по сети все значения и легко вызывает OOM на «горячих» ключах. Почти всегда заменяйте на reduceByKey/aggregateByKey.
• Путаница map и flatMap. Если на один вход нужно несколько выходов (или ноль), это flatMap. map для такого даст вложенные списки.
• Сайд-эффекты в трансформациях. Из-за ленивости и возможного пересчёта print или запись в внешнюю систему внутри map выполнятся непредсказуемое число раз. Трансформации должны быть чистыми.

Best practices

• Стройте длинные цепочки трансформаций — Spark соберёт из них один эффективный план.
• Для агрегаций по ключу предпочитайте reduceByKey/aggregateByKey, а не groupByKey.
• Если одну цепочку нужно использовать несколькими действиями — закэшируйте её, чтобы не считать заново.
• Не кладите сайд-эффекты в трансформации: они ленивы и могут выполниться несколько раз.

Итог

• Трансформации ленивы и возвращают новый набор; действия запускают вычисление и возвращают значение/запись.
• map (1→1), flatMap (1→много, с расплющиванием), filter — базовые трансформации.
• reduce — действие; reduceByKey — эффективная трансформация со сводом по ключу и локальным комбайном.
• Ленивость — фундамент: она даёт Spark весь план для конвейеризации, pushdown фильтров и отсечения колонок.
• reduceByKey почти всегда лучше groupByKey по сети.