ShapeRDD: paralell mapPartition

Author: zouzias

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/rdds/ShapeRDD.scala

@@ -112,13 +112,9 @@ class ShapeRDD[K: ClassTag, V: ClassTag]
 
     logDebug("Compute topK linkage per partition")
     val resultsByPart: RDD[(ShapeItemUUID, TopK[SparkScoreDoc])] = partitionsRDD.flatMap {
-      case partition => queriesB.value.zipWithIndex.map { case (queryPoint, index) =>
-        val results = mapper(queryPoint, partition).map(x => topKMonoid.build(x))
-          .reduceOption(topKMonoid.plus)
-          .getOrElse(topKMonoid.zero)
-
-        (index.toLong, results)
-      }
+      case partition => queriesB.value.zipWithIndex.par.map { case (queryPoint, index) =>
+        (index.toLong, topKMonoid.build(mapper(queryPoint, partition)))
+      }.toIterator
     }
 
     logDebug("Merge topK linkage results")