Geek Answers Handbook

How to parallelize a Spark scala calculation?

I have code to calculate inside Set Set of Squared Error after clustering, which I basically took from Spark mllib source code.

When I run similar code using the spark API, it runs in many different (distributed) jobs and runs successfully. When I run my code on it (which should do the same as the Spark code), I get an error. Any ideas why?

Here is the code:

import java.util.Arrays
        import org.apache.spark.mllib.linalg.{Vectors, Vector}
        import org.apache.spark.mllib.linalg._
        import org.apache.spark.mllib.linalg.distributed.RowMatrix
        import org.apache.spark.rdd.RDD
        import org.apache.spark.api.java.JavaRDD
        import breeze.linalg.{axpy => brzAxpy, inv, svd => brzSvd, DenseMatrix => BDM, DenseVector => BDV,
          MatrixSingularException, SparseVector => BSV, CSCMatrix => BSM, Matrix => BM}

        val EPSILON = {
            var eps = 1.0
            while ((1.0 + (eps / 2.0)) != 1.0) {
              eps /= 2.0
            }
            eps
          }

        def dot(x: Vector, y: Vector): Double = {
            require(x.size == y.size,
              "BLAS.dot(x: Vector, y:Vector) was given Vectors with non-matching sizes:" +
              " x.size = " + x.size + ", y.size = " + y.size)
            (x, y) match {
              case (dx: DenseVector, dy: DenseVector) =>
                dot(dx, dy)
              case (sx: SparseVector, dy: DenseVector) =>
                dot(sx, dy)
              case (dx: DenseVector, sy: SparseVector) =>
                dot(sy, dx)
              case (sx: SparseVector, sy: SparseVector) =>
                dot(sx, sy)
              case _ =>
                throw new IllegalArgumentException(s"dot doesn't support (${x.getClass}, ${y.getClass}).")
            }
         }

         def fastSquaredDistance(
              v1: Vector,
              norm1: Double,
              v2: Vector,
              norm2: Double,
              precision: Double = 1e-6): Double = {
            val n = v1.size
            require(v2.size == n)
            require(norm1 >= 0.0 && norm2 >= 0.0)
            val sumSquaredNorm = norm1 * norm1 + norm2 * norm2
            val normDiff = norm1 - norm2
            var sqDist = 0.0
            /*
             * The relative error is
             * <pre>
             * EPSILON * ( \|a\|_2^2 + \|b\\_2^2 + 2 |a^T b|) / ( \|a - b\|_2^2 ),
             * </pre>
             * which is bounded by
             * <pre>
             * 2.0 * EPSILON * ( \|a\|_2^2 + \|b\|_2^2 ) / ( (\|a\|_2 - \|b\|_2)^2 ).
             * </pre>
             * The bound doesn't need the inner product, so we can use it as a sufficient condition to
             * check quickly whether the inner product approach is accurate.
             */
            val precisionBound1 = 2.0 * EPSILON * sumSquaredNorm / (normDiff * normDiff + EPSILON)
            if (precisionBound1 < precision) {
              sqDist = sumSquaredNorm - 2.0 * dot(v1, v2)
            } else if (v1.isInstanceOf[SparseVector] || v2.isInstanceOf[SparseVector]) {
              val dotValue = dot(v1, v2)
              sqDist = math.max(sumSquaredNorm - 2.0 * dotValue, 0.0)
              val precisionBound2 = EPSILON * (sumSquaredNorm + 2.0 * math.abs(dotValue)) /
                (sqDist + EPSILON)
              if (precisionBound2 > precision) {
                sqDist = Vectors.sqdist(v1, v2)
              }
            } else {
              sqDist = Vectors.sqdist(v1, v2)
            }
            sqDist
        }

        def findClosest(
              centers: TraversableOnce[Vector],
              point: Vector): (Int, Double) = {
            var bestDistance = Double.PositiveInfinity
            var bestIndex = 0
            var i = 0
            centers.foreach { center =>
              // Since `\|a - b\| \geq |\|a\| - \|b\||`, we can use this lower bound to avoid unnecessary
              // distance computation.
              var lowerBoundOfSqDist = Vectors.norm(center, 2.0) - Vectors.norm(point, 2.0)
              lowerBoundOfSqDist = lowerBoundOfSqDist * lowerBoundOfSqDist
              if (lowerBoundOfSqDist < bestDistance) {
                val distance: Double = fastSquaredDistance(center, Vectors.norm(center, 2.0), point, Vectors.norm(point, 2.0))
                if (distance < bestDistance) {
                  bestDistance = distance
                  bestIndex = i
                }
              }
              i += 1
            }
            (bestIndex, bestDistance)
        }

         def pointCost(
              centers: TraversableOnce[Vector],
              point: Vector): Double =
            findClosest(centers, point)._2



        def clusterCentersIter: Iterable[Vector] =
            clusterCenters.map(p => p)


        def computeCostZep(indata: RDD[Vector]): Double = {
            val bcCenters = indata.context.broadcast(clusterCenters)
            indata.map(p => pointCost(bcCenters.value, p)).sum()
          }

        computeCostZep(projectedData)

