Партиционирование: сколько партиций, repartition и coalesce

Партиция это единица работы Spark: одна задача на одном ядре. От числа партиций зависит, сколько ядер кластера будут заняты, сколько памяти нужно каждой задаче и сколько файлов появится при записи. Слишком мало партиций, и сорок ядер ждут, пока три работают. Слишком много, и Spark тратит больше времени на запуск крошечных задач, чем на работу. Разберём, откуда берётся число партиций и как им управлять.

Грузчики и коробки

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

С партициями то же самое. Ориентиры: задач в стадии в два-четыре раза больше, чем ядер, а одна партиция весит порядка сотни мегабайт.

Откуда берутся партиции

При чтении файлов. Spark режет файлы на куски не больше spark.sql.files.maxPartitionBytes, по умолчанию 128 МБ. Большой файл даёт несколько партиций. Маленькие файлы Spark складывает в одну партицию, но за каждый файл добавляет условную стоимость открытия spark.sql.files.openCostInBytes, 4 МБ. Поэтому тысяча файлов по килобайту превращается не в одну партицию, а в десятки. Это и есть проблема мелких файлов: каждый файл надо найти, открыть и прочитать заголовок.

После shuffle. Группировка, соединение и сортировка раскладывают строки по spark.sql.shuffle.partitions партициям, по умолчанию 200. Для маленьких данных это слишком много, поэтому в Spark 3 появился AQE, адаптивное исполнение. Он смотрит на реальный размер данных после shuffle и сливает соседние мелкие партиции. Поэтому в первых уроках после группировки задача была одна.

По вашей команде. Число партиций меняют двумя методами:

• repartition(n) перемешивает строки по n партициям через полный shuffle. Партиций может стать и больше, и меньше, и они выйдут ровными. repartition(n, "zone_id") кладёт строки одной зоны в одну партицию.
• coalesce(n) только уменьшает число партиций, склеивая соседние без shuffle. Это дешевле, но партиции выйдут неровными, если неровными были исходные.

Типичный случай для coalesce: перед записью небольшого результата, чтобы не получить двести файлов по килобайту.