3

我是 Apache Spark 的初学者。我想过滤掉所有权重总和大于 RDD 中的常数值的组。“权重”映射也是一个 RDD。这里是一个小型的demo，需要过滤的组存储在“groups”中，常量值为12：

val groups = sc.parallelize(List("a,b,c,d", "b,c,e", "a,c,d", "e,g"))
val weights = sc.parallelize(Array(("a", 3), ("b", 2), ("c", 5), ("d", 1), ("e", 9), ("f", 4), ("g", 6)))
val wm = weights.toArray.toMap
def isheavy(inp: String): Boolean = {
  val allw = inp.split(",").map(wm(_)).sum
  allw > 12
}
val result = groups.filter(isheavy)

当输入数据非常大时，例如> 10GB，我总是遇到“java heap out of memory”错误。我怀疑它是否是由“weights.toArray.toMap”引起的，因为它将分布式 RDD 转换为 JVM 中的 Java 对象。所以我尝试直接用RDD过滤：

val groups = sc.parallelize(List("a,b,c,d", "b,c,e", "a,c,d", "e,g"))
val weights = sc.parallelize(Array(("a", 3), ("b", 2), ("c", 5), ("d", 1), ("e", 9), ("f", 4), ("g", 6)))
def isheavy(inp: String): Boolean = {
  val items = inp.split(",")
  val wm = items.map(x => weights.filter(_._1 == x).first._2)
  wm.sum > 12
}
val result = groups.filter(isheavy)

当我result.collect将此脚本加载到 spark shell 后运行时，出现“java.lang.NullPointerException”错误。有人告诉我，当一个 RDD 在另一个 RDD 中被操作时，会出现空指针异常，并建议我将权重放到 Redis 中。

那么如何在不将“权重”转换为 Map 或将其放入 Redis 的情况下获得“结果”？如果没有外部数据存储服务的帮助，是否有一种解决方案可以基于另一个类似地图的 RDD 过滤 RDD？谢谢！

4

2 回答 2

4

假设您的组是独一无二的。否则，首先通过 distinct 等使其唯一。如果 group 或 weights 很小，应该很容易。如果 group 和 weights 都很大，你可以试试这个，它可能更具可扩展性，但看起来也很复杂。

val groups = sc.parallelize(List("a,b,c,d", "b,c,e", "a,c,d", "e,g"))
val weights = sc.parallelize(Array(("a", 3), ("b", 2), ("c", 5), ("d", 1), ("e", 9), ("f", 4), ("g", 6)))
//map groups to be (a, (a,b,c,d)), (b, (a,b,c,d), (c, (a,b,c,d)....
val g1=groups.flatMap(s=>s.split(",").map(x=>(x, Seq(s))))
//j will be (a, ((a,b,c,d),3)...
val j = g1.join(weights)
//k will be ((a,b,c,d), 3), ((a,b,c,d),2) ...
val k = j.map(x=>(x._2._1, x._2._2))
//l will be ((a,b,c,d), (3,2,5,1))...
val l = k.groupByKey()
//filter by sum the 2nd
val m = l.filter(x=>{var sum = 0; x._2.foreach(a=> {sum=sum+a});sum > 12})
//we only need the original list
val result=m.map(x=>x._1)
//don't do this in real product, otherwise, all results go to driver.instead using saveAsTextFile, etc
scala> result.foreach(println)
List(e,g)
List(b,c,e)
于 2014-09-26T03:20:08.803 回答
2

“java out of memory”错误即将到来，因为 sparkspark.default.parallelism在确定拆分数时使用其属性，默认情况下是可用的内核数。

// From CoarseGrainedSchedulerBackend.scala

override def defaultParallelism(): Int = {
   conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))
}

当输入变大并且您的内存有限时，您应该增加拆分次数。

您可以执行以下操作：

 val input = List("a,b,c,d", "b,c,e", "a,c,d", "e,g") 
 val splitSize = 10000 // specify some number of elements that fit in memory.

 val numSplits = (input.size / splitSize) + 1 // has to be > 0.
 val groups = sc.parallelize(input, numSplits) // specify the # of splits.

 val weights = Array(("a", 3), ("b", 2), ("c", 5), ("d", 1), ("e", 9), ("f", 4), ("g", 6)).toMap

 def isHeavy(inp: String) = inp.split(",").map(weights(_)).sum > 12
 val result = groups.filter(isHeavy)

您也可以考虑使用spark.executor.memory.

于 2014-09-25T14:00:38.503 回答