[SPARK-5932] Updated to use bibibytes in method names. Updated spark.…

…kryoserializer.buffer.mb and spark.reducer.maxMbInFlight
apache · ilganeli · Apr 13, 2015 · Apr 14, 2015 · Apr 14, 2015 · Apr 14, 2015
commit 0cdff352881a2a6c0927c38c52b9574c96b14ce7
diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala
@@ -211,7 +211,74 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
     Utils.timeStringAsMs(get(key, defaultValue))
   }
 
+  /**
+   * Get a size parameter as bytes; throws a NoSuchElementException if it's not set. If no
+   * suffix is provided then bytes are assumed.
+   * @throws NoSuchElementException
+   */
+  def getSizeAsBytes(key: String): Long = {
+    Utils.byteStringAsBytes(get(key))
+  }
+
+  /**``
+   * Get a size parameter as bytes, falling back to a default if not set. If no
+   * suffix is provided then bytes are assumed.
+   */
+  def getSizeAsBytes(key: String, defaultValue: String): Long = {
+    Utils.byteStringAsBytes(get(key, defaultValue))
+  }
+
+  /**
+   * Get a size parameter as Kibibytes; throws a NoSuchElementException if it's not set. If no
+   * suffix is provided then Kibibytes are assumed.
+   * @throws NoSuchElementException
+   */
+  def getSizeAsKb(key: String): Long = {
+    Utils.byteStringAsMb(get(key))
+  }
 
+  /**
+   * Get a size parameter as Kibibytes, falling back to a default if not set. If no
+   * suffix is provided then Kibibytes are assumed.
+   */
+  def getSizeAsKb(key: String, defaultValue: String): Long = {
+    Utils.byteStringAsMb(get(key, defaultValue))
+  }
+
+  /**
+   * Get a size parameter as Mebibytes; throws a NoSuchElementException if it's not set. If no
+   * suffix is provided then Mebibytes are assumed.
+   * @throws NoSuchElementException
+   */
+  def getSizeAsMb(key: String): Long = {
+    Utils.byteStringAsMb(get(key))
+  }
+
+  /**
+   * Get a size parameter as Mebibytes, falling back to a default if not set. If no
+   * suffix is provided then Mebibytes are assumed.
+   */
+  def getSizeAsMb(key: String, defaultValue: String): Long = {
+    Utils.byteStringAsMb(get(key, defaultValue))
+  }
+
+  /**
+   * Get a size parameter as Gibibytes; throws a NoSuchElementException if it's not set. If no
+   * suffix is provided then Gibibytes are assumed.
+   * @throws NoSuchElementException
+   */
+  def getSizeAsGb(key: String): Long = {
+    Utils.byteStringAsGb(get(key))
+  }
+
+  /**
+   * Get a size parameter as Gibibytes, falling back to a default if not set. If no
+   * suffix is provided then Gibibytes are assumed.
+   */
+  def getSizeAsGb(key: String, defaultValue: String): Long = {
+    Utils.byteStringAsGb(get(key, defaultValue))
+  }
+
   /** Get a parameter as an Option */
   def getOption(key: String): Option[String] = {
     Option(settings.get(key)).orElse(getDeprecatedConfig(key, this))
@@ -407,7 +474,13 @@ private[spark] object SparkConf extends Logging {
         "The spark.cache.class property is no longer being used! Specify storage levels using " +
         "the RDD.persist() method instead."),
       DeprecatedConfig("spark.yarn.user.classpath.first", "1.3",
-        "Please use spark.{driver,executor}.userClassPathFirst instead."))
+        "Please use spark.{driver,executor}.userClassPathFirst instead."),
+      DeprecatedConfig("spark.reducer.maxMbInFlight", "1.4",
+        "Please use spark.reducer.maxSizeInFlight instead."),
+      DeprecatedConfig("spark.kryoserializer.buffer.mb", "1.4",
+        "Please use spark.kryoserializer.buffer instead."))
+
+
     Map(configs.map { cfg => (cfg.key -> cfg) }:_*)
   }
 
