Попробуй покрути макс размер входной партиции.

пробуй указать spark.sql.files.maxPartitionBytes=67108864
Затем последовательно уменьшай до разумного предела 4-8мб
Если не начнёт делится, поздравляю, файл читается в один блок

~600 mb, сжатия как я понимаю вообще нет, но этот момент проверю

указывается через --conf , если спарк-субмит, или интерактивно через spark.conf.set(_,_)

не знаток local режима, но если экзекьютор один (как видно на картинке), то каков был ожидаемый результат? что спарк размажет выполнение по нескольким таскам в рамках этого одного экзекьютора и тем самым добавит себе работы? вообще - теоретически интересно - стоит ожидать от локального режима поведения, характерного для нормального кластерного режима (если кто разбирался...)?

конкретно этот алгоритм один на все режимы, но учитывает spark.sql.shuffle.partitions, которых в локальном режиме должно быть поменьше

попробуйте еще spark.sql.files.maxPartitionBytes уменьшить, но у вас и так там 600мб

по идее да, должен был прочитать в пять партиций. Но вот, что-то там идёт не так и я подозреваю, что файл записан в один parquet block.

Еще тогда нужен fileStatus.toString - посмотрим, какой размер блока

Сжатие было, пересохранил без него, не помогло. maxPartitionBytes тоже не помогло.

А можно поподробнее про запись в один parquet block? Это некая неделимая сущность, которую инструментами спарка никак не разбить на отдельные партиции в принципе? И работает ли то же самое для орка?

структуру паркет файла можно же посмотреть, если там правда один "блок" (не помню, как это в паркете называется), то вряд ли что-то именно с этим файлом можно сделать (в плане параллелизма)...

Гугл не помог, как можно посмотреть кол-во блоков. В данном случае паркет записан из под пандаса.

так с лету вот это нашлось - вроде бы (как и помнил) parquet-tools могут показать структуру, вот тут (https://www.youtube.com/watch?v=rVC9F1y38oU) я бегло глянул - чел рассказывает и показывает в том числе возможности parquet-tools. Я бы начал с этого (но можно заморочиться и почитать формат самостоятельно, уверен - не очень сложно, я делал это упражнение для ORC)

как спарк читает паркетник? Имеют значение две настройки: spark.sql.files.maxPartitionBytes и spark.sql.files.openCostInBytes. Нехитрым алгоритмом
Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))
спарк находит, сколько же байт он хочет прочитать в одну партицию, и все файлы, начиная с самых крупных, пытается таким образом разделить. Например, если есть файлы 120 Мб и 260 Мб, второй будет прочитан в партиции 0-127мб, 128-255Мб, остаток от него пойдёт в третью партицию вместе с 0-120Мб+256-260Мб
А вот после этого начинается самое интересное (и тупое, на мой взгляд). Спарк начинает чтение файла с выбранного оффсета в каждой партиции, и пытается выбранным способом получить данные (схема при этом уже определена заранее). Если получается найти хоть одну строку - данные будут в этой партиции RDD, иначе она останется пустая(( Если же кодек читает данные из следующего оффсета - не беда, значит следующая партиция столько записей недополучит. Это приводит к тому, что практически невозможно определить, сколько реально записей будет в каждой партиции, а число самих партиций может быть сильно завышено, и из сотни партиций после чтения данные могут быть только в десяти (а то и в одной-двух)

И вот, тут получается, что данные читает только 31 партиция, то есть, похоже, что файл предоставляет данные только начиная с какого-то смещения и не разбит на блоки

Звучит как-то офигительно сложно, я пока мало что понял, если честно.
Посижу, перечитаю и обдумаю.)

почти наверняка ничего не сделать, файл записан в один блок, прочитать его в две части не получится

Вот распечатанный fileStatus:
HdfsNamedFileStatus{path=hdfs://localhost:59355/test/pa/snapshot/snp/part-00001-00021.c0.snappy.parquet; isDirectory=false; length=121727673; replication=1; blocksize=16777216; modification_time=1623836486086; access_time=1623836481297; …}

Так вот, в нем можно прочитать восемь блоков! Но не больше, хоть по одному байту ставь maxPartitionBytes
Теперь внимание, тот же файл, но записанный с другим размером блока:

HdfsNamedFileStatus{path=hdfs://localhost:59679/test/pa/snapshot/snp/part-00001-00021.c0.snappy.parquet; isDirectory=false; length=71321718; replication=1; blocksize=268435456; modification_time=1623836902412; access_time=1623836900014;…}

Размер файла стал меньше, в основном потому, что в нем меньше блоков и  snappy сжимает его эффективнее

после обеда могу показать то же самое, без компрессии и с другими кодеками сжатия, если интересно кому

интересно

а сколько executor выделено на сессию?