如何根据基于Spark中另一个RDD的函数过滤RDD？

Question

我是Apache Spark的初学者。我想过滤出所有在RDD中权重总和大于常数的组。 “体重”图也是RDD。这里是一个小尺寸的演示中，待过滤的被存储在“基团”的基团，所述恒定值是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堆内存不足”错误。我怀疑它是否是由“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，我得到了一个“显示java.lang.NullPointerException”错误。有人告诉我，在另一个RDD中操作RDD时，会出现空指针异常，并建议我将权重放入Redis。

那么我怎样才能得到“结果”没有转换“权重”地图，或把它放到Redis？如果有一种解决方案可以在没有外部数据存储服务的帮助下基于另一个类似地图的RDD过滤RDD？ 谢谢！

Answer 1

“java内存不足”错误即将到来，因为spark在使用spark.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执行内存大小。

Answer 2

假设你的组是唯一的。否则，首先通过不同的方式使其具有唯一性。 如果组或权重很小，应该很容易。如果组和权重都很大，您可以尝试一下，这可能更具可扩展性，但看起来也很复杂。

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)