@@ -431,8 +504,12 @@ private[spark] object SparkConf extends Logging {
     "spark.yarn.am.waitTime" -> Seq(
       AlternateConfig("spark.yarn.applicationMaster.waitTries", "1.3",
         // Translate old value to a duration, with 10s wait time per try.
-        translation = s => s"${s.toLong * 10}s"))
-    )
+        translation = s => s"${s.toLong * 10}s")),
+    "spark.reducer.maxSizeInFlight" -> Seq(
+      AlternateConfig("spark.reducer.maxMbInFlight", "1.4")),
+    "spark.kryoserializer.buffer" -> Seq(
+      AlternateConfig("spark.kryoserializer.buffer.mb", "1.4"))
+  )
 
   /**
    * A view of `configsWithAlternatives` that makes it more efficient to look up deprecated

diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -49,9 +49,9 @@ class KryoSerializer(conf: SparkConf)
   with Logging
   with Serializable {
 
-  private val bufferSizeMb = conf.getDouble("spark.kryoserializer.buffer.mb", 0.064)
+  private val bufferSizeMb = conf.getSizeAsMb("spark.kryoserializer.buffer", "64k")
   if (bufferSizeMb >= 2048) {
-    throw new IllegalArgumentException("spark.kryoserializer.buffer.mb must be less than " +
+    throw new IllegalArgumentException("spark.kryoserializer.buffer must be less than " +
       s"2048 mb, got: + $bufferSizeMb mb.")
   }
   private val bufferSize = (bufferSizeMb * 1024 * 1024).toInt

diff --git a/core/src/main/scala/org/apache/spark/shuffle/hash/BlockStoreShuffleFetcher.scala b/core/src/main/scala/org/apache/spark/shuffle/hash/BlockStoreShuffleFetcher.scala
@@ -79,7 +79,8 @@ private[hash] object BlockStoreShuffleFetcher extends Logging {
       blockManager,
       blocksByAddress,
       serializer,
-      SparkEnv.get.conf.getLong("spark.reducer.maxMbInFlight", 48) * 1024 * 1024)
+      // Note: we use getSizeAsMb to assume Mb when no suffix is provided
+      SparkEnv.get.conf.getSizeAsMb("spark.reducer.maxSizeInFlight", "48m") * 1024 * 1024) 
     val itr = blockFetcherItr.flatMap(unpackBlock)
 
     val completionIter = CompletionIterator[T, Iterator[T]](itr, {

diff --git a/core/src/test/scala/org/apache/spark/DistributedSuite.scala b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
@@ -77,7 +77,7 @@ class DistributedSuite extends FunSuite with Matchers with LocalSparkContext {
   }
 
   test("groupByKey where map output sizes exceed maxMbInFlight") {
-    val conf = new SparkConf().set("spark.reducer.maxMbInFlight", "1")
+    val conf = new SparkConf().set("spark.reducer.maxSizeInFlight", "1m")
     sc = new SparkContext(clusterUrl, "test", conf)
     // This data should be around 20 MB, so even with 4 mappers and 2 reducers, each map output
     // file should be about 2.5 MB

diff --git a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerResizableOutputSuite.scala b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerResizableOutputSuite.scala
@@ -33,7 +33,7 @@ class KryoSerializerResizableOutputSuite extends FunSuite {
   test("kryo without resizable output buffer should fail on large array") {
     val conf = new SparkConf(false)
     conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
-    conf.set("spark.kryoserializer.buffer.mb", "1")
+    conf.set("spark.kryoserializer.buffer", "1m")
     conf.set("spark.kryoserializer.buffer.max.mb", "1")
     val sc = new SparkContext("local", "test", conf)
     intercept[SparkException](sc.parallelize(x).collect())
@@ -43,7 +43,7 @@ class KryoSerializerResizableOutputSuite extends FunSuite {
   test("kryo with resizable output buffer should succeed on large array") {
     val conf = new SparkConf(false)
     conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
-    conf.set("spark.kryoserializer.buffer.mb", "1")
+    conf.set("spark.kryoserializer.buffer", "1m")
     conf.set("spark.kryoserializer.buffer.max.mb", "2")
     val sc = new SparkContext("local", "test", conf)
     assert(sc.parallelize(x).collect() === x)

diff --git a/core/src/test/scala/org/apache/spark/storage/BlockManagerReplicationSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockManagerReplicationSuite.scala
@@ -50,7 +50,7 @@ class BlockManagerReplicationSuite extends FunSuite with Matchers with BeforeAnd
   val allStores = new ArrayBuffer[BlockManager]
 
   // Reuse a serializer across tests to avoid creating a new thread-local buffer on each test
-  conf.set("spark.kryoserializer.buffer.mb", "1")
+  conf.set("spark.kryoserializer.buffer", "1m")
   val serializer = new KryoSerializer(conf)
 
   // Implicitly convert strings to BlockIds for test clarity.

diff --git a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -55,7 +55,7 @@ class BlockManagerSuite extends FunSuite with Matchers with BeforeAndAfterEach
   val shuffleManager = new HashShuffleManager(conf)
 
   // Reuse a serializer across tests to avoid creating a new thread-local buffer on each test
-  conf.set("spark.kryoserializer.buffer.mb", "1")
+  conf.set("spark.kryoserializer.buffer", "1m")
   val serializer = new KryoSerializer(conf)
 
   // Implicitly convert strings to BlockIds for test clarity.

diff --git a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
@@ -96,8 +96,8 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
     assert(Utils.byteStringAsGb("1048575k") === 0)
     assert(Utils.byteStringAsGb("1048576k") === 1)
     assert(Utils.byteStringAsGb("1k") === 0)
-    assert(Utils.byteStringAsGb("1t") === ByteUnit.TB.toGB(1))
-    assert(Utils.byteStringAsGb("1p") === ByteUnit.PB.toGB(1))
+    assert(Utils.byteStringAsGb("1t") === ByteUnit.TiB.toGiB(1))
+    assert(Utils.byteStringAsGb("1p") === ByteUnit.PiB.toGiB(1))
 
     assert(Utils.byteStringAsMb("1") === 1)
     assert(Utils.byteStringAsMb("1m") === 1)
@@ -106,23 +106,23 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
     assert(Utils.byteStringAsMb("1023k") === 0)
     assert(Utils.byteStringAsMb("1024k") === 1)
     assert(Utils.byteStringAsMb("3645k") === 3)
-    assert(Utils.byteStringAsMb("1g") === ByteUnit.GB.toMB(1))
-    assert(Utils.byteStringAsMb("1t") === ByteUnit.TB.toMB(1))
-    assert(Utils.byteStringAsMb("1p") === ByteUnit.PB.toMB(1))
+    assert(Utils.byteStringAsMb("1g") === ByteUnit.GiB.toMiB(1))
+    assert(Utils.byteStringAsMb("1t") === ByteUnit.TiB.toMiB(1))
+    assert(Utils.byteStringAsMb("1p") === ByteUnit.PiB.toMiB(1))
 
     assert(Utils.byteStringAsKb("1") === 1)
     assert(Utils.byteStringAsKb("1k") === 1)
-    assert(Utils.byteStringAsKb("1m") === ByteUnit.MB.toKB(1))
-    assert(Utils.byteStringAsKb("1g") === ByteUnit.GB.toKB(1))
-    assert(Utils.byteStringAsKb("1t") === ByteUnit.TB.toKB(1))
-    assert(Utils.byteStringAsKb("1p") === ByteUnit.PB.toKB(1))
+    assert(Utils.byteStringAsKb("1m") === ByteUnit.MiB.toKiB(1))
+    assert(Utils.byteStringAsKb("1g") === ByteUnit.GiB.toKiB(1))
+    assert(Utils.byteStringAsKb("1t") === ByteUnit.TiB.toKiB(1))
+    assert(Utils.byteStringAsKb("1p") === ByteUnit.PiB.toKiB(1))
 
     assert(Utils.byteStringAsBytes("1") === 1)
-    assert(Utils.byteStringAsBytes("1k") === ByteUnit.KB.toBytes(1))
-    assert(Utils.byteStringAsBytes("1m") === ByteUnit.MB.toBytes(1))
-    assert(Utils.byteStringAsBytes("1g") === ByteUnit.GB.toBytes(1))
-    assert(Utils.byteStringAsBytes("1t") === ByteUnit.TB.toBytes(1))
-    assert(Utils.byteStringAsBytes("1p") === ByteUnit.PB.toBytes(1))
+    assert(Utils.byteStringAsBytes("1k") === ByteUnit.KiB.toBytes(1))
+    assert(Utils.byteStringAsBytes("1m") === ByteUnit.MiB.toBytes(1))
+    assert(Utils.byteStringAsBytes("1g") === ByteUnit.GiB.toBytes(1))
+    assert(Utils.byteStringAsBytes("1t") === ByteUnit.TiB.toBytes(1))
+    assert(Utils.byteStringAsBytes("1p") === ByteUnit.PiB.toBytes(1))
 
     // Test invalid strings
     intercept[NumberFormatException] {

diff --git a/docs/configuration.md b/docs/configuration.md
@@ -366,10 +366,10 @@ Apart from these, the following properties are also available, and may be useful
 <table class="table">
 <tr><th>Property Name</th><th>Default</th><th>Meaning</th></tr>
 <tr>
-  <td><code>spark.reducer.maxMbInFlight</code></td>
-  <td>48</td>
+  <td><code>spark.reducer.maxSizeInFlight</code></td>
+  <td>48m</td>
   <td>
-    Maximum size (in megabytes) of map outputs to fetch simultaneously from each reduce task. Since
+    Maximum size of map outputs to fetch simultaneously from each reduce task. Since
     each output requires us to create a buffer to receive it, this represents a fixed memory
     overhead per reduce task, so keep it small unless you have a large amount of memory.
   </td>
@@ -650,10 +650,10 @@ Apart from these, the following properties are also available, and may be useful
   </td>
 </tr>
 <tr>
-  <td><code>spark.kryoserializer.buffer.mb</code></td>
-  <td>0.064</td>
+  <td><code>spark.kryoserializer.buffer</code></td>
+  <td>64k</td>
   <td>
-    Initial size of Kryo's serialization buffer, in megabytes. Note that there will be one buffer
+    Initial size of Kryo's serialization buffer. Note that there will be one buffer
      <i>per core</i> on each worker. This buffer will grow up to
      <code>spark.kryoserializer.buffer.max.mb</code> if needed.
   </td>

diff --git a/docs/tuning.md b/docs/tuning.md
@@ -60,7 +60,7 @@ val sc = new SparkContext(conf)
 The [Kryo documentation](https://github.com/EsotericSoftware/kryo) describes more advanced
 registration options, such as adding custom serialization code.
 
-If your objects are large, you may also need to increase the `spark.kryoserializer.buffer.mb`
+If your objects are large, you may also need to increase the `spark.kryoserializer.buffer`
 config property. The default is 2, but this value needs to be large enough to hold the *largest*
 object you will serialize.
 

diff --git a/examples/src/main/scala/org/apache/spark/examples/mllib/MovieLensALS.scala b/examples/src/main/scala/org/apache/spark/examples/mllib/MovieLensALS.scala
@@ -100,7 +100,7 @@ object MovieLensALS {
     val conf = new SparkConf().setAppName(s"MovieLensALS with $params")
     if (params.kryo) {
       conf.registerKryoClasses(Array(classOf[mutable.BitSet], classOf[Rating]))
-        .set("spark.kryoserializer.buffer.mb", "8")
+        .set("spark.kryoserializer.buffer", "8m")
     }
     val sc = new SparkContext(conf)
 

diff --git a/network/common/src/main/java/org/apache/spark/network/util/ByteUnit.java b/network/common/src/main/java/org/apache/spark/network/util/ByteUnit.java
@@ -25,46 +25,46 @@ public enum ByteUnit {
   },
 
   /** Kibibyte (KiB) = 1024 Byte */
