performance tuning

Author: Dubovsky Jakub

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark
 
+import java.util.Arrays
+
 import scala.language.implicitConversions
 
 import java.io._
@@ -1409,7 +1411,7 @@ object SparkContext extends Logging {
     simpleWritableConverter[Boolean, BooleanWritable](_.get)
 
   implicit def bytesWritableConverter(): WritableConverter[Array[Byte]] = {
-    simpleWritableConverter[Array[Byte], BytesWritable](bw => bw.getBytes.take(bw.getLength))
+    simpleWritableConverter[Array[Byte], BytesWritable](bw => Arrays.copyOfRange(bw.getBytes, 0, bw.getLength))
   }
 
   implicit def stringWritableConverter(): WritableConverter[String] =

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -24,11 +24,16 @@ import org.apache.hadoop.io.BytesWritable
 class SparkContextSuite extends FunSuite {
   test("test of writing spark scala test") {
     val bytesWritable = new BytesWritable()
-    bytesWritable.set((1 to 10).map(_.toByte).toArray, 0, 10)
-    bytesWritable.set((1 to 5).map(_.toByte).toArray, 0, 5)
+    val inputArray = (1 to 10).map(_.toByte).toArray
+    bytesWritable.set(inputArray, 0, 10)
+    bytesWritable.set(inputArray, 0, 5)
 
     val converter = SparkContext.bytesWritableConverter()
     val byteArray = converter.convert(bytesWritable)
     assert(byteArray.length === 5)
+
+    bytesWritable.set(inputArray, 0, 0)
+    val byteArray2 = converter.convert(bytesWritable)
+    assert(byteArray2.length === 0)
   }
 }