我正在尝试实现一个在Spark中运行良好的Hadoop Map/Reduce作业。Spark应用程序定义如下:
val data = spark.textFile(file, 2).cache()
val result = data
.map(//some pre-processing)
.map(docWeightPar => (docWeightPar(0),docWeightPar(1))))
.flatMap(line => MyFunctions.combine(line))
.reduceByKey( _ + _)
我的函数 MyFunctions.combine
在哪里
def combine(tuples: Array[(String, String)]): IndexedSeq[(String,Double)] =
for (i <- 0 to tuples.length - 2;
j <- 1 to tuples.length - 1
) yield (toKey(tuples(i)._1,tuples(j)._1),tuples(i)._2.toDouble * tuples(j)._2.toDouble)
combine
函数生成的键值对数量取决于输入列表的大小,当输入列表很大时,可能会抛出异常。在Hadoop Map Reduce环境中,我没有遇到问题,因为combine
函数产生输出结果的时候,Hadoop会将映射键值对写入磁盘。但是Spark似乎会一直在内存中保存,直到遇到java.lang.OutOfMemoryError: GC overhead limit exceeded
错误。我可能做错了一些基本的东西,但是我找不到解决方法。由于我对Scala和Spark都不太了解,所以我不确定问题是由其中一个还是两个共同导致的。目前我正在尝试在自己的笔记本电脑上运行这个程序,对于长度不太长的tuples
数组输入,它可以正常工作。