Как разбить паркетные файлы на многие разделы в Spark?

Итак, у меня есть только 1 файл паркета, который я читаю с помощью Spark (используя материал SQL), и я бы хотел, чтобы он обрабатывался со 100 разделами. Я попытался установить spark.default.parallelism на 100, мы также попытались изменить сжатие паркета на none (из gzip). Независимо от того, что мы делаем, на первом этапе искровой работы есть только один раздел (после того, как происходит тасование, он перераспределяется на 100, а затем очевидно, что вещи намного быстрее).

Теперь, согласно нескольким источникам (например, ниже), паркет должен быть расщепляемым (даже при использовании gzip!), поэтому я очень смущен и хотел бы получить некоторые советы.

https://www.safaribooksonline.com/library/view/hadoop-application-architectures/9781491910313/ch01.html

Я использую искру 1.0.0, и, по-видимому, значение по умолчанию для spark.sql.shuffle.partitions равно 200, поэтому этого не может быть. На самом деле все значения по умолчанию для parallelism намного больше 1, поэтому я не понимаю, что происходит.

Ответ 1

Вы должны написать свои файлы паркета с меньшим размером блока. Значение по умолчанию составляет 128 МБ на блок, но его можно настроить, установив конфигурацию parquet.block.size в parquet.block.size.

Источник ParquetOuputFormat здесь, если вы хотите копаться в деталях.

Размер блока - это минимальный объем данных, который вы можете прочитать из файла паркета, который является логически читаемым (так как паркет является столбцовым, вы не можете просто разбить на строки или что-то тривиальное, как это), поэтому у вас не может быть больше потоков чтения чем входные блоки.

Ответ 2

Возможно, для вашего файла паркета требуется только один блок HDFS. Создайте большой файл паркета с большим количеством блоков HDFS и загрузите его

val k = sc.parquetFile("the-big-table.parquet")
k.partitions.length

Вы увидите такое же количество разделов, как блоки HDFS. Это отлично работает для меня (spark-1.1.0)

Ответ 3

Вы упомянули, что хотите управлять распределением во время записи в паркет. При создании паркета из паркета RDD сохраняются разделы RDD. Итак, если вы создадите RDD и укажите 100 разделов и из фреймворка с паркетным форматом, тогда он будет писать 100 отдельных паркетных файлов в fs. Для чтения вы можете указать параметр spark.sql.shuffle.partitions.

Ответ 4

Чтобы достичь этого, вы должны использовать SparkContext для установки свойства конфигурации Hadoop (sc.hadoopConfiguration) mapreduce.input.fileinputformat.split.maxsize.

Установив это свойство на меньшее значение, чем hdfs.blockSize, вы получите столько разделов, сколько количество разделов.

Например:
Когда hdfs.blockSize= 134217728 (128 МБ),
и считывается один файл, который содержит ровно один полный блок,
и mapreduce.input.fileinputformat.split.maxsize= 67108864 (64 МБ)

Тогда будут два раздела, в которые будут считываться данные расщепления.

Ответ 5

Новый способ сделать это (Spark 2.x) - настройка

spark.sql.files.maxPartitionBytes

Источник: https://issues.apache.org/jira/browse/SPARK-17998 (официальная документация пока не верна, отсутствует .sql)

По моему опыту, настройки Hadoop больше не действуют.