Try and fix a lot of style issues, we still have a ways to go but it should be slightly nicer to read

Author: holdenk

scalastyle-config.xml

@@ -67,8 +67,8 @@
  </check>
  <check level="warning" class="org.scalastyle.scalariform.NoWhitespaceBeforeLeftBracketChecker" enabled="true"></check>
  <check level="warning" class="org.scalastyle.scalariform.NoWhitespaceAfterLeftBracketChecker" enabled="true"></check>
- <check level="warning" class="org.scalastyle.scalariform.ReturnChecker" enabled="true"></check>
- <check level="warning" class="org.scalastyle.scalariform.NullChecker" enabled="true"></check>
+ <check level="warning" class="org.scalastyle.scalariform.ReturnChecker" enabled="false"></check>
+ <check level="warning" class="org.scalastyle.scalariform.NullChecker" enabled="false"></check>
  <check level="warning" class="org.scalastyle.scalariform.NoCloneChecker" enabled="true"></check>
  <check level="warning" class="org.scalastyle.scalariform.NoFinalizeChecker" enabled="true"></check>
  <check level="warning" class="org.scalastyle.scalariform.CovariantEqualsChecker" enabled="true"></check>

src/main/1.3-only/scala/com/holdenkarau/spark/testing/LocalSparkContext.scala

@@ -23,8 +23,12 @@ import org.scalatest.BeforeAndAfterAll
 import org.scalatest.BeforeAndAfterEach
 import org.scalatest.Suite
 
