SPARK-2626 [DOCS] Stop SparkContext in all examples

Call SparkContext.stop() in all examples (and touch up minor nearby code style issues while at it)

Author: Sean Owen <[email protected]>

Closes #2575 from srowen/SPARK-2626 and squashes the following commits:

5b2baae [Sean Owen] Call SparkContext.stop() in all examples (and touch up minor nearby code style issues while at it)

Author: srowen

examples/src/main/java/org/apache/spark/examples/JavaSparkPi.java

@@ -31,7 +31,6 @@
  * Usage: JavaSparkPi [slices]
  */
 public final class JavaSparkPi {
-  
 
   public static void main(String[] args) throws Exception {
     SparkConf sparkConf = new SparkConf().setAppName("JavaSparkPi");
@@ -61,5 +60,7 @@ public Integer call(Integer integer, Integer integer2) {
     });
 
     System.out.println("Pi is roughly " + 4.0 * count / n);
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/sql/JavaSparkSQL.java

@@ -61,7 +61,8 @@ public static void main(String[] args) throws Exception {
     // Load a text file and convert each line to a Java Bean.
     JavaRDD<Person> people = ctx.textFile("examples/src/main/resources/people.txt").map(
       new Function<String, Person>() {
-        public Person call(String line) throws Exception {
+        @Override
+        public Person call(String line) {
           String[] parts = line.split(",");
 
           Person person = new Person();
@@ -82,6 +83,7 @@ public Person call(String line) throws Exception {
     // The results of SQL queries are SchemaRDDs and support all the normal RDD operations.
     // The columns of a row in the result can be accessed by ordinal.
     List<String> teenagerNames = teenagers.map(new Function<Row, String>() {
+      @Override
       public String call(Row row) {
         return "Name: " + row.getString(0);
       }
@@ -104,6 +106,7 @@ public String call(Row row) {
     JavaSchemaRDD teenagers2 =
       sqlCtx.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19");
     teenagerNames = teenagers2.map(new Function<Row, String>() {
+      @Override
       public String call(Row row) {
           return "Name: " + row.getString(0);
       }
@@ -136,6 +139,7 @@ public String call(Row row) {
     // The results of SQL queries are JavaSchemaRDDs and support all the normal RDD operations.
     // The columns of a row in the result can be accessed by ordinal.
     teenagerNames = teenagers3.map(new Function<Row, String>() {
+      @Override
       public String call(Row row) { return "Name: " + row.getString(0); }
     }).collect();
     for (String name: teenagerNames) {
@@ -162,12 +166,15 @@ public String call(Row row) {
 
     JavaSchemaRDD peopleWithCity = sqlCtx.sql("SELECT name, address.city FROM people2");
     List<String> nameAndCity = peopleWithCity.map(new Function<Row, String>() {
+      @Override
       public String call(Row row) {
         return "Name: " + row.getString(0) + ", City: " + row.getString(1);
       }
     }).collect();
     for (String name: nameAndCity) {
       System.out.println(name);
     }
+
+    ctx.stop();
   }
 }

examples/src/main/python/avro_inputformat.py

@@ -78,3 +78,5 @@
     output = avro_rdd.map(lambda x: x[0]).collect()
     for k in output:
         print k
+
+    sc.stop()

examples/src/main/python/parquet_inputformat.py

@@ -57,3 +57,5 @@
     output = parquet_rdd.map(lambda x: x[1]).collect()
     for k in output:
         print k
+
+    sc.stop()

examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala

@@ -136,5 +136,7 @@ object CassandraCQLTest {
         classOf[CqlOutputFormat],
         job.getConfiguration()
       )
+
+    sc.stop()
   }
 }

examples/src/main/scala/org/apache/spark/examples/CassandraTest.scala

@@ -126,6 +126,8 @@ object CassandraTest {
       }
     }.saveAsNewAPIHadoopFile("casDemo", classOf[ByteBuffer], classOf[List[Mutation]],
       classOf[ColumnFamilyOutputFormat], job.getConfiguration)
+
+    sc.stop()
   }
 }
 

examples/src/main/scala/org/apache/spark/examples/GroupByTest.scala

@@ -44,11 +44,11 @@ object GroupByTest {
         arr1(i) = (ranGen.nextInt(Int.MaxValue), byteArr)
       }
       arr1
-    }.cache
+    }.cache()
     // Enforce that everything has been calculated and in cache
-    pairs1.count
+    pairs1.count()
 
-    println(pairs1.groupByKey(numReducers).count)
+    println(pairs1.groupByKey(numReducers).count())
 
     sc.stop()
   }

examples/src/main/scala/org/apache/spark/examples/LogQuery.scala

@@ -79,5 +79,7 @@ object LogQuery {
       .reduceByKey((a, b) => a.merge(b))
       .collect().foreach{
         case (user, query) => println("%s\t%s".format(user, query))}
+
+    sc.stop()
   }
 }

examples/src/main/scala/org/apache/spark/examples/bagel/WikipediaPageRank.scala

@@ -21,7 +21,6 @@ import org.apache.spark._
 import org.apache.spark.SparkContext._
 
 import org.apache.spark.bagel._
-import org.apache.spark.bagel.Bagel._
 
 import scala.xml.{XML,NodeSeq}
 
@@ -78,9 +77,9 @@ object WikipediaPageRank {
       (id, new PRVertex(1.0 / numVertices, outEdges))
     })
     if (usePartitioner) {
-      vertices = vertices.partitionBy(new HashPartitioner(sc.defaultParallelism)).cache
+      vertices = vertices.partitionBy(new HashPartitioner(sc.defaultParallelism)).cache()
     } else {
-      vertices = vertices.cache
+      vertices = vertices.cache()
     }
     println("Done parsing input file.")
 
@@ -100,7 +99,9 @@ object WikipediaPageRank {
       (result
        .filter { case (id, vertex) => vertex.value >= threshold }
        .map { case (id, vertex) => "%s\t%s\n".format(id, vertex.value) }
-       .collect.mkString)
+       .collect().mkString)
     println(top)
+
+    sc.stop()
   }
 }

examples/src/main/scala/org/apache/spark/examples/sql/RDDRelation.scala

@@ -51,7 +51,7 @@ object RDDRelation {
     val rddFromSql = sql("SELECT key, value FROM records WHERE key < 10")
 
     println("Result of RDD.map:")
-    rddFromSql.map(row => s"Key: ${row(0)}, Value: ${row(1)}").collect.foreach(println)
+    rddFromSql.map(row => s"Key: ${row(0)}, Value: ${row(1)}").collect().foreach(println)
 
     // Queries can also be written using a LINQ-like Scala DSL.
     rdd.where('key === 1).orderBy('value.asc).select('key).collect().foreach(println)
@@ -68,5 +68,7 @@ object RDDRelation {
     // These files can also be registered as tables.
     parquetFile.registerTempTable("parquetFile")
     sql("SELECT * FROM parquetFile").collect().foreach(println)
+
+    sc.stop()
   }
 }