Как установить размер группы строк для файлов в формате hdf?

Question

Я провожу некоторые эксперименты с размером блока (dfs.block.size) и размером группы строк (parquet.block.size) в hdfs.

У меня большой набор данных в формате hdf, и я хочу скопировать данные с различными размерами блоков и групп строк для тестирования. Я могу скопировать данные с другим размером блока, используя:

hdfs dfs -D dfs.block.size=67108864 -D parquet.block.size=67108864 -cp /new_sample_parquet /new_sample_parquet_64M

Но изменяется только dfs.block.size. Я проверяю с hdfs dfs -stat для размера блока, и parquet-tools meta для размера группы строк. На самом деле, если я заменю parquet.block.size с blah.blah.blah это имеет тот же эффект. Я даже вошел в спарк-снаряд и установил parquet.block.size свойство вручную с помощью

sc.hadoopConfiguration.setInt("parquet.block.size", 67108864).

Я использую hadoop 3.1.0. Я получил имя собственности parquet.block.size отсюда

Вот первые 10 строк вывода моей попытки

row group 1:                    RC:4140100 TS:150147503 OFFSET:4
row group 2:                    RC:3520100 TS:158294646 OFFSET:59176084
row group 3:                    RC:880100 TS:80122359 OFFSET:119985867
row group 4:                    RC:583579 TS:197303521 OFFSET:149394540
row group 5:                    RC:585594 TS:194850776 OFFSET:213638039
row group 6:                    RC:2620100 TS:130170698 OFFSET:277223867
row group 7:                    RC:2750100 TS:136761819 OFFSET:332088066
row group 8:                    RC:1790100 TS:86766854 OFFSET:389772650
row group 9:                    RC:2620100 TS:125876377 OFFSET:428147454
row group 10:                   RC:1700100 TS:83791047 OFFSET:483600973

Как вы можете видеть, TS (общий размер) намного больше, чем 64 МБ (67108864 байт)

Моя текущая теория:

Я делаю это в spark-shell:

sc.hadoopConfiguration.setInt("parquet.block.size", 67108864)
val a = spark.read.parquet("my_sample_data")
a.rdd.getNumPartitions // 1034
val s = a.coalesce(27)
s.write.format("parquet").mode("Overwrite").options(Map("dfs.block.size" -> "67108864")).save("/my_new_sample_data")

Так что, возможно, это потому, что мои входные данные уже имеют 1034 раздела. Я действительно не уверен. Мои данные имеют около 118 столбцов в строке.

Answer 1

parquet.block.size собственность влияет только на паркет писателей. hdfs dfs -cp Команда копирует файлы независимо от их содержимого, с другой стороны. parquet.block.size следовательно, свойство игнорируется hdfs dfs -cp,

Представьте, что у вас есть приложение, которое делает снимки экрана в формате JPG или PNG, в зависимости от файла конфигурации. Вы делаете копию этих скриншотов с cp команда. Естественно, даже если вы измените нужный формат изображения в файле конфигурации, cp Команда всегда будет создавать выходные файлы в формате изображения исходных файлов, независимо от файла конфигурации. Конфигурационный файл используется только приложением скриншота, а не cp, Вот как parquet.block.size собственность работает также.

Что вы можете сделать, чтобы изменить размер блока, это переписать файл. Вы упомянули, что у вас есть spark-shell, Используйте это, чтобы переписать файл Parquet, выполнив

sc.hadoopConfiguration.setInt("parquet.block.size", 67108864)
var df = spark.read.parquet("/path/to/input.parquet")
df.write.parquet("/path/to/output")

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

$ spark-shell
scala> sc.hadoopConfiguration.setInt("parquet.block.size", 200000)
scala> var df = spark.read.parquet("/tmp/infile.parquet")
df: org.apache.spark.sql.DataFrame = [field0000: binary, field0001: binary ... 78 more fields]
scala> df.write.parquet("/tmp/200K")
scala> df.write.format("parquet").mode("Overwrite").options(Map("parquet.block.size" -> "300000")).save("/tmp/300K")
scala> :quit
$ hadoop fs -copyToLocal /tmp/{200K,300K} /tmp
$ parquet-tools meta /tmp/infile.parquet | grep "row group" | head -n 3
row group 1:  RC:4291 TS:5004800 OFFSET:4
row group 2:  RC:3854 TS:4499360 OFFSET:5004804
row group 3:  RC:4293 TS:5004640 OFFSET:10000000
$ parquet-tools meta /tmp/200K/part-00000-* | grep "row group" | head -n 3
row group 1:   RC:169 TS:202080 OFFSET:4
row group 2:   RC:168 TS:201760 OFFSET:190164
row group 3:   RC:169 TS:203680 OFFSET:380324
$ parquet-tools meta /tmp/300K/part-00000-* | grep "row group" | head -n 3
row group 1:   RC:254 TS:302720 OFFSET:4
row group 2:   RC:255 TS:303280 OFFSET:284004
row group 3:   RC:263 TS:303200 OFFSET:568884

Посмотрев на значения TS, вы увидите, что входной файл имеет размер группы строк 4,5-5 МБ, а выходные файлы имеют размеры групп строк 200 КБ и 300 КБ соответственно. Это показывает, что значение установлено с помощью sc.hadoopConfiguration становится "по умолчанию", в то время как другой метод, который вы упомянули в комментарии ниже, включающий df.options переопределяет это по умолчанию.

Обновление 2: Теперь, когда вы опубликовали свой вывод, я могу видеть, что происходит. В вашем случае происходит сжатие, увеличивая объем данных, которые будут помещаться в группы строк. Размер группы строк применяется к сжатым данным, но TS показывает размер несжатых данных. Однако вы можете определить размер групп строк, вычтя их начальные смещения. Например, сжатый размер вашей первой группы строк равен 59176084 - 4 = 59176080 байт или меньше (поскольку заполнение также может иметь место). Я скопировал ваши результаты в /tmp/rowgroups.dat на моем компьютере и вычислил размеры вашей группы строк, выполнив следующую команду:

$ cat /tmp/rowgroups.dat | sed 's/.*OFFSET://' | numinterval
59176080
60809783
29408673
64243499
63585828
54864199
57684584
38374804
55453519

(The numinterval Команда находится в num-utils package в Ubuntu.) Как видите, все ваши группы строк меньше указанного вами размера группы строк. (Причиной, по которой они не соответствуют указанному размеру, является PARQUET-1337.)