Spark: как работает partitionBy (DataFrameWriter)?

Question

Я использую partitionBy, но я не совсем понимаю, почему мы должны это использовать.

У меня есть запись в формате csv, подобная этой:

--------------------------- ---------
name | age | entranceDate | dropDate |
--------------------------------------
Tom  | 12  | 2019-10-01   | null     |
--------------------------------------
Mary | 15  | 2019-10-01   | null     |
--------------------------------------

Что произойдетесли я использую:

String[] partitions =
new String[] {
  "name",
  "entranceDate"
};

df.write()
.partitionBy(partitions)
.mode(SaveMode.Append)
.parquet(parquetPath);

А что если я разделю на столбец null:

String[] partitions =
new String[] {
  "name",
  "dropDate"
};

df.write()
.partitionBy(partitions)
.mode(SaveMode.Append)
.parquet(parquetPath);

Может кто-нибудь объяснить, как это работает?Спасибо.

Answer 1

Поведение df.write.partitionBy работает следующим образом:

• Для каждого раздела фрейма данных получить уникальные значения столбцов в аргументе partitionBy
• Запишите данные для каждой уникальной комбинации в отдельный файл

В приведенном выше примере предположим, что ваш фрейм данных содержит 10 разделов.Предположим, что разделы 1-5 имеют 5 уникальных комбинаций имени и даты въезда, разделы 6-10 имеют 10 уникальных комбинаций имени и даты въезда.Каждая комбинация имени и даты входа будет записана в отдельный файл.Таким образом, разделы 1-5 каждый будет записан в 5 файлов и разделы 6-10, каждый будет разделен на 10 файлов.Общее количество файлов, сгенерированных операцией записи, будет 5 * 5 + 5 * 10 = 75. partitionBy просматривает уникальные значения комбинации столбцов.Из документации API:

Разделение вывода по заданным столбцам в файловой системе.Если указано, выходные данные размещаются в файловой системе аналогично схеме разбиения Hive.Например, когда мы разбиваем набор данных по годам, а затем по месяцам, макет каталога будет выглядеть так: - год = 2016 / месяц = ​​01 / - год = 2016 / месяц = ​​02 /

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

Это применимо для всех файловых источников данных (например, Parquet, JSON), начиная с Spark.2.1.0.

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

Answer 2

DataFrameWriter partitionBy принимает независимо текущие разделы DataFrame и записывает каждый раздел, разделенный на уникальные значения переданных столбцов.

Давайте возьмем ваш пример и предположим, что мыуже есть два раздела DF, и мы хотим partitionBy() только с одним столбцом - name.

Раздел 1

--------------------------- ---------
name | age | entranceDate | dropDate |
--------------------------------------
Tom  | 12  | 2019-10-01   | null     |
--------------------------------------
Mary | 15  | 2019-10-01   | null     |
--------------------------------------
Tom  | 15  | 2019-10-01   | null     |
--------------------------------------

Раздел 2

--------------------------- ---------
name | age | entranceDate | dropDate |
--------------------------------------
Tom  | 12  | 2019-10-01   | null     |
--------------------------------------
Tom | 15  | 2019-10-01   | null     |
--------------------------------------
Tom  | 15  | 2019-10-01   | null     |
--------------------------------------

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

В случае нескольких столбцов partitionBy () принимает комбинацию значений.