Нужно меньше паркетных напильников

Вы можете переразметить перед сохранением:

rdd.toDF.repartition("Some Column").write.mode(SaveMode.Append).partitionBy("Some Column")

Раньше у меня была эта проблема.

На самом деле вы не можете управлять разделом файлов, потому что это зависит от исполнителя.

Чтобы обойти это, используйте метод coalesce для перетасовки, и вы можете создать столько разделов, сколько хотите, но это неэффективный способ, вам также необходимо установить достаточно памяти драйвера для обработки этой операции.

df = df.coalesce(numPartitions).write.partitionBy(""yyyyy").parquet("xxxx")

Я также столкнулся с этой проблемой. Проблема в том, что если вы используете объединение, каждый раздел получает одинаковое количество файлов паркета. Теперь разные разделы имеют разный размер, поэтому в идеале мне нужно разное объединение для каждого раздела.

Это будет очень дорого, если вы откроете много маленьких файлов. Допустим, вы открываете 1k файлов, и каждый размер файла далек от значения вашего parquet.block.size.

Вот мои предложения:

• Создайте задание, которое сначала объединит ваши входные файлы паркета, чтобы иметь меньшее количество файлов, размер которых близок или равен parquet.block.size. Размер блока по умолчанию составляет 128 МБ, хотя его можно настроить, обновив parquet.block.size. Spark хотел бы, чтобы ваш файл паркета был близок или равен значению вашего parquet.block.size. Размер блока — это размер группы строк, буферизуемой в памяти.
• Или обновите свое задание Spark, чтобы просто читать ограниченное количество файлов.
• Или, если у вас есть большая машина и/или ресурсы, просто сделайте правильный тюнинг.

В запросе Hive есть способ объединить маленькие файлы в один большой. Это недоступно в искровом sql. Кроме того, сокращение spark.sql.shuffle.partitions не поможет с Dataframe API. Я попробовал решение ниже, и оно сгенерировало меньшее количество файлов паркета (с 800 файлов паркета до 29).

Предположим, что данные загружаются в фрейм данных df

1. Создайте временную таблицу в улье.

   df.createOrReplaceTempView("tempTable")
   spark.sql("CREATE TABLE test_temp LIKE test")
   spark.sql("INSERT INTO TABLE test_temp SELECT * FROM tempTable")
   

   test_temp будет содержать небольшие паркетные файлы.

2. Заполнить конечную таблицу куста из временной таблицы

   spark.sql("INSERT INTO test SELECT * FROM test_temp")
   

Окончательная таблица будет содержать файлы меньшего размера. Удалить временную таблицу после заполнения финальной таблицы.