Merge pull request apache#482 from tdas/streaming-example-fix

pwendell · pwendell · commit 0367981d4776 · 2014-01-20T22:25:50.000-08:00
Added StreamingContext.awaitTermination to streaming examples

StreamingContext.start() currently starts a non-daemon thread which prevents termination of a Spark Streaming program even if main function has exited. Since the expected behavior of a streaming program is to run until explicitly killed, this was sort of fine when spark streaming applications are launched from the command line. However, when launched in Yarn-standalone mode, this did not work as the driver effectively got terminated when the main function exits. So SparkStreaming examples did not work on Yarn.

This addition to the examples ensures that the examples work on Yarn and also ensures that everyone learns that StreamingContext.awaitTermination() being necessary for SparkStreaming programs to wait.

The true bug-fix of making sure all threads by Spark Streaming are daemon threads is left for post-0.9.
diff --git a/examples/src/main/java/org/apache/spark/streaming/examples/JavaFlumeEventCount.java b/examples/src/main/java/org/apache/spark/streaming/examples/JavaFlumeEventCount.java
@@ -70,5 +70,6 @@ public String call(Long in) {
     }).print();
 
     ssc.start();
+    ssc.awaitTermination();
   }
 }
diff --git a/examples/src/main/java/org/apache/spark/streaming/examples/JavaKafkaWordCount.java b/examples/src/main/java/org/apache/spark/streaming/examples/JavaKafkaWordCount.java
@@ -104,5 +104,6 @@ public Integer call(Integer i1, Integer i2) {
 
     wordCounts.print();
     jssc.start();
+    jssc.awaitTermination();
   }
 }
diff --git a/examples/src/main/java/org/apache/spark/streaming/examples/JavaNetworkWordCount.java b/examples/src/main/java/org/apache/spark/streaming/examples/JavaNetworkWordCount.java
@@ -84,5 +84,6 @@ public Integer call(Integer i1, Integer i2) {
 
     wordCounts.print();
     ssc.start();
+    ssc.awaitTermination();
   }
 }
diff --git a/examples/src/main/java/org/apache/spark/streaming/examples/JavaQueueStream.java b/examples/src/main/java/org/apache/spark/streaming/examples/JavaQueueStream.java
@@ -80,5 +80,6 @@ public Integer call(Integer i1, Integer i2) {
 
     reducedStream.print();
     ssc.start();
+    ssc.awaitTermination();
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/ActorWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/ActorWordCount.scala
@@ -171,5 +171,6 @@ object ActorWordCount {
     lines.flatMap(_.split("\\s+")).map(x => (x, 1)).reduceByKey(_ + _).print()
 
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/FlumeEventCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/FlumeEventCount.scala
@@ -60,5 +60,6 @@ object FlumeEventCount {
     stream.count().map(cnt => "Received " + cnt + " flume events." ).print()
 
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/HdfsWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/HdfsWordCount.scala
@@ -50,6 +50,7 @@ object HdfsWordCount {
     val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
     wordCounts.print()
     ssc.start()
+    ssc.awaitTermination()
   }
 }
 
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/KafkaWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/KafkaWordCount.scala
@@ -61,6 +61,7 @@ object KafkaWordCount {
     wordCounts.print()
     
     ssc.start()
+    ssc.awaitTermination()
   }
 }
 
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala
@@ -101,5 +101,6 @@ object MQTTWordCount {
     val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
     wordCounts.print()
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/NetworkWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/NetworkWordCount.scala
@@ -54,5 +54,6 @@ object NetworkWordCount {
     val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
     wordCounts.print()
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/RawNetworkGrep.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/RawNetworkGrep.scala
@@ -61,5 +61,6 @@ object RawNetworkGrep {
     union.filter(_.contains("the")).count().foreachRDD(r =>
       println("Grep count: " + r.collect().mkString))
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/RecoverableNetworkWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/RecoverableNetworkWordCount.scala
@@ -114,5 +114,6 @@ object RecoverableNetworkWordCount {
         createContext(master, ip, port, outputPath)
       })
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/StatefulNetworkWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/StatefulNetworkWordCount.scala
@@ -65,5 +65,6 @@ object StatefulNetworkWordCount {
     val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)
     stateDstream.print()
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/TwitterAlgebirdCMS.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/TwitterAlgebirdCMS.scala
@@ -110,5 +110,6 @@ object TwitterAlgebirdCMS {
     })
 
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/TwitterAlgebirdHLL.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/TwitterAlgebirdHLL.scala
@@ -87,5 +87,6 @@ object TwitterAlgebirdHLL {
     })
 
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/TwitterPopularTags.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/TwitterPopularTags.scala
@@ -69,5 +69,6 @@ object TwitterPopularTags {
     })
 
     ssc.start()
