MAHOUT-1785: Replace 'spark.kryoserializer.buffer.mb' from Spark conf…

…ig with 'spark.kryoserializer.buffer'
apache · Nov 8, 2015 · ec10963 · ec10963
1 parent c2b077f
commit ec10963
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 7 deletions.
diff --git a/spark/src/main/scala/org/apache/mahout/drivers/MahoutSparkDriver.scala b/spark/src/main/scala/org/apache/mahout/drivers/MahoutSparkDriver.scala
@@ -75,7 +75,11 @@ abstract class MahoutSparkDriver extends MahoutDriver {
   override protected def start() : Unit = {
     if (!_useExistingContext) {
       sparkConf.set("spark.kryo.referenceTracking", "false")
-        .set("spark.kryoserializer.buffer.mb", "200")// this is default for Mahout optimizer, change it with -D option
+        .set("spark.kryoserializer.buffer.mb", "200m")// this is default for Mahout optimizer, change it with -D option
+        // the previous has been marked deprecated as of Spark 1.4 by the below line,
+        // remove the above line when Spark finally retires above for below
+        .set("spark.kryoserializer.buffer", "200m")
+
 
       if (parser.opts("sparkExecutorMem").asInstanceOf[String] != "")
         sparkConf.set("spark.executor.memory", parser.opts("sparkExecutorMem").asInstanceOf[String])

diff --git a/spark/src/test/scala/org/apache/mahout/sparkbindings/test/DistributedSparkSuite.scala b/spark/src/test/scala/org/apache/mahout/sparkbindings/test/DistributedSparkSuite.scala
@@ -18,12 +18,13 @@
 package org.apache.mahout.sparkbindings.test
 
 import org.apache.log4j.{Level, Logger}
-import org.scalatest.{ConfigMap, BeforeAndAfterAllConfigMap, Suite}
-import org.apache.spark.SparkConf
-import org.apache.mahout.sparkbindings._
-import org.apache.mahout.test.{DistributedMahoutSuite, MahoutSuite}
 import org.apache.mahout.math.drm.DistributedContext
-import collection.JavaConversions._
+import org.apache.mahout.sparkbindings._
+import org.apache.mahout.test.DistributedMahoutSuite
+import org.apache.spark.SparkConf
+import org.scalatest.{ConfigMap, Suite}
+
+import scala.collection.JavaConversions._
 
 trait DistributedSparkSuite extends DistributedMahoutSuite with LoggerConfiguration {
   this: Suite =>
@@ -39,7 +40,8 @@ trait DistributedSparkSuite extends DistributedMahoutSuite with LoggerConfigurat
       // Do not run MAHOUT_HOME jars in unit tests.
       addMahoutJars = !isLocal,
       sparkConf = new SparkConf()
-          .set("spark.kryoserializer.buffer.mb", "40")
+          .set("spark.kryoserializer.buffer.mb", "40m")
+          .set("spark.kryoserializer.buffer", "40m")
           .set("spark.akka.frameSize", "30")
           .set("spark.default.parallelism", "10")
           .set("spark.executor.memory", "2G")