Посмотрите компрессию, если gzip, то всегда будет 1 поток чтения и записи

Неправда. Это если сам файл пожат gzip. Если внутри паркета используется gzip, то будет столько потоков сколько блоков (это если совсем просто, если сложно, то выше описали)

Читается он всё равно в один поток, по крайней мере в датабриксе так

Только недавно проверял

Так, вот что попробуем посмотреть:
1. Сохраняем файл с опцией .option("parquet.block.size", 16*1024*1024)
.option("compression", "none"), так будет немного нагляднее. Получаем:
HdfsNamedFileStatus{path=hdfs://localhost:61827/test/pa/snapshot/snp/part-00001-00000.c0.parquet; isDirectory=false; length=238468917; replication=1; blocksize=16777216; …}

spark.conf.set("spark.sql.files.maxPartitionBytes", 16 * 1024 * 1024)
spark.conf.set("spark.sql.files.openCostInBytes", 16 * 1024 * 1024)
val df = spark.table(files.head.getPath.toString)
println(df.count)
210000
println(df.rdd.getNumPartitions)
12
df.groupBy(fn.spark_partition_id() as "sid").count.orderBy("sid").collect().foreach(println)
[0,14786]
[1,14781]
[2,14777]
[3,14779]
[4,14779]
[5,14786]
[6,14777]
[7,14786]
[8,14777]
[9,14785]
[10,14776]
[11,14786]
[12,14775]
[13,14786]
[14,3064]

spark.conf.set("spark.sql.files.maxPartitionBytes", 512 * 1024) //полмегабайта всего))
spark.conf.set("spark.sql.files.openCostInBytes", 4 * 1024 * 1024)
val df2 = spark.read.parquet("/test/pa/snapshot/snp/part-00001-00000.c0.parquet")
210000
455
[16,14786]
[48,14781]
[80,14777]
[111,14779]
[143,14779]
[175,14786]
[207,14777]
[239,14786]
[272,14777]
[304,14785]
[336,14776]
[368,14786]
[400,14775]
[432,14786]
[451,3064]
Что видим: без сжатия размер блока HDFS соответствует реальному размеру блока паркета, файл состоит из пятнадцати блоков, как его ни читай

2. .option("parquet.block.size", 16*1024*1024)
.option("compression", «gzip»)
HdfsNamedFileStatus{path=hdfs://localhost:63362/test/pa/snapshot/snp/part-00001-00000.c0.gz.parquet; isDirectory=false; length=151256095; replication=1; blocksize=16777216; }
spark.conf.set("spark.sql.files.maxPartitionBytes", 16 * 1024 * 1024)
spark.conf.set("spark.sql.files.openCostInBytes", 16 * 1024 * 1024)
val df = spark.table(files.head.getPath.toString)
210000
10
[0,23301]
[1,23308]
[2,23303]
[3,23315]
[4,23303]
[5,23300]
[6,23309]
[7,23303]
[8,23298]
[9,260]

spark.conf.set("spark.sql.files.maxPartitionBytes", 512 * 1024)
spark.conf.set("spark.sql.files.openCostInBytes", 4 * 1024 * 1024)
val df2 = spark.read.parquet("/test/pa/snapshot/snp/part-00001-00000.c0.gz.parquet»)
210000
289
[6,14786]
[17,8515]
[38,14790]
[49,8518]
[70,14779]
[81,8524]
[102,14790]
[113,8525]
[134,14779]
[145,8524]
[166,14777]
[177,8523]
[198,14790]
[209,8519]
[230,14779]
[241,8524]
[262,14773]
[273,8525]
[288,260]

А вот тут уже интереснее, сжатый файл читается в десять партиций, потому что ParquetWriter накапливает записи в блок до достижения им указанного размера, но он становится меньше после сжатия! При этом размер сопоставленного HDFS блока остаётся таким же, как и указанный при записи. А вот если читать по маленьким кусочкам, то все пятнадцать блоков можно найти, но куча партиций по прежнему пустые

Да, как в телеге посты оформлять я не знаю до сих пор, неудобно капец((

можно код обернуть в "```"

чтобы было вот так

Не совсем понятно, чем отличается блок от партиции. Я пересохранил паркет в 64 партициях df.repartition(64).write. Это вроде как помогло, теперь задействованы все ядра.

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

и там теперь 64 файла по 20 мб?

Спасибо, чуть попозже, может быть))

Ну так вот, тоже хочу задать вопрос, точнее перезадать старый:

Хочу делать много бродкастов в одном большом плане. Сами бродкасты очень мелкие, им хватает примерно 1 Гб хипа на экзекуторах. Но на драйвере приходится выставлять 64+Гб, а это многовато. Похоже, что вся проблема в создании большого количества бродкастов в один момент времени, потому что я вижу, как за секунды выполняются сотни джобов и после этого всё в порядке, если памяти много. Ну или приложение зависает, если памяти мало на драйвере. Как можно снизить параллельность подгрузки бродкастов, или может кто ещё что-то знает об этом?

Да

Средства работы с данными очень разнообразны, поддержка форматов хранения у них может быть реализована по-разному, и если спарк неплохо готов читать свои записанные файлы, то пандас пишет и читает по-своему, и в свою очередь спарковские директории-файлы тоже может не очень-то хорошо воспринимать. Другое дело, выполнить df.toPandas(), там уже всё прочитано. Остается только передать из одной виртуальной машины в другую))

Приветствую всех!
Используем spark 3.0.2 on k8s в клиентском режиме. После непредвиденного завершения работы драйвера, CoarseGrainedExecutorBackend завершается (with exit code 1), но сами поды не завершают свою работу и остаются в статусе running. У кого-нибудь получалось решить эту проблему коробочными средствами?

Пообщался с датабриксерами, а итоге решили, что из-за гигантских шаффлов диски на виртуалках переполняются и достигают лимита, и воркер отстреливается

бывает, а у датабриксов есть мониторы железа типа ганглии чето? чтоб посмотреть чо на нодах

Ага

Но там очень сложно связать все ошибки в одну