Разделение паркета Spark: большое количество файлов

Ответ 1

Во-первых, я бы действительно избегал использования coalesce, поскольку это часто продвигается в цепочке преобразований и может разрушить параллелизм вашей работы (я спрашивал об этой проблеме здесь: Как предотвратить оптимизацию Spark)

Записать 1 файл на паркетный раздел очень просто (см. метод записи Spark для данных с фрейма, записывающий множество маленьких файлов):

data.repartition($"key").write.partitionBy("key").parquet("/location")

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

data.repartition($"key",$"another_key").write.partitionBy("key").parquet("/location")

another_key может быть другим атрибутом вашего набора данных или производным атрибутом, использующим некоторые операции по модулю или округлению существующих атрибутов. Вы даже можете использовать оконные функции с row_number поверх key, а затем округлить это чем-то вроде

data.repartition($"key",floor($"row_number"/N)*N).write.partitionBy("key").parquet("/location")

Это поместит ваши записи N в 1 файл паркета

используя orderBy

Вы также можете контролировать количество файлов без перераспределения, упорядочив свой фрейм данных соответственно:

data.orderBy($"key").write.partitionBy("key").parquet("/location")

Это приведет к spark.sql.shuffle.partitions во всех разделах (по умолчанию 200). Даже полезно добавить второй столбец порядка после $key, так как паркет запомнит порядок данных и запишет статистику соответственно. Например, вы можете заказать по идентификатору:

data.orderBy($"key",$"id").write.partitionBy("key").parquet("/location")

Это не изменит количество файлов, но улучшит производительность при запросе файла паркета для заданных key и id. Смотрите, например, https://www.slideshare.net/RyanBlue3/parquet-performance-tuning-the-missing-guide и https://db-blog.web.cern.ch/blog/luca-canali/2017-06-diving-spark-and-parquet-workloads-example

Искра 2. 2+

Начиная с Spark 2.2, вы также можете играть с новой опцией maxRecordsPerFile, чтобы ограничить количество записей в файле. Вы все равно получите как минимум N файлов, если у вас N разделов, но вы можете разбить файл, записанный на 1 раздел (задачу), на более мелкие куски:

df.write
.option("maxRecordsPerFile", 10000)
...

Смотрите, например, http://www.gatorsmile.io/anticipated-feature-in-spark-2-2-max-records-written-per-file/ и спарк-запись на диск с N файлами менее N разделов

Ответ 2

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

import org.apache.spark.sql.functions.rand

df.repartition(numPartitions, $"some_col", rand)
  .write.partitionBy("some_col")
  .parquet("partitioned_lake")

В этом посте подробно описываются все параметры разделения, которые должны использоваться в сочетании с partitionBy.

Ответ 3

Это работает для меня очень хорошо:

data.repartition(n, "key").write.partitionBy("key").parquet("/location")

Он создает N файлов в каждом выходном разделе (каталоге) и (анекдотически) быстрее, чем использование coalesce и (опять же, анекдотически, в моем наборе данных) быстрее, чем только перераспределение на выходе.

Если вы работаете с S3, я также рекомендую делать все на локальных дисках (Spark много делает для создания/переименования/удаления файлов во время записи), и как только все закончится, используйте hadoop FileUtil (или просто aws cli), чтобы скопировать все над:

import java.net.URI
import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
// ...
  def copy(
          in : String,
          out : String,
          sparkSession: SparkSession
          ) = {
    FileUtil.copy(
      FileSystem.get(new URI(in), sparkSession.sparkContext.hadoopConfiguration),
      new Path(in),
      FileSystem.get(new URI(out), sparkSession.sparkContext.hadoopConfiguration),
      new Path(out),
      false,
      sparkSession.sparkContext.hadoopConfiguration
    )
  }

Редактировать: Согласно обсуждению в комментариях:

Вы - набор данных со столбцом раздела YEAR, но в каждом данном ГОДЕ содержится много разных данных. Таким образом, один год может иметь 1 ГБ данных, а другой может иметь 100 ГБ.

Вот psuedocode для одного способа справиться с этим:

val partitionSize = 10000 // Number of rows you want per output file.
val yearValues = df.select("YEAR").distinct
distinctGroupByValues.each((yearVal) -> {
  val subDf = df.filter(s"YEAR = $yearVal")
  val numPartitionsToUse = subDf.count / partitionSize
  subDf.repartition(numPartitionsToUse).write(outputPath + "/year=$yearVal")
})

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

Другой способ сделать это - написать свой собственный разделитель, но я понятия не имею, что с этим связано, поэтому я не могу предоставить какой-либо код.