Диагностика проблем масштабирования Spark алгоритма

Решение

Типовые проблемы масштабирования Spark алгоритмов

1. Дисбаланс данных (Data Skew)

Это самая частая проблема. Когда данные распределены неравномерно по партициям, некоторые worker'ы обрабатывают в 100+ раз больше данных, чем другие.

Пример:

# Плохо — ключи распределены неравномерно
df.groupBy("user_id").count().show()
# Если у одного пользователя 99% всех записей — один partition получит все данные

2. Недостаток памяти (Out of Memory)

• Broadcast join больших таблиц (> 2 ГБ) в памяти каждого executor'а
• Collect() на больших данных
• Accumulator'ы, растущие без контроля
• Shuffle'ы с большим числом ключей

3. Network Shuffle

При join'е двух больших таблиц Spark должен переместить данные по сети. Если объём shuffle'а больше доступной памяти, начинаются disk spills (запись на диск), что замедляет выполнение в 10+ раз.

4. Каскадные операции без кэширования

# Плохо
df1 = load_data()
df2 = df1.filter(...).groupBy(...).agg(...)
df3 = df2.join(other_df)
df4 = df3.select(...)
df4.write.parquet(...)
# df1 пересчитывается 4 раза

# Хорошо
df1 = load_data()
df2 = df1.filter(...).groupBy(...).agg()
df2.cache()  # Кэшируем
df3 = df2.join(other_df)
df4 = df3.select(...)
df4.write.parquet(...)

Диагностика дисбаланса данных (Data Skew)

Способ 1: Анализ через Spark UI

1. Запустить алгоритм: spark-submit script.py
2. Открыть Spark UI (localhost:4040)
3. Перейти на вкладку Stages
4. Посмотреть распределение времени выполнения по partition'ам
5. Если одна partition'а работает намного дольше — это data skew

Способ 2: Программная диагностика

# Посчитаем распределение ключей
key_dist = df.groupBy("partition_key").count()
stats = key_dist.describe(["count"]).show()

# Если stddev большой (> mean) — дисбаланс
key_dist.orderBy(col("count").desc()).show(5)

Способ 3: Проверка физического плана

df_result.explain(extended=True)
# Ищем SortMergeJoin с большим shuffle'ом

Типы join-операций в Spark

1. Broadcast Join (оптимальный для малых таблиц)

# Применяется автоматически, если таблица < spark.sql.autoBroadcastJoinThreshold (2GB по умолчанию)
small_df = spark.read.parquet("path/to/small_table")
large_df = spark.read.parquet("path/to/large_table")

result = large_df.join(
    broadcast(small_df), 
    "join_key"
)

Преимущества: O(1) network, очень быстро Ограничения: малая таблица должна полностью поместиться в памяти

2. SortMergeJoin (стандартный)

# Используется для join'а двух больших таблиц
df1.join(df2, df1.id == df2.id)

Процесс:

1. Обе таблицы сортируются по join ключу
2. Данные распределяются по одинаковым partition'ам
3. Merge произходит локально на каждом partition'е

Время: O(n log n) из-за сортировки, но не требует broadcast

3. HashJoin (для очень больших таблиц)

# Явное использование
df1.hint("shuffle_hash").join(df2, "id")

Лучше: когда одна таблица значительно больше другой

4. SortMergeJoin с salting (решение для skew'а)

from pyspark.sql.functions import col, rand

# Добавляем random salt к горячим ключам
def salt_skewed_keys(df, key_col, num_buckets):
    return df.withColumn(
        "salted_key",
        col(key_col).cast("string").concat(
            (rand() * num_buckets).cast("int").cast("string")
        )
    )

df1_salted = salt_skewed_keys(df1, "id", 10)
df2_salted = salt_skewed_keys(df2, "id", 10)

# После join'а нужно дедупликировать
result = df1_salted.join(df2_salted, "salted_key")

Оптимизация и поиск узких мест

1. Используй Adaptive Query Execution (AQE)

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

AQE автоматически:

• Переоптимизирует план во время выполнения
• Объединяет маленькие partition'ы
• Переключается на broadcast join если таблица оказалась меньше

2. Профилирование с PySpark profiler

from pyspark import profiler

conf = spark.sparkContext.getConf()
conf.set("spark.python.profile", "true")
conf.set("spark.python.profile.dump.path", "/tmp/spark_profile")

3. Анализ Spark Event Log

# Запустить с логированием
spark-submit --conf spark.eventLog.enabled=true script.py

# Открыть в Spark History Server
# Анализировать: Stage Duration, Shuffle Read/Write, GC Time

Стратегия постепенного увеличения объёма данных

Этап 1: Валидация на семплировании (1%)

df_small = df.sample(fraction=0.01, seed=42)
df_result = process_algorithm(df_small)
df_result.show()
print(f"Memory used: {df_small.memory_usage(deep=True).sum()}")

Этап 2: Постепенное увеличение (5%, 10%, 25%)

for sample_pct in [0.01, 0.05, 0.1, 0.25, 0.5, 1.0]:
    print(f"\nTesting with {sample_pct*100}% data")
    df_sample = df.sample(fraction=sample_pct, seed=42)
    
    start_time = time.time()
    result = process_algorithm(df_sample)
    duration = time.time() - start_time
    
    result.cache().count()  # Force execution
    print(f"Execution time: {duration:.2f}s")
    print(f"Result rows: {result.count()}")
    
    # Проверка на линейное масштабирование
    # Если время выполнения растёт экспоненциально — есть problem

Этап 3: Профилирование на 25% данных

На этом объёме уже видны проблемы, но выполнение относительно быстро.

Этап 4: Финальная проверка на полном датасете

result = process_algorithm(df)
result.write.mode("overwrite").parquet("output_path")

Итоговый чеклист

• Всегда включай AQE (spark.sql.adaptive.enabled=true)
• Используй broadcast join для малых таблиц
• Диагностируй data skew через Spark UI
• Применяй salting для горячих ключей
• Кэшируй промежуточные результаты
• Тестируй на 1%, 5%, 25%, затем 100% данных
• Мониторь shuffle read/write в Spark UI
• Настраивай число партиций (обычно = число cores × 2-3)