Блог

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

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

Итак, в общем, вы всегда должны убедиться, что вы перераспределяете свои данные перед записью, чтобы убедиться, что расположение данных соответствует вашим настройкам разделения / компоновки улья (т. Е. Каждая комбинация разделов / блоков не разделяется между несколькими задачами), в противном случае вы можете в конечном итоге записать огромное количество файлов.

Ваши примеры

Я думаю, что существует некоторое недопонимание, поэтому давайте рассмотрим их один за другим.

Сначала несколько, чтобы проверить мое понимание:

df.repartition(100) = 100 выходных файлов одинакового размера

Да — данные будут случайным образом равномерно перемешаны в 100 разделов, вызывая 100 задач, каждая из которых будет записывать ровно один файл.

df.repartition('colA') = 10 выходных файлов разного размера, поскольку каждый файл будет содержать все строки для 1 значения colA

Нет — количество разделов для перетасовки не указано, поэтому по умолчанию оно будет равно 200. Таким образом, у вас будет 200 задач, не более 10 из которых будут содержать какие-либо данные (может быть меньше из-за коллизий хэшей), так что в итоге у вас будет 190 пустых файлов и 10 с данными. * Примечание: с AQE в spark 3 spark может решить объединить 200 разделов в меньшее количество, когда поймет, что большинство из них очень маленькие. Я не знаю точной логики, поэтому технически ответ на самом деле «200 или меньше, только 10 будут содержать данные».

df.repartition('colB') = 1000 выходных файлов Нет — Как и выше, данные будут перетасованы в 200 разделов. Однако в этом случае все они (вероятно) будут содержать данные, поэтому вы получите 200 файлов примерно одинакового размера.

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

df.repartition(50, 'colA') = 50 выходных файлов?

Да — Аналогично предыдущему, за исключением того, что теперь мы переопределили количество разделов с 200 до 50. Итак, 10 файлов с данными, 40 пустых. (или меньше из-за AQE)

df.repartition(50, 'colB') = 50 выходных файлов, поэтому некоторые файлы будут содержать более одного значения ColB?

Да — как и раньше, мы получим 50 файлов немного разных размеров в зависимости от того, как работают хэши значений ColB.

Разделы улья:

(Я думаю, что приведенные ниже примеры написаны с учетом df того, что для начала используется 100 разделов, а не 200, как указано, поэтому я собираюсь пойти с этим)

output.write_dataframe(df, partition_cols=['colA']) = 1000 выходных файлов (потому что я получаю потенциально 100 файлов в каждом из 10 разделов улья 10)

Да — у вас будет 100 задач, каждая из которых будет записывать один файл для каждого значения colA, которое они видят. Таким образом, до 1000 файлов в случае случайного распределения данных.

output.write_dataframe(df, partition_cols=['colB']) = 10 000 выходных файлов

Нет — здесь отсутствует 0. 100 задач, каждая из которых может записывать до 1000 файлов (по одному для каждого значения ColB), в общей сложности до 100 000 файлов.

output.write_dataframe(df, partition_cols=['colA', 'colB']) = 100 000 выходных файлов

Нет — 100 задач, каждая из которых будет записывать один файл для каждой комбинации столбцов разделов, которые она видит. Таких комбинаций 10 000, так что это может записать до 100 * 10 000 = 1 000 000 файлов!

output.write_dataframe(df.repartition('colA'), partition_cols=['colA']) = 10 выходных файлов разного размера (по 1 файлу в каждом разделе улья)

Да — repartition наши данные будут перетасованы в 200 задач, но только 10 будут содержать данные. Каждый будет содержать ровно одно значение colA , поэтому будет записан ровно один файл. Остальные 190 задач не будут записывать файлы. Итак, ровно 10 файлов.

Группирование:

Опять же, предполагая 100 разделов для df, а не 200

output.write_dataframe(df, bucket_cols=[‘colB’], bucket_count=100) = 100 выходных файлов? В эксперименте, похоже, это не так

Нет — поскольку мы не тщательно распределили данные, у нас есть 100 задач с (возможно) случайно распределенными данными. Каждая задача будет записывать один файл на каждую корзину, которую она видит. Таким образом, это может записать до 100 * 100 = 10 000 файлов!

output.write_dataframe(df, bucket_cols=[‘colA’], bucket_count=10) = 10 выходных файлов?

Нет — Аналогично описанному выше, 100 задач, каждая из которых может записывать до 10 файлов. Итак, в наихудшем случае здесь 1000 файлов.

output.write_dataframe(df.repartition(‘colA’), bucket_cols=[‘colA’], bucket_count=10) = ???

Теперь мы корректируем расположение данных перед записью, у нас будет 200 задач, не более 10 из которых будут содержать какие-либо данные. Каждое значение colA будет существовать только в одной задаче.

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

Примечание: Из-за коллизий хэшей один или несколько сегментов могут быть пустыми, поэтому мы можем получить не ровно 10.

Теперь все вместе:

Опять же, предполагая 100 разделов для df, а не 200

output.write_dataframe(df, partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200) = ???

100 задач. 10 разделов улья. 200 сегментов. В худшем случае каждая задача записывает по одному файлу на комбинацию раздела улья / корзины. т.е. 100 * 10 * 200 = 200 000 файлов.

output.write_dataframe(df.repartition(‘colA’, ‘colB’), partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200) = ??? — Это та команда, которую я хочу использовать в конце? И все, что находится ниже по потоку, сначала фильтруется на colA, чтобы воспользоваться разделением улья, а затем присоединяется к ColB, чтобы воспользоваться объединением?

Этот хитрый. У нас есть 200 задач, и данные тщательно перемешиваются, поэтому каждая комбинация colA / ColB находится только в одной задаче. Итак, все кажется хорошим.

НО каждый сегмент содержит несколько значений ColB , и мы ничего не сделали, чтобы убедиться, что весь сегмент локализован для одной задачи spark.

Таким образом, в худшем случае мы могли бы получить один файл на значение ColB на раздел улья (значение colA). т.е. 10 * 1000 = 10000 файлов.

Учитывая наши конкретные параметры, мы можем добиться немного большего, просто сосредоточившись на оптимальном расположении сегментов:

 output.write_dataframe(df.repartition(200, ‘colB’), partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200)
  

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

Тогда мы получим один файл для каждого значения colA в задаче (вероятно, 10, поскольку colA перемешивается случайным образом), так что не более 200 * 10 = 2000 файлов.

Это лучшее, что мы можем сделать, предполагая, что colA и ColB не коррелируют.

Заключение

Не существует универсального подхода к управлению размерами файлов.

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

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

Самое главное — понять, как взаимодействуют эти 3 концепции (как описано в разделе «Фон» выше), чтобы вы могли рассуждать о том, что произойдет с первыми участниками.