mesos/spark — pull request #706 by ScrapCodes (2013-07-16 07:42:08+00:00)
Title: Spark Streaming stop application mechanism

This is to continue the discussion, that started at mailing list regarding idiomatic way to stop spark streaming app inside a receiver.

Following is the example that shows how can a streaming application can be stopped using stopApp function. The example will stop streaming app once an end of stream is reached on receiver. I am not sure if its a great thing to do. But can be useful incase we want to detect a situation like connect exception and we would not want to continue. Currently streaming continues even if receiver did not start at all !

```scala
package spark.examples

import spark.storage.StorageLevel
import spark.streaming.dstream.NetworkReceiver
import java.net.Socket
import java.io.{BufferedReader, InputStreamReader}
import spark.streaming.{Seconds, StreamingContext}
import spark.streaming.StreamingContext._

class SocketTextStreamReceiver(host: String,
port: Int
) extends NetworkReceiver[String] {

protected lazy val blocksGenerator: BlockGenerator =
new BlockGenerator(StorageLevel.MEMORY_ONLY_SER_2)

protected def onStart() = {
blocksGenerator.start()
val socket = new Socket(host, port)
val dataInputStream = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"))
var data: String = ""
while (data != null) {
data = dataInputStream.readLine();
blocksGenerator += data
}
println("Reached end of stream!!")
stopApp() //<-- this can be called to stop the streaming application.
}

protected def onStop() {
blocksGenerator.stop()
}

}

object DocSample {

def main(args: Array[String]) {
val ssc = new StreamingContext("local[2]", "WordCountCustomStreamSource",
Seconds(1))
val lines = ssc.networkStream[String](new SocketTextStreamReceiver(
"localhost", 8445))
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)

wordCounts.print()
ssc.start()

}
}
```