-  KB {
-    public long toBytes(long d) { return x(d, C_KB); }
+  KiB {
+    public long toBytes(long d) { return x(d, C_KiB); }
 
-    public long convert(long d, ByteUnit u) { return u.toKB(d); }
+    public long convert(long d, ByteUnit u) { return u.toKiB(d); }
   },
 
   /** Mebibyte (MiB) = (1024^2) Byte */
-  MB {
-    public long toBytes(long d) { return x(d, C_MB); }
+  MiB {
+    public long toBytes(long d) { return x(d, C_MiB); }
 
-    public long convert(long d, ByteUnit u) { return u.toMB(d); }
+    public long convert(long d, ByteUnit u) { return u.toMiB(d); }
   },
 
   /** Gibibyte (GiB) = (1024^3) Byte */
-  GB {
-    public long toBytes(long d) { return x(d, C_GB);
+  GiB {
+    public long toBytes(long d) { return x(d, C_GiB);
     }
 
-    public long convert(long d, ByteUnit u) { return u.toGB(d); }
+    public long convert(long d, ByteUnit u) { return u.toGiB(d); }
   },
 
   /** Tebibyte (TiB) = (1024^4) Byte */
-  TB {
-    public long toBytes(long d) { return x(d, C_TB); }
+  TiB {
+    public long toBytes(long d) { return x(d, C_TiB); }
 
-    public long convert(long d, ByteUnit u) { return u.toTB(d); }
+    public long convert(long d, ByteUnit u) { return u.toTiB(d); }
   },
 
   /** Pebibyte (PB) = (1024^5) Byte */
-  PB {
-    public long toBytes(long d) { return x(d, C_PB); }
+  PiB {
+    public long toBytes(long d) { return x(d, C_PiB); }
 
-    public long convert(long d, ByteUnit u) { return u.toPB(d); }
+    public long convert(long d, ByteUnit u) { return u.toPiB(d); }
   };
 
-  static final long C_KB = 1024l;
-  static final long C_MB = (long) Math.pow(1024l, 2l);
-  static final long C_GB = (long) Math.pow(1024l, 3l);
-  static final long C_TB = (long) Math.pow(1024l, 4l);
-  static final long C_PB = (long) Math.pow(1024l, 5l);
+  static final long C_KiB = 1024l;
+  static final long C_MiB = (long) Math.pow(1024l, 2l);
+  static final long C_GiB = (long) Math.pow(1024l, 3l);
+  static final long C_TiB = (long) Math.pow(1024l, 4l);
+  static final long C_PiB = (long) Math.pow(1024l, 5l);
 
   static final long MAX = Long.MAX_VALUE;
 
@@ -78,19 +78,18 @@ static long x(long d, long m) {
     if (d < -over) return Long.MIN_VALUE;
     return d * m;
   }
+
+  public long convert(long d, ByteUnit u) { throw new AbstractMethodError(); }
 
   public long toBytes(long d) { throw new AbstractMethodError(); }
-  public long convert(long d, ByteUnit u) { throw new AbstractMethodError(); }
 
-  public long toKB(long d) { return toBytes(d) / C_KB; }
+  public long toKiB(long d) { return toBytes(d) / C_KiB; }
 
-  public long toMB(long d) { return toBytes(d) / C_MB; }
+  public long toMiB(long d) { return toBytes(d) / C_MiB; }
 
-  public long toGB(long d) { return toBytes(d) / C_GB; }
+  public long toGiB(long d) { return toBytes(d) / C_GiB; }
 
-  public long toTB(long d) { return toBytes(d) / C_TB; }
+  public long toTiB(long d) { return toBytes(d) / C_TiB; }
 
-  public long toPB(long d) { return toBytes(d) / C_PB; }
-
-
+  public long toPiB(long d) { return toBytes(d) / C_PiB; }
 }