Changing the JavaSerializer reset to occur every 1000 objects.

Author: kellrott

core/src/main/scala/org/apache/spark/serializer/JavaSerializer.scala

@@ -24,8 +24,18 @@ import org.apache.spark.util.ByteBufferInputStream
 
 private[spark] class JavaSerializationStream(out: OutputStream) extends SerializationStream {
   val objOut = new ObjectOutputStream(out)
+  var counter = 0;
   //Calling reset to avoid memory leak: http://stackoverflow.com/questions/1281549/memory-leak-traps-in-the-java-standard-api
-  def writeObject[T](t: T): SerializationStream = { objOut.writeObject(t); objOut.reset(); this }
+  def writeObject[T](t: T): SerializationStream = {
+    objOut.writeObject(t);
+    if (counter >= 1000) {
+      objOut.reset();
+      counter = 0;
+    } else {
+      counter+=1;
+    }
+    this
+  }
   def flush() { objOut.flush() }
   def close() { objOut.close() }
 }