Собралось много реакций на предыдущем посте про Spark, делаю еще один!
Repartition в Spark. Зачем это вообще нужно?
В pandas не задумываешься про куски данных: читаете DataFrame и сразу работаешь с ним целиком. В Spark всё иначе: данные делятся на партиции (шарды), которые обрабатываются разными воркерами. Repartition позволяет управлять тем, как и насколько равномерно эти куски разбросаны по кластеру.
❓ Зачем?
⚖️ Баланс нагрузки на кластер. Spark работает быстрее, если данные распределены по всем воркерам более-менее равномерно. Если партиций мало, часть узлов простаивает, остальные тянут всё на себе и теряется весь смысл распределённых вычислений.
🚤 Ускоряет джойны и агрегации. Самая частая боль в Spark - это медленные джойны или группировки. Причина часто в том, что данные по ключу раскиданы неравномерно. Если сделать .repartition("key") перед джойном Spark сможет склеить нужные куски локально, а не гонять данные по всему кластеру.
📝 Экономит память и снижает риск падений приложений. Иногда Spark после фильтрации или select делает ОЧЕНЬ перекошенные партиции: на одной куча данных, на другой почти ничего. Это может привести к OutOfMemory именно на одном воркере, при том что на других куча свободной памяти. Repartition выравнивает данные и размазывает нагрузку.
🗃️ Контроль количества файлов на выходе. Когда записываешь данные в parquet/csv, Spark по дефолту делает столько файлов, сколько партиций в DataFrame.
Если хочешь один файл — обязательно делайте .repartition(1) перед записью, иначе получишь кучу маленьких частей.
📝 Как это выглядит на практике
🔗 Джойны (делаем repartition по ключу объединения таблиц, так проще собрать ключи, разбросанные по кластеру)
df_left = df_left.repartition("user_id")
df_right = df_right.repartition("user_id")
df_joined = df_left.join(df_right, on="user_id", how="inner")✍️ Запись (в примере ниже указано то, что на выходе мы получаем один файл).
df_result.repartition(1).write.parquet("result.parquet")☝️ Изменяем количество партиций вручную.
df = df.repartition(50) # вручную задаём 50 партиций
Обычно количество партиций автоматически подтягивается из конфига приложения, возможно, при настройке видели параметр
spark.sql.shuffle.partitions⚠ Самое важное в этом посте, что нужно искать размен между количеством партиций и размером задач на воркеры.
1️⃣
Слишком много партиций. Куча маленьких задач, и на маленьких данных становится только хуже, по скорости проседает.
2️⃣
Слишком мало партиций. Неэффективно, Spark теряет свою распределённость, одна нода делает всю работу.
Вообще в DA / DS / ML / DE мы всегда работаем с разменом (трейд-оффами) и все упирается в задачи, которые мы решаем)
Пишем дальше про Spark или нет?
🐳 — Пишем, давай еще, очень интересно
🤝 — Давай уже про что-то другое!