我有一个 csv 文件存储了维度 6,365x214
的用户项数据,我通过使用 的 columnSimilarities() 找到用户与用户的相似性>org.apache.spark.mllib.linalg.distributed.CoordinateMatrix
。
我的代码是这样的:
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.linalg.distributed.{RowMatrix,
MatrixEntry, CoordinateMatrix}
import org.apache.spark.rdd.RDD
def rddToCoordinateMatrix(input_rdd: RDD[String]) : CoordinateMatrix = {
// Convert RDD[String] to RDD[Tuple3]
val coo_matrix_input: RDD[Tuple3[Long,Long,Double]] = input_rdd.map(
line => line.split(',').toList
).map{
e => (e(0).toLong, e(1).toLong, e(2).toDouble)
}
// Convert RDD[Tuple3] to RDD[MatrixEntry]
val coo_matrix_matrixEntry: RDD[MatrixEntry] = coo_matrix_input.map(e => MatrixEntry(e._1, e._2, e._3))
// Convert RDD[MatrixEntry] to CoordinateMatrix
val coo_matrix: CoordinateMatrix = new CoordinateMatrix(coo_matrix_matrixEntry)
return coo_matrix
}
// Read CSV File to RDD[String]
val input_rdd: RDD[String] = sc.textFile("user_item.csv")
// Read RDD[String] to CoordinateMatrix
val coo_matrix = rddToCoordinateMatrix(input_rdd)
// Transpose CoordinateMatrix
val coo_matrix_trans = coo_matrix.transpose()
// Convert CoordinateMatrix to RowMatrix
val mat: RowMatrix = coo_matrix_trans.toRowMatrix()
// Compute similar columns perfectly, with brute force
// Return CoordinateMatrix
val simsPerfect: CoordinateMatrix = mat.columnSimilarities()
// CoordinateMatrix to RDD[MatrixEntry]
val simsPerfect_entries = simsPerfect.entries
simsPerfect_entries.count()
// Write results to file
val results_rdd = simsPerfect_entries.map(line => line.i+","+line.j+","+line.value)
results_rdd.saveAsTextFile("similarity-output")
// Close the REPL terminal
System.exit(0)
并且,当我在 spark-shell 上运行这个脚本时
在运行代码 simsPerfect_entries.count()
行后,我得到以下错误:
java.lang.OutOfMemoryError: GC overhead limit exceeded
更新:
我尝试了很多其他人已经给出的解决方案,但我没有成功。
1 通过增加每个执行程序进程使用的内存量 spark.executor.memory=1g
2 通过减少用于驱动程序进程的内核数量
spark.driver.cores=1
给我一些解决这个问题的方法。
最佳答案
所有 Spark 转换都是惰性的,直到您真正实现它。当您定义 RDD 到 RDD 的数据操作时,Spark 只是将操作链接在一起,而不执行实际计算。因此,当您调用 simsPerfect_entries.count()
时,将执行操作链并获得您的号码。
错误 GC overhead limit exceeded
表示 JVM 垃圾收集器事件非常频繁,以至于停止执行代码。由于以下原因,GC 事件可能会如此之高:
- 您生产了太多小 object 并立即丢弃它们。看起来你不是。
- 您的数据不适合您的 JVM 堆。就像您尝试将 2GB 的文本文件加载到 RAM 中,但只有 1GB 的 JVM 堆。看起来是你的情况。
要解决此问题,请尝试增加 JVM 堆的数量:
- 您的工作节点(如果您有分布式 Spark 设置)。
- 您的 spark-shell 应用。
关于scala - Apache Spark 中的大型 RDD [MatrixEntry] 超出了 GC 开销限制,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/30251223/