apache-spark - 为什么聚合的 Spark Parquet 文件比原始文件大？

Question

我正在尝试为最终用户创建一个聚合文件，以避免让他们处理具有更大文件的多个源。为此，我:
A) 遍历所有源文件夹，去除最常请求的 12 个字段，在这些结果位于同一位置的新位置分离出 Parquet 文件。
B) 我尝试返回在步骤 A 中创建的文件，并通过按 12 个字段分组来重新聚合它们，以将其减少为每个唯一组合的汇总行。

我发现的是，步骤 A 以 5:1 的比例减少了有效负载(大约 250 演出变成了 48.5 演出)。然而，步骤 B 不是进一步减少，而是比步骤 A 增加 50%。但是，我的计数匹配。

这是使用 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)

Answer 1

一般来说，像 Parquet 这样的列式存储格式在数据分布(数据组织)和单个列的基数方面非常敏感。数据越有组织，基数越低，存储效率就越高。

作为您应用的聚合，必须对数据进行洗牌。当您检查执行计划时，您会看到它正在使用哈希分区器。这意味着聚合后的分布可能比原始数据的分布效率低。同时sum可以减少行数但增加 rCount 的基数柱子。

您可以尝试不同的工具来纠正这个问题，但并非所有工具都在 Spark 1.5.2 中可用:

按基数较低的列(由于完全洗牌而相当昂贵)或 sortWithinPartitions 对完整数据集进行排序.

使用 partitionBy DataFrameWriter的方法使用低基数列对数据进行分区。

使用 bucketBy和 sortBy DataFrameWriter 的方法(Spark 2.0.0+) 使用分桶和本地排序改进数据分布。