Fixed Scala style

Author: bijaybisht

streaming/src/main/scala/org/apache/spark/streaming/binning/BinStream.scala

@@ -21,6 +21,7 @@ import scala.reflect.ClassTag
 
 import org.apache.spark.streaming.dstream.DStream
 
-class BinStream[T: ClassTag](@transient ds: DStream[T], sizeInNumBatches: Int, delayInNumBatches: Int) {
+class BinStream[T: ClassTag](
+  @transient ds: DStream[T], sizeInNumBatches: Int, delayInNumBatches: Int) {
   def getDStream = ds
 }

streaming/src/main/scala/org/apache/spark/streaming/binning/BinStreamer.scala

@@ -22,7 +22,9 @@ import org.apache.spark.streaming.Time
 import scala.reflect.ClassTag
 
 
-class BinStreamer[T: ClassTag](@transient ds: DStream[T], getStartTime: (T) => Time, getEndTime: (T) => Time) extends Serializable {
+class BinStreamer[T: ClassTag](
+    @transient ds: DStream[T], getStartTime: (T) => Time, getEndTime: (T) => Time
+  ) extends Serializable {
 
   def prorate(binStart: Time, binEnd: Time)(x: T) = {
 
@@ -34,7 +36,8 @@ class BinStreamer[T: ClassTag](@transient ds: DStream[T], getStartTime: (T) => T
       }
       else {
 
-        // Even though binStart is not inclusive, setting s = binStart  implies limit s as x approaches binStart+
+        // Even though binStart is not inclusive, 
+        // setting s = binStart  implies limit s as x approaches binStart+
         val s = if (sx > binStart) sx else binStart
 
         val e = if (ex < binEnd) ex else binEnd

streaming/src/main/scala/org/apache/spark/streaming/dstream/BinAlignedWindowDStream.scala

@@ -36,11 +36,14 @@ class BinAlignedWindowDStream[T: ClassTag](
 
   override def slideDuration: Duration = parent.slideDuration
 
-  override def parentRememberDuration: Duration = rememberDuration + parent.slideDuration * sizeNumBatches * (delayNumBins + 1)
+  override def parentRememberDuration: Duration = 
+    rememberDuration + parent.slideDuration * sizeNumBatches * (delayNumBins + 1)
 
   override def compute(validTime: Time): Option[RDD[T]] = {
 
-    val binStart = (validTime - Duration(1)).floor(slideDuration * sizeNumBatches) - slideDuration * sizeNumBatches * delayNumBins
+    val binStart = 
+      (validTime - Duration(1)).floor(slideDuration * sizeNumBatches) - 
+      slideDuration * sizeNumBatches * delayNumBins
 
     if ((validTime - binStart).isMultipleOf(slideDuration * sizeNumBatches)) {
       val currentWindow = new Interval(binStart + slideDuration, validTime)

streaming/src/main/scala/org/apache/spark/streaming/dstream/ProratedEventDStream.scala

@@ -44,7 +44,8 @@ class ProratedEventDStream[T: ClassTag](parent: DStream[T],
     // Assumption: start(x) <= end(x) <= boundaryEnd
     //
 
-    def binStart = (validTime - Duration(1)).floor(slideDuration * sizeNumBatches) - slideDuration * sizeNumBatches * delayNumBins
+    def binStart = (validTime - Duration(1)).floor(slideDuration * sizeNumBatches) - 
+      slideDuration * sizeNumBatches * delayNumBins
     def binEnd = binStart + slideDuration * sizeNumBatches
 
     parent.getOrCompute(validTime).map(

streaming/src/main/scala/org/apache/spark/streaming/dstream/PulsatingWindowDStream.scala

@@ -35,11 +35,13 @@ class PulsatingWindowDStream[T: ClassTag](parent: DStream[T],
 
   override def slideDuration: Duration = parent.slideDuration
 
-  override def parentRememberDuration: Duration = rememberDuration + parent.slideDuration * sizeNumBatches * (delayNumBins + 1)
+  override def parentRememberDuration: Duration = 
+    rememberDuration + parent.slideDuration * sizeNumBatches * (delayNumBins + 1)
 
   override def compute(validTime: Time): Option[RDD[T]] = {
 
-    val binStart = (validTime - Duration(1)).floor(slideDuration * sizeNumBatches) - slideDuration * sizeNumBatches * delayNumBins
+    val binStart = (validTime - Duration(1)).floor(slideDuration * sizeNumBatches) - 
+      slideDuration * sizeNumBatches * delayNumBins
 
     val currentWindow = new Interval(binStart + slideDuration, validTime)