pyspark: эффективно иметь partitionBy записывать то же количество полных разделов, что и исходная таблица

У меня возник вопрос, связанный с функцией pyspark repartitionBy repartitionBy() которую я изначально разместил в комментарии по этому вопросу. Меня попросили опубликовать его как отдельный вопрос, так вот вот:

Я понимаю, что df.partitionBy(COL) будет записывать все строки с каждым значением COL в свою собственную папку и что каждая папка (предполагая, что строки были ранее распределены по всем разделам каким-либо другим ключом), имеет примерно одинаковое число файлов, которые ранее были во всей таблице. Я нахожу это поведение раздражающим. Если у меня есть большая таблица с 500 разделами, и я использую partitionBy(COL) в некоторых столбцах атрибутов, теперь у меня есть, например, 100 папок, каждая из которых содержит 500 (сейчас очень маленьких) файлов.

Я бы хотел, чтобы поведение partitionBy(COL), но с примерно таким же размером файла и количеством файлов, что и у меня изначально.

В качестве демонстрации предыдущий вопрос разделяет игрушечный пример, где у вас есть таблица с 10 разделами и сделать partitionBy(dayOfWeek) и теперь у вас есть 70 файлов, потому что в каждой папке 10. Я бы хотел ~ 10 файлов, по одному на каждый день и, возможно, 2 или 3 дня, у которых больше данных.

Может ли это быть легко достигнуто? Что-то вроде df.write().repartition(COL).partitionBy(COL) похоже, что он может работать, но я беспокоюсь, что (в случае очень большой таблицы, которая должна быть разбита на несколько папок), чтобы сначала объединить это к небольшому количеству разделов, прежде чем делать partitionBy(COL) кажется плохой идеей.

Любые предложения приветствуются!

Ответ 1

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

(1) df.repartition(numPartitions, * cols).write.partitionBy(* cols).parquet(writePath)

Сначала будет использоваться разделение на основе хеша, чтобы гарантировать, что ограниченное число значений из COL попадет в каждый раздел. В зависимости от значения, numPartitions вами для numPartitions, некоторые разделы могут быть пустыми, в то время как другие могут быть переполнены значениями - для тех, кто не знает почему, прочитайте это. Затем, когда вы вызываете partitionBy в DataFrameWriter, каждое уникальное значение в каждом разделе будет помещено в отдельный файл.

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

(2) df.sort(sortCols).write.parquet(writePath)

Эта опция прекрасно работает, когда вы хотите (1) файлы, которые вы пишете, иметь почти равные размеры (2) точный контроль над количеством записанных файлов. Этот подход сначала глобально сортирует ваши данные, а затем находит разбиения, которые разбивают данные на k разделов равномерного размера, где k указано в конфигурации config spark.sql.shuffle.partitions. Это означает, что все значения с одинаковыми значениями вашего ключа сортировки смежны друг с другом, но иногда они разделяют разделение и находятся в разных файлах. Это, если ваш вариант использования требует, чтобы все строки с одинаковым ключом были в одном разделе, то не используйте этот подход.

Есть два дополнительных бонуса: (1) путем сортировки данных их размер на диске часто может быть уменьшен (например, сортировка всех событий по user_id, а затем по времени приведет к большому количеству повторений в значениях столбцов, что способствует сжатию) и (2 ) если вы записываете в формат файла, который поддерживает его (например, Parquet), то последующие читатели могут оптимально считывать данные с помощью предиката push-down, потому что средство записи паркета запишет значения MAX и MIN каждого столбца в метаданных, позволяя считыватель для пропуска строк, если в запросе указаны значения за пределами диапазона (min, max).

Обратите внимание, что сортировка в Spark обходится дороже, чем просто перераспределение и требует дополнительного этапа. За кулисами Spark сначала определяет разбиения на одном этапе, а затем перетасовывает данные в эти разбиения на другом этапе.

(3) df.rdd.partitionBy(customPartitioner).toDF(). Write.parquet(writePath)

Если вы используете spark в Scala, то вы можете написать клиентский разделитель, который сможет преодолеть надоедливые ошибки разделителя на основе хеша. К сожалению, не вариант в PySpark. Если вы действительно хотите написать собственный разделитель в pySpark, я обнаружил, что это возможно, хотя и немного неловко, используя rdd.repartitionAndSortWithinPartitions:

df.rdd \
  .keyBy(sort_key_function) \  # Convert to key-value pairs
  .repartitionAndSortWithinPartitions(numPartitions=N_WRITE_PARTITIONS, 
                                      partitionFunc=part_func) \
  .values() # get rid of keys \
.toDF().write.parquet(writePath)

Может быть, кто-то еще знает более простой способ использовать пользовательский разделитель на фрейме данных в pyspark?

Ответ 2

df.write().repartition(COL).partitionBy(COL) запишет один файл на раздел. Это не будет работать хорошо, если один из ваших разделов содержит много данных. например если один раздел содержит 100 ГБ данных, Spark попытается записать файл размером 100 ГБ, и ваша работа, вероятно, будет взорвана.

df.write().repartition(2, COL).partitionBy(COL) запишет не более двух файлов на раздел, , как описано в этом ответе. Этот подход хорошо работает для наборов данных, которые не очень искажены (поскольку оптимальное количество файлов на раздел примерно одинаково для всех разделов).

В этом ответе объясняется, как записать больше файлов для разделов с большим количеством данных и меньше файлов для небольших разделов.