Comment filtrer un RDD selon une fonction de base d'un autre RDD dans Spark?

Je suis un débutant de Apache Spark. Je veux filtrer tous les groupes dont la somme des poids est supérieure à une valeur constante dans un RDD. Le "poids" de la carte est aussi un RDD. Voici une petite taille de la démo, les groupes à être filtrée est stocké dans des "groupes", la valeur de la constante est de 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)

Lorsque les données d'entrée est très grande, > 10 GO par exemple, j'ai toujours la rencontre d'une "java tas de message d'erreur" mémoire. Je doute que si elle est causée par le "poids".toArray.toMap", parce qu'il convertir un distribué des RDD à un objet Java dans la JVM. J'ai donc essayé de filtre avec RDD directement:

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)

Quand j'ai couru result.collect après le chargement de ce script dans spark shell, j'ai eu un "java.lang.NullPointerException" erreur. Quelqu'un m'a dit quand un RDD est manipulé dans un autre RDD, il y aura un nullpointer exception, et de me suggérer de mettre le poids dans le Redis.

Alors, comment puis-je obtenir le "résultat" sans convertir "poids" de la Carte, ou de le mettre dans le Redis? Si il y a une solution pour filtrer un RDD basée sur une autre carte comme CA sans l'aide extérieure de la banque de données de service?
Merci!

OriginalL'auteur Chad | 2014-09-25

  1. 2

    La java "out of memory" l'erreur est venue parce que l'étincelle utilise son spark.default.parallelism propriété, tandis que la détermination de nombre de divisions, qui par défaut est le nombre de cœurs disponibles.

    //From CoarseGrainedSchedulerBackend.scala

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

    Lorsque l'entrée devient grand, et vous avez peu de mémoire, vous devez augmenter le nombre de divisions.

    Vous pouvez faire quelque chose comme suit:

     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)

    Vous pouvez également envisager d'augmenter exécuteur de la taille de la mémoire à l'aide de spark.executor.memory.

    agrandir spark.executor.memory fonctionne.

    OriginalL'auteur Shyamendra Solanki

  2. 4

    Supposons que votre groupe est unique. Sinon, le rendre unique distincts, etc.
    Si le groupe ou le poids est faible, il devrait être facile. Si à la fois le groupe et les poids sont énormes, vous pouvez essayer ce qui peut être plus évolutif, mais aussi semble compliqué.

    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)

    OriginalL'auteur zhang zhan