Что такое lazy evaluation в Spark и почему это важно?

Lazy Evaluation в Apache Spark

Lazy evaluation — это ключевая концепция в Spark, при которой преобразования данных не выполняются сразу, а откладываются до момента вызова действия (action). Это критически важно для оптимизации и производительности.

Как это работает?

Без lazy evaluation (Pandas, обычный Python):

df = pd.read_csv('1gb_file.csv')  # Загружает 1GB
df = df[df['age'] > 18]  # Фильтрует сразу (выполняется)
df = df.groupby('city').sum()  # Группирует сразу (выполняется)
result = df.head(10)  # Показывает 10 строк
# Проблема: загрузили и обработали ВСЕ данные, хотя нужны только 10 строк!

С lazy evaluation (Spark):

df = spark.read.csv('1gb_file.csv')  # Не загружает сразу (отложено)
df = df.filter(df.age > 18)  # Добавляет в план (не выполняет)
df = df.groupby('city').sum()  # Добавляет в план (не выполняет)
result = df.head(10)  # ТЕПЕРЬ выполняет только нужную часть
# Spark оптимизирует: считает только 10 строк после фильтрации

Transformations vs Actions

Transformations (Lazy - откладываются):

# ВСЕ эти операции НЕ выполняются сразу!
df = df.select('user_id', 'email')  # Lazy
df = df.filter(df.age > 18)  # Lazy
df = df.groupby('city').count()  # Lazy
df = df.orderby('count')  # Lazy
df = df.distinct()  # Lazy
df = df.join(other_df, on='user_id')  # Lazy

Spark строит граф вычислений (DAG - Directed Acyclic Graph).

Actions (Eager - выполняются сразу):

# ТОЛЬКО ЭТИ операции ДЕЙСТВИТЕЛЬНО выполняют вычисления
result = df.collect()  # Action: возвращает все данные
result = df.count()  # Action: подсчитывает строки
result = df.first()  # Action: возвращает первую строку
result = df.head(10)  # Action: возвращает 10 строк
df.show()  # Action: выводит в консоль
df.write.parquet('output')  # Action: пишет в файл
df.write.csv('output')  # Action: пишет CSV

Пример оптимизации через Lazy Evaluation

Неправильно (много actions):

df = spark.read.csv('data.csv')
df = df.filter(df.age > 18)

# Action 1 — полностью читает и фильтрует
count = df.count()  # Action
print(f"Users: {count}")

# Action 2 — СНОВА читает и фильтрует!
result = df.head(10)  # Action
print(result)

# Action 3 — СНОВА читает и фильтрует!
df.write.parquet('output')  # Action

# Результат: один и тот же фильтр выполнили 3 раза!

Правильно (кэширование + один action):

df = spark.read.csv('data.csv')
df = df.filter(df.age > 18)
df.cache()  # Кэшируем в памяти

# Все операции используют кэш
count = df.count()  # Action 1 (кэширует результат)
result = df.head(10)  # Action 2 (из кэша)
df.write.parquet('output')  # Action 3 (из кэша)

# Результат: фильтр выполнили один раз, остальное из кэша

DAG (Directed Acyclic Graph)

Spark визуализирует план вычислений:

df = spark.read.parquet('orders.parquet')
df = df.filter(df.amount > 100)  # Lazy
df = df.groupby('customer_id').agg({'amount': 'sum'})  # Lazy
df = df.orderby('amount')  # Lazy

# DAG выглядит примерно так:
# ReadParquetFile → Filter(amount > 100) → GroupBy → OrderBy

# При вызове action все операции объединяются
result = df.collect()  # Action

Spark может видеть ВСЕ операции и оптимизировать глобально!

Практические примеры

1. Predicate Pushdown (оптимизация)

# Без lazy evaluation (неправильно):
df = pd.read_parquet('1tb_file.parquet')  # Загружает 1TB в память
df = df[df['date'] >= '2024-01-01']  # ПОТОМ фильтрует

# С lazy evaluation (правильно):
df = spark.read.parquet('1tb_file.parquet')  # Не загружает
df = df.filter(df.date >= '2024-01-01')  # Добавляет в план

# При выполнении Spark АВТОМАТИЧЕСКИ:
# 1. Читает Parquet файл
# 2. Применяет фильтр ДО загрузки в память (predicate pushdown)
# 3. Загружает только нужные строки
# Результат: вместо 1TB в памяти только нужное количество

2. Кэширование для повторяющихся операций

# Читаем большой датасет
df = spark.read.parquet('large_data.parquet')

# Делаем сложную трансформацию (lazy)
df = df.filter(df.status == 'active')
df = df.groupby('region').agg({'revenue': 'sum'})

# Кэшируем, потому что будем использовать несколько раз
df.cache()

# Несколько operations используют кэш
total = df.count()  # Action 1
top10 = df.orderby('revenue').head(10)  # Action 2 (из кэша)
df.write.parquet('output')  # Action 3 (из кэша)

# Без кэша: каждая операция пересчитывала трансформацию
# С кэшем: трансформация один раз, остальное из памяти

3. Проблема: Ненужные трансформации

# Плохо
df = spark.read.parquet('data.parquet')
df = df.select('user_id', 'email')  # Выбираем 2 колонки из 100
df = df.groupby('user_id').count()  # Группируем

# При выполнении Spark читает ВСЕ 100 колонок, потом селектит
# Результат: неправильное использование памяти

# Хорошо
df = spark.read.parquet('data.parquet')
df = df.select('user_id', 'email')  # Lazy

# При выполнении Spark智聪 оптимизирует:
# "Мне нужны только user_id и email из Parquet"
# Читает только эти 2 колонки
# Результат: экономия памяти и скорость

Явное управление вычислениями

df = spark.read.csv('data.csv')
df = df.filter(df.age > 18)

# Явно запустить план и посмотреть
df.explain()  # Показывает физический план

# Вывод:
# == Physical Plan ==
# *(1) Filter (age#0 > 18)
# +- *(1) FileScan csv [age#0, ...] Batched: false, ...

# Это показывает, как Spark ДЕЙСТВИТЕЛЬНО выполнит операции

Когда lazy evaluation может быть проблемой

# Неправильно: предполагаешь, что данные уже загружены
df = spark.read.csv('data.csv')
df = df.filter(df.age > 18)

# Вызовешь функцию, которая работает с df
process_data(df)  # Передаёшь НЕ загруженные данные!

# Функция может упасть, потому что не знает про lazy evaluation
def process_data(df):
    # Если здесь нет action, то ничего не выполнится!
    print(df.columns)  # OK, это не action
    # df.count()  # Нужен action, чтобы данные загрузились

Best Practices

# 1. Фильтруй как можно раньше
df = spark.read.parquet('data.parquet')
df = df.filter(df.date >= '2024-01-01')  # Сразу фильтруем
df = df.select('user_id', 'amount')  # Выбираем нужные колонки

# 2. Кэшируй перед несколькими actions
df.cache()
count = df.count()
result = df.show()

# 3. Удаляй кэш когда не нужен
df.unpersist()

# 4. Используй explain() для отладки
df.explain()

# 5. Избегай collect() на больших датасетах (вернёт всё в памяти)
# Вместо этого: write, head(), take(n)

Вывод

Lazy evaluation — это суперсила Spark:

1. Оптимизация глобально — видит весь DAG
2. Predicate pushdown — фильтрует в источнике
3. Экономия памяти — не загружает ненужное
4. Экономия вычислений — пропускает ненужные операции
5. Кэширование — повторное использование результатов

Без понимания lazy evaluation можно написать очень медленный Spark код. С пониманием — код будет летать!