[MINOR] Remove inappropriate type notation and extra anonymous closure within functional transformations

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

@@ -149,8 +149,8 @@ case class ExceptionFailure(
     this(e, accumUpdates, preserveCause = true)
   }
 
-  def exception: Option[Throwable] = exceptionWrapper.flatMap {
-    (w: ThrowableSerializationWrapper) => Option(w.exception)
+  def exception: Option[Throwable] = exceptionWrapper.flatMap { w =>
+    Option(w.exception)
   }
 
   override def toErrorString: String =

core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala

@@ -131,7 +131,7 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
    */
   def flatMapToDouble(f: DoubleFlatMapFunction[T]): JavaDoubleRDD = {
     def fn: (T) => Iterator[jl.Double] = (x: T) => f.call(x).asScala
-    new JavaDoubleRDD(rdd.flatMap(fn).map((x: jl.Double) => x.doubleValue()))
+    new JavaDoubleRDD(rdd.flatMap(fn).map(x => x.doubleValue()))
   }
 
   /**
@@ -173,7 +173,7 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
     def fn: (Iterator[T]) => Iterator[jl.Double] = {
       (x: Iterator[T]) => f.call(x.asJava).asScala
     }
-    new JavaDoubleRDD(rdd.mapPartitions(fn).map((x: jl.Double) => x.doubleValue()))
+    new JavaDoubleRDD(rdd.mapPartitions(fn).map(x => x.doubleValue()))
   }
 
   /**

core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala

@@ -35,9 +35,8 @@ private[ui] class ApplicationPage(parent: MasterWebUI) extends WebUIPage("app")
   def render(request: HttpServletRequest): Seq[Node] = {
     val appId = request.getParameter("appId")
     val state = master.askWithRetry[MasterStateResponse](RequestMasterState)
-    val app = state.activeApps.find(_.id == appId).getOrElse({
-      state.completedApps.find(_.id == appId).getOrElse(null)
-    })
+    val app = state.activeApps.find(_.id == appId)
+      .getOrElse(state.completedApps.find(_.id == appId).orNull)
     if (app == null) {
       val msg = <div class="row-fluid">No running application with ID {appId}</div>
       return UIUtils.basicSparkPage(msg, "Not Found")

core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala

@@ -68,7 +68,10 @@ private[deploy] class DriverRunner(
 
   private var clock: Clock = new SystemClock()
   private var sleeper = new Sleeper {
-    def sleep(seconds: Int): Unit = (0 until seconds).takeWhile(f => {Thread.sleep(1000); !killed})
+    def sleep(seconds: Int): Unit = (0 until seconds).takeWhile { _ =>
+      Thread.sleep(1000)
+      !killed
+    }
   }
 
   /** Starts a thread to run and manage the driver. */

core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala

@@ -191,9 +191,9 @@ private class PartitionCoalescer(maxPartitions: Int, prev: RDD[_], balanceSlack:
     // initializes/resets to start iterating from the beginning
     def resetIterator(): Iterator[(String, Partition)] = {
       val iterators = (0 to 2).map( x =>
-        prev.partitions.iterator.flatMap(p => {
+        prev.partitions.iterator.flatMap { p =>
           if (currPrefLocs(p).size > x) Some((currPrefLocs(p)(x), p)) else None
-        } )
+        }
       )
       iterators.reduceLeft((x, y) => x ++ y)
     }

core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala

@@ -65,11 +65,11 @@ class JdbcRDD[T: ClassTag](
   override def getPartitions: Array[Partition] = {
     // bounds are inclusive, hence the + 1 here and - 1 on end
     val length = BigInt(1) + upperBound - lowerBound
-    (0 until numPartitions).map(i => {
+    (0 until numPartitions).map { i =>
       val start = lowerBound + ((i * length) / numPartitions)
       val end = lowerBound + (((i + 1) * length) / numPartitions) - 1
       new JdbcPartition(i, start.toLong, end.toLong)
-    }).toArray
+    }.toArray
   }
 
   override def compute(thePart: Partition, context: TaskContext): Iterator[T] = new NextIterator[T]

core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala

@@ -150,10 +150,10 @@ private object ParallelCollectionRDD {
         slices
       case _ =>
         val array = seq.toArray // To prevent O(n^2) operations for List etc
-        positions(array.length, numSlices).map({
+        positions(array.length, numSlices).map {
           case (start, end) =>
             array.slice(start, end).toSeq
-        }).toSeq
+        }.toSeq
     }
   }
 }

examples/src/main/scala/org/apache/spark/examples/mllib/StreamingTestExample.scala

@@ -59,10 +59,10 @@ object StreamingTestExample {
 
     val conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample")
     val ssc = new StreamingContext(conf, batchDuration)
-    ssc.checkpoint({
+    ssc.checkpoint {
       val dir = Utils.createTempDir()
       dir.toString
-    })
+    }
 
     // $example on$
     val data = ssc.textFileStream(dataDir).map(line => line.split(",") match {

examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala

@@ -116,7 +116,7 @@ object RecoverableNetworkWordCount {
     val lines = ssc.socketTextStream(ip, port)
     val words = lines.flatMap(_.split(" "))
     val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
-    wordCounts.foreachRDD { (rdd: RDD[(String, Int)], time: Time) =>
+    wordCounts.foreachRDD { (rdd, time) =>
       // Get or register the blacklist Broadcast
       val blacklist = WordBlacklist.getInstance(rdd.sparkContext)
       // Get or register the droppedWordsCounter Accumulator

examples/src/main/scala/org/apache/spark/examples/streaming/SqlNetworkWordCount.scala

@@ -59,7 +59,7 @@ object SqlNetworkWordCount {
     val words = lines.flatMap(_.split(" "))
 
     // Convert RDDs of the words DStream to DataFrame and run SQL query
-    words.foreachRDD { (rdd: RDD[String], time: Time) =>
+    words.foreachRDD { (rdd, time) =>
       // Get the singleton instance of SQLContext
       val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
       import sqlContext.implicits._

mllib/src/main/scala/org/apache/spark/mllib/api/python/Word2VecModelWrapper.scala

@@ -56,7 +56,9 @@ private[python] class Word2VecModelWrapper(model: Word2VecModel) {
   }
 
   def getVectors: JMap[String, JList[Float]] = {
-    model.getVectors.map({case (k, v) => (k, v.toList.asJava)}).asJava
+    model.getVectors.map { case (k, v) =>
+      (k, v.toList.asJava)
+    }.asJava
   }
 
   def save(sc: SparkContext, path: String): Unit = model.save(sc, path)

mllib/src/main/scala/org/apache/spark/mllib/evaluation/BinaryClassificationMetrics.scala

@@ -189,8 +189,9 @@ class BinaryClassificationMetrics @Since("1.3.0") (
       Iterator(agg)
     }.collect()
     val partitionwiseCumulativeCounts =
-      agg.scanLeft(new BinaryLabelCounter())(
-        (agg: BinaryLabelCounter, c: BinaryLabelCounter) => agg.clone() += c)
+      agg.scanLeft(new BinaryLabelCounter()) { (agg, c) =>
+        agg.clone() += c
+      }
     val totalCount = partitionwiseCumulativeCounts.last
     logInfo(s"Total counts: $totalCount")
     val cumulativeCounts = binnedCounts.mapPartitionsWithIndex(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -212,7 +212,7 @@ class Analyzer(
      *  represented as the bit masks.
      */
     def bitmasks(r: Rollup): Seq[Int] = {
-      Seq.tabulate(r.groupByExprs.length + 1)(idx => {(1 << idx) - 1})
+      Seq.tabulate(r.groupByExprs.length + 1)(idx => (1 << idx) - 1)
     }
 
     /*

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala

@@ -168,9 +168,9 @@ object FromUnsafeProjection {
    * Returns an UnsafeProjection for given Array of DataTypes.
    */
   def apply(fields: Seq[DataType]): Projection = {
-    create(fields.zipWithIndex.map(x => {
+    create(fields.zipWithIndex.map { x =>
       new BoundReference(x._2, x._1, true)
-    }))
+    })
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -300,11 +300,9 @@ object SetOperationPushDown extends Rule[LogicalPlan] with PredicateHelper {
       assert(children.nonEmpty)
       val (deterministic, nondeterministic) = partitionByDeterministic(condition)
       val newFirstChild = Filter(deterministic, children.head)
-      val newOtherChildren = children.tail.map {
-        child => {
-          val rewrites = buildRewrites(children.head, child)
-          Filter(pushToRight(deterministic, rewrites), child)
-        }
+      val newOtherChildren = children.tail.map { child =>
+        val rewrites = buildRewrites(children.head, child)
+        Filter(pushToRight(deterministic, rewrites), child)
       }
       Filter(nondeterministic, Union(newFirstChild +: newOtherChildren))
 

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoin.scala

@@ -60,9 +60,9 @@ case class ShuffledHashJoin(
     val context = TaskContext.get()
     val relation = HashedRelation(iter, buildKeys, taskMemoryManager = context.taskMemoryManager())
     // This relation is usually used until the end of task.
-    context.addTaskCompletionListener((t: TaskContext) =>
+    context.addTaskCompletionListener { _ =>
       relation.close()
-    )
+    }
     relation
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/stat/StatFunctions.scala

@@ -431,7 +431,7 @@ private[sql] object StatFunctions extends Logging {
       s"exceed 1e4. Currently $columnSize")
     val table = counts.groupBy(_.get(0)).map { case (col1Item, rows) =>
       val countsRow = new GenericMutableRow(columnSize + 1)
-      rows.foreach { (row: Row) =>
+      rows.foreach { row =>
         // row.get(0) is column 1
         // row.get(1) is column 2
         // row.get(2) is the frequency

sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala

@@ -152,10 +152,8 @@ private[hive] case class HiveGenericUDF(
     var i = 0
     while (i < children.length) {
       val idx = i
-      deferredObjects(i).asInstanceOf[DeferredObjectAdapter].set(
-        () => {
-          children(idx).eval(input)
-        })
+      deferredObjects(i).asInstanceOf[DeferredObjectAdapter]
+        .set(() => children(idx).eval(input))
       i += 1
     }
     unwrap(function.evaluate(deferredObjects), returnInspector)

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

@@ -895,9 +895,9 @@ abstract class DStream[T: ClassTag] (
     logInfo(s"Slicing from $fromTime to $toTime" +
       s" (aligned to $alignedFromTime and $alignedToTime)")
 
-    alignedFromTime.to(alignedToTime, slideDuration).flatMap(time => {
+    alignedFromTime.to(alignedToTime, slideDuration).flatMap { time =>
       if (time >= zeroTime) getOrCompute(time) else None
-    })
+    }
   }
 
   /**