Я пытаюсь создать сводный файл для использования конечными пользователями, чтобы они не обрабатывали несколько источников с файлами гораздо большего размера. Для этого я: A) перебираю все исходные папки, удаляя 12 наиболее часто запрашиваемых полей, раскручивая паркетные файлы в новом месте, где эти результаты совмещены. Б) Я пытаюсь вернуться к файлам, созданным на шаге А, и повторно объединить их, сгруппировав по 12 полям, чтобы свести их к итоговой строке для каждой уникальной комбинации.
Я обнаружил, что шаг A снижает полезную нагрузку 5: 1 (примерно 250 гигабайт становятся 48,5 гигабайтами). Шаг B, однако, вместо дальнейшего уменьшения, увеличьте это на 50% по сравнению с шагом A. Однако мои подсчеты совпадают.
Здесь используется Spark 1.5.2. Мой код, измененный только для замены имен полей на field1 ... field12, чтобы сделать его более читабельным, приведен ниже с результатами, которые я отметил.
Хотя я не обязательно ожидаю еще одного сокращения 5: 1, я не знаю, что делаю неправильно, чтобы увеличить объем хранилища для меньшего количества строк с той же схемой. Кто-нибудь может помочь мне понять, что я сделал не так?
Спасибо!
//for each eventName found in separate source folders, do the following:
//spit out one row with key fields from the original dataset for quicker availability to clients
//results in a 5:1 reduction in size
val sqlStatement = "Select field1, field2, field3, field4, field5, field6, field7, field8, field9, field10, field11, field12, cast(1 as bigint) as rCount from table"
sqlContext.sql(sqlCommand).coalesce(20).write.parquet("<aws folder>" + dt + "/" + eventName + "/")
//results in over 700 files with a total of 16,969,050,506 rows consuming 48.65 gigs of storage space in S3, compressed
//after all events are processed, aggregate the results
val sqlStatement = "Select field1, field2, field3, field4, field5, field6, field7, field8, field9, field10, field11, field12, sum(rCount) as rCount from results group by field1, field2, field3, field4, field5, field6, field7, field8, field9, field10, field11, field12"
//Use a wildcard to search all sub-folders created above
sqlContext.read.parquet("<aws folder>" + dt + "/*/").registerTempTable("results")
sqlContext.sql(sqlStatement).coalesce(20).saveAsParquetFile("<a new aws folder>" + dt + "/")
//This results in 3,295,206,761 rows with an aggregate value of 16,969,050,506 for rCount but consumes 79.32 gigs of storage space in S3, compressed
//The parquet schemas created (both tables match):
|-- field1: string (nullable = true) (10 characters)
|-- field2: string (nullable = true) (15 characters)
|-- field3: string (nullable = true) (50 characters max)
|-- field4: string (nullable = true) (10 characters)
|-- field5: string (nullable = true) (10 characters)
|-- field6: string (nullable = true) (10 characters)
|-- field7: string (nullable = true) (16 characters)
|-- field8: string (nullable = true) (10 characters)
|-- field9 string (nullable = true) (15 characters)
|-- field10: string (nullable = true)(20 characters)
|-- field11: string (nullable = true)(14 characters)
|-- field12: string (nullable = true)(14 characters)
|-- rCount: long (nullable = true)
|-- dt: string (nullable = true)