-/** Manages a local `sc` {@link SparkContext} variable, correctly stopping it after each test. */
-trait LocalSparkContext extends BeforeAndAfterEach with BeforeAndAfterAll { self: Suite =>
+/**
+ * Manages a local `sc` {@link SparkContext} variable,
+ * correctly stopping it after each test.
+ */
+trait LocalSparkContext extends BeforeAndAfterEach
+    with BeforeAndAfterAll { self: Suite =>
 
   @transient var sc: SparkContext = _
 
@@ -43,10 +47,9 @@ trait LocalSparkContext extends BeforeAndAfterEach with BeforeAndAfterAll { self
 
 object LocalSparkContext {
   def stop(sc: SparkContext) {
-    if (sc != null) {
-      sc.stop()
-    }
-    // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
+    Option(sc).foreach(_.stop())
+    // To avoid Akka rebinding to the same port, since it doesn't
+    // unbind immediately on shutdown.
     System.clearProperty("spark.driver.port")
   }
 

src/main/1.3/scala/com/holdenkarau/spark/testing/DataframeGenerator.scala

@@ -16,32 +16,42 @@ object DataframeGenerator {
    * @param minPartitions minimum number of partitions, defaults to 1.
    * @return Arbitrary DataFrames generator of the required schema.
    */
-  def arbitraryDataFrame(sqlContext: SQLContext, schema: StructType, minPartitions: Int = 1): Arbitrary[DataFrame] = {
+  def arbitraryDataFrame(
+    sqlContext: SQLContext, schema: StructType, minPartitions: Int = 1):
+      Arbitrary[DataFrame] = {
     arbitraryDataFrameWithCustomFields(sqlContext, schema, minPartitions)()
   }
 
   /**
-   * Creates a DataFrame Generator for the given Schema, and the given custom generators.
-   * custom generators should be in the form of (column index, generator function).
+   * Creates a DataFrame Generator for the given Schema, and the given custom
+   * generators.
+   * Custom generators should be specified as a list of:
+   * (column index, generator function) tuples.
    *
    * Note: The given custom generators should match the required schema,
    * for ex. you can't use Int generator for StringType.
    *
-   * Note 2: The ColumnGenerator* accepted as userGenerators has changed.  ColumnGenerator is now the base class of the
-   * accepted generators, users upgrading to 0.6 need to change their calls to use Column.  Futher explanation can be
-   * found in the release notes, and in the class descriptions at the bottom of this file.
+   * Note 2: The ColumnGenerator* accepted as userGenerators has changed.
+   * ColumnGenerator is now the base class of the
+   * accepted generators, users upgrading to 0.6 need to change their calls
+   * to use Column.  Further explanation can be found in the release notes, and
+   * in the class descriptions at the bottom of this file.
    *
    * @param sqlContext     SQL Context.
    * @param schema         The required Schema.
    * @param minPartitions  minimum number of partitions, defaults to 1.
-   * @param userGenerators custom user generators in the form of (column index, generator function).
-   *                       column index starts from 0 to length - 1
+   * @param userGenerators custom user generators in the form of:
+   *                       (column index, generator function).
+   *                       where column index starts from 0 to length - 1
    * @return Arbitrary DataFrames generator of the required schema.
    */
-  def arbitraryDataFrameWithCustomFields(sqlContext: SQLContext, schema: StructType, minPartitions: Int = 1)
-                                        (userGenerators: ColumnGenerator*): Arbitrary[DataFrame] = {
+  def arbitraryDataFrameWithCustomFields(
+    sqlContext: SQLContext, schema: StructType, minPartitions: Int = 1)
+    (userGenerators: ColumnGenerator*): Arbitrary[DataFrame] = {
 
-    val arbitraryRDDs = RDDGenerator.genRDD(sqlContext.sparkContext, minPartitions)(getRowGenerator(schema, userGenerators))
+    val arbitraryRDDs = RDDGenerator.genRDD(
+      sqlContext.sparkContext, minPartitions)(
+      getRowGenerator(schema, userGenerators))
     Arbitrary {
       arbitraryRDDs.map(sqlContext.createDataFrame(_, schema))
     }
@@ -60,22 +70,31 @@ object DataframeGenerator {
   /**
    * Creates row generator for the required schema and with user's custom generators.
    *
-   * Note: Custom generators should match the required schema, for ex. you can't use Int generator for StringType.
+   * Note: Custom generators should match the required schema, for example
+   * you can't use Int generator for StringType.
    *
    * @param schema           the required Row's schema.
-   * @param customGenerators user custom generator, this is useful if the user want to
-   *                         Control specific columns generation.
+   * @param customGenerators user custom generator, this is useful if the you want
+   *                         to control specific columns generation.
    * @return Gen[Row]
    */
-  def getRowGenerator(schema: StructType, customGenerators: Seq[ColumnGenerator]): Gen[Row] = {
-    val generators: List[Gen[Any]] = createGenerators(schema.fields, customGenerators)
-    val listGen: Gen[List[Any]] = Gen.sequence[List[Any], Any](generators)
-    val generator: Gen[Row] = listGen.map(list => Row.fromSeq(list))
+  def getRowGenerator(
+    schema: StructType, customGenerators: Seq[ColumnGenerator]): Gen[Row] = {
+    val generators: List[Gen[Any]] =
+      createGenerators(schema.fields, customGenerators)
+    val listGen: Gen[List[Any]] =
+      Gen.sequence[List[Any], Any](generators)
+    val generator: Gen[Row] =
+      listGen.map(list => Row.fromSeq(list))
     generator
   }
 
-  private def createGenerators(fields: Array[StructField], userGenerators: Seq[ColumnGenerator]): List[Gen[Any]] = {
-    val generatorMap = userGenerators.map(generator => (generator.columnName -> generator)).toMap
+  private def createGenerators(
+    fields: Array[StructField],
+    userGenerators: Seq[ColumnGenerator]):
+      List[Gen[Any]] = {
+    val generatorMap = userGenerators.map(
+      generator => (generator.columnName -> generator)).toMap
     (0 until fields.length).toList.map(index => {
       if (generatorMap.contains(fields(index).name)) {
         generatorMap.get(fields(index).name).get match {
@@ -87,7 +106,8 @@ object DataframeGenerator {
     })
   }
 
-  private def getGenerator(dataType: DataType, generators: Seq[ColumnGenerator] = Seq()): Gen[Any] = {
+  private def getGenerator(
+    dataType: DataType, generators: Seq[ColumnGenerator] = Seq()): Gen[Any] = {
     dataType match {
       case ByteType => Arbitrary.arbitrary[Byte]
       case ShortType => Arbitrary.arbitrary[Short]
@@ -102,7 +122,7 @@ object DataframeGenerator {
       case DateType => Arbitrary.arbLong.arbitrary.map(new Date(_))
       case arr: ArrayType => {
         val elementGenerator = getGenerator(arr.elementType)
-        return Gen.listOf(elementGenerator)
+        Gen.listOf(elementGenerator)
       }
       case map: MapType => {
         val keyGenerator = getGenerator(map.keyType)
@@ -112,31 +132,37 @@ object DataframeGenerator {
           value <- valueGenerator
         } yield (key, value)
 
-        return Gen.mapOf(keyValueGenerator)
+        Gen.mapOf(keyValueGenerator)
       }
-      case row: StructType => return getRowGenerator(row, generators)
-      case _ => throw new UnsupportedOperationException(s"Type: $dataType not supported")
+      case row: StructType => getRowGenerator(row, generators)
+      case _ => throw new UnsupportedOperationException(
+        s"Type: $dataType not supported")
     }
   }
 
 }
 
 /**
- * Previously ColumnGenerator. Allows the user to specify a generator for a specific column
+ * Previously ColumnGenerator. Allows the user to specify a generator for a
+ * specific column.
  */
-class Column(val columnName: String, generator: => Gen[Any]) extends ColumnGenerator {
+class Column(val columnName: String, generator: => Gen[Any])
+    extends ColumnGenerator {
   lazy val gen = generator
 }
 
 /**
- * ColumnList allows users to specify custom generators for a list of columns inside a StructType column
+ * ColumnList allows users to specify custom generators for a list of
+ * columns inside a StructType column.
  */
-class ColumnList(val columnName: String, generators: => Seq[ColumnGenerator]) extends ColumnGenerator {
+class ColumnList(val columnName: String, generators: => Seq[ColumnGenerator])
+    extends ColumnGenerator {
   lazy val gen = generators
 }
 
 /**
- * ColumnGenerator - prevously Column; it is now the base class for all ColumnGenerators
+ * ColumnGenerator - prevously Column; it is now the base class for all
+ * ColumnGenerators.
  */
 abstract class ColumnGenerator extends java.io.Serializable {
   val columnName: String

src/main/1.3/scala/com/holdenkarau/spark/testing/JavaDataFrameSuiteBase.scala

@@ -1,6 +1,7 @@
 package com.holdenkarau.spark.testing
 
-class JavaDataFrameSuiteBase extends SharedJavaSparkContext with DataFrameSuiteBaseLike with JavaTestSuite {
+class JavaDataFrameSuiteBase extends
+    SharedJavaSparkContext with DataFrameSuiteBaseLike with JavaTestSuite {
 
   override def beforeAllTestCasesHook() {
     sqlBeforeAllTestCases()

src/main/1.3/scala/com/holdenkarau/spark/testing/JavaRDDComparisons.scala

@@ -16,7 +16,8 @@ object JavaRDDComparisons extends RDDComparisonsLike with JavaTestSuite {
    * Compare two RDDs. If they are equal returns None, otherwise
    * returns Some with the first mismatch. Assumes we have the same partitioner.
    */
-  def compareRDDWithOrder[T](expected: JavaRDD[T], result: JavaRDD[T]): Option[(Option[T], Option[T])] = {
+  def compareRDDWithOrder[T](expected: JavaRDD[T], result: JavaRDD[T]):
+      Option[(Option[T], Option[T])] = {
     implicit val ctag = Utils.fakeClassTag[T]
     compareRDDWithOrder(expected.rdd, result.rdd)
   }
@@ -33,13 +34,16 @@ object JavaRDDComparisons extends RDDComparisonsLike with JavaTestSuite {
    * Compare two RDDs where we do not require the order to be equal.
    * If they are equal returns None, otherwise returns Some with the first mismatch.
    *
-   * @return None if the two RDDs are equal, or Some That contains first mismatch information.
-   *         Mismatch information will be Tuple3 of: (key, number of times this key occur in expected RDD,
+   * @return None if the two RDDs are equal, or Some that contains the first
+   *         mismatch information. Mismatch information will be Tuple3 of:
+   *         (key, number of times this key occur in expected RDD,
    *         number of times this key occur in result RDD)
    */
-  def compareRDD[T](expected: JavaRDD[T], result: JavaRDD[T]): Option[(T, Integer, Integer)] = {
+  def compareRDD[T](expected: JavaRDD[T], result: JavaRDD[T]):
+      Option[(T, Integer, Integer)] = {
     implicit val ctag = Utils.fakeClassTag[T]
-    compareRDD(expected.rdd, result.rdd).map(x => (x._1, Integer.valueOf(x._2), Integer.valueOf(x._3)))
+    compareRDD(expected.rdd, result.rdd).
+      map(x => (x._1, Integer.valueOf(x._2), Integer.valueOf(x._3)))
   }
 
 }

src/main/1.3/scala/com/holdenkarau/spark/testing/JavaStreamingSuitebase.scala

@@ -18,7 +18,8 @@ package com.holdenkarau.spark.testing
 
 import java.util.{List => JList}
 
-import org.apache.spark.api.java.function.{Function => JFunction, Function2 => JFunction2}
+import org.apache.spark.api.java.function.{
+  Function => JFunction, Function2 => JFunction2}
 import org.apache.spark.streaming.api.java._
 import org.apache.spark.streaming.dstream.DStream
 import org.junit.Assert._
@@ -29,11 +30,13 @@ import scala.language.implicitConversions
 import scala.reflect.ClassTag
 
 /**
- * This is the base trait for Spark Streaming testsuite. This provides basic functionality
- * to run user-defined set of input on user-defined stream operations, and verify the output.
+ * This is the base trait for Spark Streaming testsuite. This provides basic
+ * functionality to run user-defined set of input on user-defined stream operations,
+ * and verify the output matches as expected.
+ *
  * This implementation is designed to work with JUnit for java users.
  *
- * Note: this always uses the manual clock
+ * Note: this always uses the manual clock to control Spark Streaming's batches.
  */
 class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
 
@@ -67,7 +70,9 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
         compareArrays[V](expectedOutput(i).toArray, output(i).toArray)
       } else {
         implicit val config = Bag.configuration.compact[V]
-        compareArrays[V](Bag(expectedOutput(i): _*).toArray, Bag(output(i): _*).toArray)
+        compareArrays[V](
+          Bag(expectedOutput(i): _*).toArray,
+          Bag(output(i): _*).toArray)
       }
     }
 
@@ -77,7 +82,8 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
   /**
    * Test unary DStream operation with a list of inputs, with number of
    * batches to run same as the number of input values.
-   * You can simulate the input batch as a List of values or as null to simulate empty batch.
+   *
+   * Each input micro-batch is a list of values or as null to simulate empty batch.
    *
    * @param input          Sequence of input collections
    * @param operation      Binary DStream operation to be applied to the 2 inputs
@@ -93,7 +99,8 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
   /**
    * Test unary DStream operation with a list of inputs, with number of
    * batches to run same as the number of input values.
-   * You can simulate the input batch as a List of values or as null to simulate empty batch.
+   *
+   * Each input micro-batch is a list of values or as null to simulate empty batch.
    *
    * @param input          Sequence of input collections
    * @param operation      Binary DStream operation to be applied to the 2 inputs
@@ -120,18 +127,23 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
       operation.call(new JavaDStream[U](input)).dstream
     }
 
-    withOutputAndStreamingContext(setupStreams[U, V](sInput, wrappedOperation)) {
+    withOutputAndStreamingContext(
+      setupStreams[U, V](sInput, wrappedOperation)) {
+
       (outputStream, ssc) =>
-        val output: Seq[Seq[V]] = runStreams[V](outputStream, ssc, numBatches, expectedOutput.size)
-        verifyOutput[V](output, sExpectedOutput, ordered)
+      val output: Seq[Seq[V]] =
+        runStreams[V](outputStream, ssc, numBatches, expectedOutput.size)
+      verifyOutput[V](output, sExpectedOutput, ordered)
     }
   }
 
 
   /**
    * Test binary DStream operation with two lists of inputs, with number of
-   * batches to run same as the number of input values. The size of the two input lists Should be the same.
-   * You can simulate the input batch as a List of values or as null to simulate empty batch.
+   * batches to run same as the number of input values. The size of the two input
+   * lists should be equal.
+   *
+   * Each input micro-batch is a list of values or as null to simulate empty batch.
    *
    * @param input1         First sequence of input collections
    * @param input2         Second sequence of input collections
@@ -148,8 +160,10 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
 
   /**
    * Test binary DStream operation with two lists of inputs, with number of
-   * batches to run same as the number of input values. The size of the two input lists Should be the same.
-   * You can simulate the input batch as a List of values or as null to simulate empty batch.
+   * batches to run same as the number of input values. The size of the two input
+   * lists should be equal.
+   *
+   * Each input micro-batch is a list of values or as null to simulate empty batch.
    *
    * @param input1         First sequence of input collections
    * @param input2         Second sequence of input collections
@@ -166,7 +180,8 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
       expectedOutput: JList[JList[W]],
       ordered: Boolean): Unit = {
 
-    assertEquals("Length of the input lists are not equal", input1.length, input2.length)
+    assertEquals("Length of the input lists are not equal",
+      input1.length, input2.length)
     val numBatches = input1.size
 
     implicit val ctagU = Utils.fakeClassTag[U]
@@ -181,10 +196,13 @@ class JavaStreamingSuiteBase extends JavaSuiteBase with StreamingSuiteCommon {
       operation.call(new JavaDStream[U](input1), new JavaDStream[V](input2)).dstream
     }
 
-    withOutputAndStreamingContext(setupStreams[U, V, W](sInput1, sInput2, wrappedOperation)) {
+    withOutputAndStreamingContext(
+      setupStreams[U, V, W](sInput1, sInput2, wrappedOperation)) {
+
       (outputStream, ssc) =>
-        val output = runStreams[W](outputStream, ssc, numBatches, expectedOutput.size)
-        verifyOutput[W](output, sExpectedOutput, ordered)
+      val output = runStreams[W](
+        outputStream, ssc, numBatches, expectedOutput.size)
+      verifyOutput[W](output, sExpectedOutput, ordered)
     }
   }
 

src/main/1.3/scala/com/holdenkarau/spark/testing/PerTestSparkContext.scala

@@ -23,8 +23,11 @@ import org.scalatest.BeforeAndAfterAll
 import org.scalatest.BeforeAndAfterEach
 import org.scalatest.Suite
 
-/** Provides a local `sc` {@link SparkContext} variable, correctly stopping it after each test.
- * The stopping logic is provided in {@link LocalSparkContext} */
+/**
+ * Provides a local `sc`
+ * {@link SparkContext} variable, correctly stopping it after each test.
+ * The stopping logic is provided in {@link LocalSparkContext}.
+ */
 trait PerTestSparkContext extends LocalSparkContext with BeforeAndAfterEach
     with SparkContextProvider { self: Suite =>