[SPARK-1853] Show Streaming application code context (file, line number) in Spark Stages UI

Please review: [SPARK-1853] Show Streaming application code context (file, line number) in Spark Stages UI

_Screenshot after fix:_

![screen shot 2014-07-18 at 7 31 54 am](https://cloud.githubusercontent.com/assets/668134/3785835/e70e3416-19ca-11e4-9d0e-44f3311d65df.png)

_Sample code used: Tutorial.scala_
```scala
1   package org.apache.spark.examples.streaming
2
3   import org.apache.spark._
4   import org.apache.spark.SparkContext._
5   import org.apache.spark.streaming._
6   import org.apache.spark.streaming.twitter._
7   import org.apache.spark.streaming.StreamingContext._
8   import org.apache.spark.examples.streaming.TutorialHelper._
9   import org.apache.spark.util.{CallSite, Utils}
10
11 object Tutorial {
12  def main(args: Array[String]) {
13    
14    // Checkpoint directory
15    val checkpointDir = TutorialHelper.getCheckpointDirectory()
16
17   // Configure Twitter credentials
18    val apiKey = ""
19   val apiSecret = ""
20    val accessToken = ""
21    val accessTokenSecret = ""
22    TutorialHelper.configureTwitterCredentials(apiKey, apiSecret, accessToken, accessTokenSecret)
23
24    // Your code goes here
25    val sc: SparkContext = new SparkContext(new SparkConf().setAppName("Twitter Example").setMaster("spark://ec2-54-241-226-42.us-west-1.compute.amazonaws.com:7077"))
26    val ssc = new StreamingContext(sc, Seconds(5))
27    val tweets = TwitterUtils.createStream(ssc, None)
28
29    val statuses = tweets.map(status => status.getText())
30    val words = statuses.flatMap(status => status.split(" "))
31
32    val hashtags = words.filter(word => word.startsWith("#"))
33    val counts = hashtags.map(tag => (tag, 1)).reduceByKeyAndWindow(_ + _, _ - _, Seconds(60 * 5), Seconds(10))    
34
35    val sortedCounts = counts.map { case(tag, count) => (count, tag) }
36                         .transform(rdd => rdd.sortByKey(false))
37    sortedCounts.foreach(rdd =>
38      println("\nTop 10 hashtags:\n" + rdd.take(10).mkString("\n")))
39
40    ssc.checkpoint(checkpointDir)
41    ssc.start()
42    ssc.awaitTermination()
43
44   }
45 }

```
Dealing with this separately in two code paths is prone to errors in the future. Can you make the `compute(time) match { ... }` return into a variable and use that later. Something like this.

```
val rddOption = compute(time) match {
    // no setCallSite inside this
}
setCallSite(prevCallSite)   // only one instance of setCallSite, common to both code paths.
return rddOption
```

And while on that, can you also add `return` at other code paths in this function. That is, the `None` in lines 332 and 335 in this commit?