I believe that I use all the same parallelization tasks as the spark, but this does not work for me. Any tips on distributing my code / help on why memory overflow occurs in my code would be very helpful

Here is a link to the source code in a spark that is very similar: KMeansModel and KMeans

, :

val clusters = KMeans.train(projectedData, numClusters, numIterations)

val clusterCenters = clusters.clusterCenters




// Evaluate clustering by computing Within Set Sum of Squared Errors
val WSSSE = clusters.computeCost(projectedData)
println("Within Set Sum of Squared Errors = " + WSSSE)

:

org.apache.spark.SparkException: - : 1 94.0 4 , : 1.3 94.0 (TID 37663, ip-172-31-13-209.ec2.internal): java.lang.StackOverflowError at $iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$$$$$ c57ec8bf9b0d5f6161b97741d596ff0 $$$$ Wc $$ IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ iwC $$ iwC.dot(: 226) at $iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$$$$$ c57ec8bf9b0d5f6161b97741d596ff0 $$$$ Wc $$ IWC $$ IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ iwC.dot(: 226) ...

:

stacktrace: at org.apache.spark.scheduler.DAGScheduler.org $apache $spark $scheduler $DAGScheduler $$ failJobAndIndependentStages (DAGScheduler.scala: 1431) at org.apache.spark.scheduler.DAGScheduler $$ anonfun $abortStage $1.apply(DAGScheduler.scala: 1419) org.apache.spark.scheduler.DAGScheduler $$ anonfun $abortStage $1.apply(DAGScheduler.scala: 1418) scala.collection.mutable.ResizableArray $class.foreach(ResizableArray.scala: 59) scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala: 47) org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala: 1418) org.apache. spark.scheduler.DAGScheduler $$ anonfun $handleTaskSetFailed $1.apply(DAGScheduler.scala: 799) org.apache.spark.scheduler.DAGScheduler $$ anonfun $handleTaskSetFailed $1.apply(DAGScheduler.scala: 799) scala. Option.foreach(Option.scala: 236) org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala: 799) org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGSchedul er.scala: 1640) org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala: 1599) org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala: 1588) org.apache.spark.util.EventLoop $$ anon $1.run(EventLoop.scala: 48) org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala: 620) org.apache.spark.SparkContext.runJob(SparkContext.scala: 1832) org.apache.spark.SparkContext.runJob(SparkContext.scala: 1952) org.apache.spark.rdd.RDD $$ anonfun $fold $1.apply(RDD.scala: 1088) org.apache. spark.rdd.RDDOperationScope $.withScope(RDDOperationScope.scala: 150) org.apache.spark.rdd.RDDOperationScope $.withScope(RDDOperationScope.scala: 111) org.apache.spark.rdd.RDD.withScope(RDD. scala: 316) org.apache.spark.rdd.RDD.fold(RDD.scala: 1082) org.apache.spark.rdd.DoubleRDDFunctions $$ anonfun $sum $1.apply $mcD $sp (DoubleRDDFunctions.scala: 34) at org.apache.spark.rdd.DoubleRDDFunctions $$ anonfun $sum $1.apply(DoubleRDDFunctions.scala: 34) a t org.apache.spark.rdd.DoubleRDDFunctions $$ anonfun $sum $1.apply(DoubleRDDFunctions.scala: 34) org.apache.spark.rdd.RDDOperationScope $.withScope(RDDOperationScope.scala: 150) org.apache. spark.rdd.RDDOperationScope $.withScope(RDDOperationScope.scala: 111) org.apache.spark.rdd.RDD.withScope(RDD.scala: 316) org.apache.spark.rdd.DoubleRDDFunctions.sum(DoubleRDDFunctions.scala: 33)

+4
1

, : dot :

def dot(x: Vector, y: Vector): Double = {
        require(x.size == y.size,
          "BLAS.dot(x: Vector, y:Vector) was given Vectors with non-matching sizes:" +
          " x.size = " + x.size + ", y.size = " + y.size)
        (x, y) match {
          case (dx: DenseVector, dy: DenseVector) =>
            dot(dx, dy)
          case (sx: SparseVector, dy: DenseVector) =>
            dot(sx, dy)
          case (dx: DenseVector, sy: SparseVector) =>
            dot(sy, dx)
          case (sx: SparseVector, sy: SparseVector) =>
            dot(sx, sy)
          case _ =>
            throw new IllegalArgumentException(s"dot doesn't support (${x.getClass}, ${y.getClass}).")
        }
     }

dot , , .

stacktrace - , :

java.lang.StackOverflowError at $iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$ iwC $$$$$$ c57ec8bf9b0d5f6161b97741d596ff0 $$ $$ Wc $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ $$ IWC IWC IWC $$ $$ iwC.dot (: 226)

+2

More articles:


All Articles