带聚合函数的Spark SQL的内存消耗

Question

我有一个配置单元表，只有很少的bigint或string列，超过3800万行，总大小略大于1GB，测试环境是一个小型独立集群，有4个工作节点，每个都有8GB内存，Spark 1.4。在Spark sql shell中，我尝试执行一个sql

create table bla as select user_id, brand_id, sum(cnt) from foo group by user_id, brand_id

有好几次，工作总是停留在第一阶段，几乎没有任务悬而未决。

GC报告似乎表明没有足够的内存来存储临时对象，进程正在等待完整的GC完成。

一个挂起节点的GC输出：

4572.821:[GC[PSYoungGen:1707488K-

我的问题是：我的sql真的需要超过32GB的内存来运行1GB的数据吗？还是有我刚刚错过的快速优化设置？如果我在查询中避免求和/平均值，我可以用更少的记录量，甚至是完整的数据成功地执行sql，如

select user_id, brand_id, count(cnt)

或

select user_id, brand_id, concat(xx)

GC之后，每个节点上的所有工作和内存占用永远不会超过1或2GB。

第二个问题：初始阶段总是使用10个分区，如何更改并行性？

谢谢万春

Answer 1

针对您的具体问题：

>

有两种方法可以更改并行度。首先，您可以更改进审量文件。其次，您可以使用re分区（num分区）显式重新分区数据。在RDD文档中查找此方法。

您可以根据现有分区的数量以编程方式重新分区，以达到所需的并行级别或数据文件粒度，例如。，

val df = ctx.jsonFile("file://...")
df.repartition(10 * df.rdd.partitions.size)

通过使用df.rdd.partitions.size，您还可以根据集群大小动态调整分区数量，例如，获取分区数量，例如，执行器CPU数量的4倍等。

如果您遇到内存问题，增加分区数量是找到在集群约束内工作的处理管道的良好起点。

还有其他与内存相关的优化：

>

如果在reduce阶段内存不足，可能需要增加spark。洗牌内存提取

如果您的操作无法从缓存中受益，您可以减少spark.storage.memoryFraction。

有关配置Spark的更多信息。

Answer 2

我只是想更新一下，我真的相信当前的spark版本在一些聚合函数上的实现不是很理想。e、 g.如果在查询中用count（）或max（）替换sum（），那么查询将在30-40秒内成功返回。最终，我让sum（）查询以双倍的工作线程工作，因此使用双倍的内存，作业在3分钟内完成。出于对钨的好奇，我建造了spark1。5在源代码外，启动在spark1中花费3分钟的相同查询。4，完成时间为40到55秒。万春，干杯