Я пытаюсь создать сводный файл для конечных пользователей, чтобы они не обрабатывали несколько источников с файлами большего размера. Для этого я: 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)