Comment résoudre l'erreur Apache Spark StackOverflowError après plusieurs unions ?

Question

J'ai un programme Spark Scala qui utilise une API REST pour obtenir des données lot par lot, et une fois que toutes les données sont récupérées, je les traite.

Programme actuel :

• Pour chaque lot, créer un RDD et le fusionner avec le RDD précédent. créé à l'aide de l'appel API précédent rdd.union(currentRdd) .

• Opérer sur le RDD final

Un programme simple pour reproduire le problème :

    def main(args: Array[String]) = {
     val conf = new SparkConf().setAppName("Union test").setMaster("local[1]")
     val sc = new SparkContext(conf)
     val limit = 1000;
     var rdd = sc.emptyRDD[Int]
     for (x <- 1 to limit) {
       val currentRdd = sc.parallelize(x to x + 3)
       rdd = rdd.union(currentRdd)
     }
     println(rdd.sum())
   }

Problème : - Lorsque le nombre de lots est élevé, le programme lance un StackOverflowError : Exception in thread "main" java.lang.StackOverflowError at org.apache.spark.rdd.UnionRDD$$anonfun$1.apply

Je suppose que lorsque le nombre de lots augmente, le graphe de dépendance RDD devient vraiment complexe et provoque l'erreur.

Quelle est la meilleure façon de résoudre ce problème ?

Answer 1

Il existe déjà SparkContext.union qui sait comment calculer correctement un union de plusieurs RDD s :

val rdds = List.tabulate(limit + 1)(x => sc.parallelize(x to x + 3))
val rdd = sc.union(rdds)

Vous pouvez également essayer d'utiliser cette pour éviter la création d'une longue chaîne de fonctions d'aide. union s :

val rdds = List.tabulate(limit + 1)(x => sc.parallelize(x to x + 3))
val rdd = balancedReduce(rdds)(_ union _)

La raison pour laquelle cela devrait fonctionner est essentiellement la même que dans la réponse liée : O(n) chaîne de union souffle la pile, O(log(n)) -Arbre binaire élevé de union ne le fait pas.