+    ssc.awaitTermination()
   }
 }
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/ZeroMQWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/ZeroMQWordCount.scala
@@ -91,5 +91,6 @@ object ZeroMQWordCount {
     val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
     wordCounts.print()
     ssc.start()
+    ssc.awaitTermination()
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -70,5 +70,6 @@ public String call(Long in) {`
`70`	`70`	`}).print();`
`71`	`71`
`72`	`72`	`ssc.start();`
	`73`	`+ ssc.awaitTermination();`
`73`	`74`	`}`
`74`	`75`	`}`
Original file line number	Diff line number	Diff line change
`@@ -104,5 +104,6 @@ public Integer call(Integer i1, Integer i2) {`
`104`	`104`
`105`	`105`	`wordCounts.print();`
`106`	`106`	`jssc.start();`
	`107`	`+ jssc.awaitTermination();`
`107`	`108`	`}`
`108`	`109`	`}`
Original file line number	Diff line number	Diff line change
`@@ -84,5 +84,6 @@ public Integer call(Integer i1, Integer i2) {`
`84`	`84`
`85`	`85`	`wordCounts.print();`
`86`	`86`	`ssc.start();`
	`87`	`+ ssc.awaitTermination();`
`87`	`88`	`}`
`88`	`89`	`}`
Original file line number	Diff line number	Diff line change
`@@ -80,5 +80,6 @@ public Integer call(Integer i1, Integer i2) {`
`80`	`80`
`81`	`81`	`reducedStream.print();`
`82`	`82`	`ssc.start();`
	`83`	`+ ssc.awaitTermination();`
`83`	`84`	`}`
`84`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -171,5 +171,6 @@ object ActorWordCount {`
`171`	`171`	`lines.flatMap(_.split("\\s+")).map(x => (x, 1)).reduceByKey(_ + _).print()`
`172`	`172`
`173`	`173`	`ssc.start()`
	`174`	`+ ssc.awaitTermination()`
`174`	`175`	`}`
`175`	`176`	`}`
Original file line number	Diff line number	Diff line change
`@@ -60,5 +60,6 @@ object FlumeEventCount {`
`60`	`60`	`stream.count().map(cnt => "Received " + cnt + " flume events." ).print()`
`61`	`61`
`62`	`62`	`ssc.start()`
	`63`	`+ ssc.awaitTermination()`
`63`	`64`	`}`
`64`	`65`	`}`
Original file line number	Diff line number	Diff line change
`@@ -50,6 +50,7 @@ object HdfsWordCount {`
`50`	`50`	`val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)`
`51`	`51`	`wordCounts.print()`
`52`	`52`	`ssc.start()`
	`53`	`+ ssc.awaitTermination()`
`53`	`54`	`}`
`54`	`55`	`}`
`55`	`56`
Original file line number	Diff line number	Diff line change
`@@ -61,6 +61,7 @@ object KafkaWordCount {`
`61`	`61`	`wordCounts.print()`
`62`	`62`
`63`	`63`	`ssc.start()`
	`64`	`+ ssc.awaitTermination()`
`64`	`65`	`}`
`65`	`66`	`}`
`66`	`67`
Original file line number	Diff line number	Diff line change
`@@ -101,5 +101,6 @@ object MQTTWordCount {`
`101`	`101`	`val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)`
`102`	`102`	`wordCounts.print()`
`103`	`103`	`ssc.start()`
	`104`	`+ ssc.awaitTermination()`
`104`	`105`	`}`
`105`	`106`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,5 +54,6 @@ object NetworkWordCount {`
`54`	`54`	`val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)`
`55`	`55`	`wordCounts.print()`
`56`	`56`	`ssc.start()`
	`57`	`+ ssc.awaitTermination()`
`57`	`58`	`}`
`58`	`59`	`}`