如何在Spark上执行大型计算

Question

我在Hive中有2个表：user和item，我试图计算每个表的2个特征之​​间的余弦相似度，用于2个表之间的笛卡尔积，即交叉连接。

有大约20000 users和5000 items导致100 million行计算。我使用带有12个内核的Hive Cluster上的Scala Spark运行计算。

代码有点像这样：

val pairs = userDf.crossJoin(itemDf).repartition(100)
val results = pairs.mapPartitions(computeScore)  // computeScore is a function to compute the similarity scores I need

由于Hadoop集群上的内存问题（GC分配失败），Spark作业将始终失败。如果我将计算减少到大约1000万，它肯定会工作 - 不到15分钟。

如何在不增加硬件规格的情况下计算整个集合？如果工作需要更长的时间来运行并且不会中途失败，我会很好。

Answer 1

如果你看看Spark文档，你会发现spark使用不同的数据管理策略。用户可以通过spark配置文件中的配置或直接在代码或脚本中启用这些策略。

关于数据管理政策的文档下方：

“MEMORY_AND_DISK”策略对您有好处，因为如果数据（RDD）不适合ram，则剩余的分区将存储在硬盘中。但是，如果您经常需要访问硬盘，这种策略可能会很慢。

Answer 2

这样做的步骤很少：1。在交叉连接后检查预期的数据量并将其除以200，因为spark.sql.shuffle.partitions默认为200.每个分区必须有超过1 GB的原始数据。 2.计算每个行的大小并乘以另一个表行计数，您将能够估计粗略的体积。与CSV文件3相比，该过程在Parquet中工作得更好.spark.sql.shuffle.partitions需要根据总数据量/ 500 MB设置4. spark.shuffle.minNumPartitionsToHighlyCompress需要设置比Shuffle Partition少一点5.基于两个文件/表的连接列对源镶木地板数据进行打字.6。考虑到堆空间，提供高Spark执行器内存并管理Java堆内存