GitHub PullRequestReviewCommentEvent at 2014-01-27 21:00:44
Repository: incubator-spark
Actor: tdas
Payload: {"comment":{"url":"https://api.github.com/repos/apache/incubator-spark/pulls/comments/9205587","id":9205587,"diff_hunk":"@@ -7,74 +7,460 @@ title: Spark Streaming Programming Guide\n {:toc}\n \n # Overview\n-A Spark Streaming application is very similar to a Spark application; it consists of a *driver program* that runs the user's `main` function and continuous executes various *parallel operations* on input streams of data. The main abstraction Spark Streaming provides is a *discretized stream* (DStream), which is a continuous sequence of RDDs (distributed collections of elements) representing a continuous stream of data. DStreams can be created from live incoming data (such as data from a socket, Kafka, etc.) or can be generated by transforming existing DStreams using parallel operators like `map`, `reduce`, and `window`. The basic processing model is as follows: \n-(i) While a Spark Streaming driver program is running, the system receives data from various sources and and divides it into batches. Each batch of data is treated as an RDD, that is, an immutable parallel collection of data. These input RDDs are saved in memory and replicated to two nodes for fault-tolerance. This sequence of RDDs is collectively called an InputDStream.\n-(ii) Data received by InputDStreams are processed using DStream operations. Since all data is represented as RDDs and all DStream operations as RDD operations, data is automatically recovered in the event of node failures.  \n+Spark Streaming is an extension of the core Spark API that allows enables high-throughput,\n+fault-tolerant stream processing of live data streams. Data can be ingested from many sources\n+like Kafka, Flume, Twitter, ZeroMQ or plain old TCP sockets and be processed using complex\n+algorithms expressed with high-level functions like `map`, `reduce`, `join` and `window`.\n+Finally, processed data can be pushed out to files, SQL or NoSQL databases,\n+or directly update live dashboards. In fact, you can apply Spark's in-built\n+[machine learning](mllib-guide.html) algorithms, and\n+[graph processing](graphx-programming-guide.html) algorithms on data streams.\n+\n+<p style=\"text-align: center;\">\n+  <img\n+    src=\"img/streaming-arch.png\"\n+    title=\"Spark Streaming architecture\"\n+    alt=\"Spark Streaming\"\n+    width=\"70%\"\n+  />\n+</p>\n+\n+Internally, it works as follows. Spark Streaming receives live input data streams and divides\n+the data into batches, which are then processed by the Spark engine to generate the final\n+stream of result in batches.\n+\n+<p style=\"text-align: center;\">\n+  <img src=\"img/streaming-flow.png\"\n+       title=\"Spark Streaming data flow\"\n+       alt=\"Spark Streaming\"\n+       width=\"70%\" />\n+</p>\n+\n+Spark Streaming provides a high-level abstraction called *discretized stream* or *DStream*,\n+which represents a continuous stream of data. DStreams can be created either from input data\n+stream from sources such as Kafka and Flume, or by applying high-level\n+operations on other DStreams. Internally, a DStream is represented as a sequence of\n+[RDDs](api/core/index.html#org.apache.spark.rdd.RDD).\n+\n+This guide shows some how to start writing Spark Streaming programs with DStreams. You can\n+write Spark Streaming programs in Scala or Java and this guide presents examples in both\n+languages. You will find tabs throughout this guide that let's you choose between Scala and Java\n+code snippets.\n+\n+***************************************************************************************************  \n+\n+# A Quick Example\n+Before we go into the details of how to write your own Spark Streaming program,\n+let's take a quick look at what a simple Spark Streaming program looks like. Let's say we want to\n+count the number of words in text data received over a simple socket connection. All you need to\n+do is as follows. You can click on the following tabs to choose between the Scala or the Java\n+example.\n+\n+<div class=\"codetabs\">\n+<div data-lang=\"scala\"  markdown=\"1\" >\n+\n+First, we create a\n+[StreamingContext](api/streaming/index.html#org.apache.spark.streaming.StreamingContext) object,\n+which is the main entry point for all streaming\n+functionality. Besides Spark's configuration, we specify that any DStream would be processed\n+in 1 second batches.\n \n-This guide shows some how to start programming with DStreams. \n+{% highlight scala %}\n+// Create a StreamingContext with a SparkConf configuration\n+val ssc = new StreamingContext(sparkConf, Seconds(1))\n+{% endhighlight %}\n+\n+Using this context, we then create a new DStream from\n+by specifying the IP address and port of the data server.\n+\n+{% highlight scala %}\n+// Create a DStream that will connect to serverIP:serverPort\n+val lines = ssc.socketTextStream(serverIP, serverPort)\n+{% endhighlight %}\n+\n+This `lines` DStream represents the stream of data that will be received from the data\n+server. Each records in this DStream is a line of text. Next, we want to split the lines by\n+space into words.\n+\n+{% highlight scala %}\n+// Split each line into words\n+val words = lines.flatMap(_.split(\" \"))\n+{% endhighlight %}\n+\n+`flatMap` is a one-to-many DStream operation that creates a new DStream by\n+generating multiple new records from each records of the source DStream. In this case,\n+each line will be split into multiple words, generating the `words` DStream. Next,\n+we want to count the words.\n+\n+{% highlight scala %}\n+// Count each word in each batch\n+val pairs = words.map(word => (word, 1))\n+val wordCounts = pairs.reduceByKey(_ + _)\n+\n+// Print a few of the counts to the console\n+wordCount.print()\n+{% endhighlight %}\n+\n+The `words` DStream is further mapped (one-to-one transformation) to a DStream of `(word,\n+1)` pairs, which is then reduced to get the frequency of words in each batch of data.\n+Finally, `wordCounts.print()` will print a few of the counts generated every second.\n+\n+Note that when these lines are executed, Spark Streaming only sets up the computation it\n+will perform when it is started, and no real processing has started yet. To start the processing\n+after all the transformations have been setup, we finally call\n+\n+{% highlight scala %}\n+ssc.start()             // Start the computation\n+ssc.awaitTermination()  // Wait for the computation to terminate\n+{% endhighlight %}\n+\n+The complete code can be found in the Spark Streaming example\n+[NetworkWordCount](https://github.com/apache/incubator-spark/tree/master/examples/src/main/scala/\n+org/apache/spark/streaming/examples/NetworkWordCount.scala).\n+<br>\n+\n+\n+\n+</div>\n+<div data-lang=\"java\" markdown=\"1\">\n+\n+First, we create a\n+[JavaStreamingContext](api/streaming/index.html#org.apache.spark.streaming.api.java\n+.JavaStreamingContext) object,\n+which is the main entry point for all streaming\n+functionality. Besides Spark's configuration, we specify that any DStream would be processed\n+in 1 second batches.\n+\n+{% highlight java %}\n+// Create a StreamingContext with a SparkConf configuration\n+JavaStreamingContext jssc = StreamingContext(sparkConf, new Duration(1000))\n+{% endhighlight %}\n+\n+Using this context, we then create a new DStream from\n+by specifying the IP address and port of the data server.\n+\n+{% highlight java %}\n+// Create a DStream that will connect to serverIP:serverPort\n+JavaDStream<String> lines = jssc.socketTextStream(serverIP, serverPort);\n+{% endhighlight %}\n+\n+This `lines` DStream represents the stream of data that will be received from the data\n+server. Each records in this stream is a line of text. Then, we want to split the the lines by\n+space into words.\n+\n+{% highlight java %}\n+// Split each line into words\n+JavaDStream<String> words = lines.flatMap(\n+  new FlatMapFunction<String, String>() {\n+    @Override public Iterable<String> call(String x) {\n+      return Lists.newArrayList(x.split(\" \"));\n+    }\n+  });\n+{% endhighlight %}\n+\n+`flatMap` is a DStream operation that creates a new DStream by\n+generating multiple new records from each records of the source DStream. In this case,\n+each line will be split into multiple words, generating the `words` DStream.\n+Note that we defined the transformation using a\n+[FlatMapFunction](api/core/index.html#org.apache.spark.api.java.function.FlatMapFunction) object.\n+As we will discover along the way, there are a number of such convenience classes in the Java API\n+that help define DStream transformations.\n+\n+Next, we want to count the words.\n+\n+{% highlight java %}\n+// Count each word in each batch\n+JavaPairDStream<String, Integer> pairs = words.map(\n+  new PairFunction<String, String, Integer>() {\n+    @Override public Tuple2<String, Integer> call(String s) throws Exception {\n+      return new Tuple2<String, Integer>(s, 1);\n+    }\n+  });\n+JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(\n+  new Function2<Integer, Integer, Integer>() {\n+    @Override public Integer call(Integer i1, Integer i2) throws Exception {\n+      return i1 + i2;\n+    }\n+  });\n+wordCount.print();     // Print a few of the counts to the console\n+{% endhighlight %}\n+\n+The `words` DStream is further mapped (one-to-one transformation) to a DStream of `(word,\n+1)` pairs, using a [PairFunction](api/core/index.html#org.apache.spark.api.java.function\n+.PairFunction) object. Then, it is reduced to get the frequency of words in each batch of data,\n+using a [Function2](api/core/index.html#org.apache.spark.api.java.function.Function2) object.\n+Finally, `wordCounts.print()` will print a few of the counts generated every second.\n+\n+Note that when these lines are executed, Spark Streaming only sets up the computation it\n+will perform when it is started, and no real processing has started yet. To start the processing\n+after all the transformations have been setup, we finally call\n+\n+{% highlight java %}\n+jssc.start();              // Start the computation\n+jssc.awaitTermination();   // Wait for the computation to terminate\n+{% endhighlight %}\n+\n+The complete code can be found in the Spark Streaming example\n+[JavaNetworkWordCount](https://github.com/apache/incubator-spark/tree/master/examples/src/main/java/\n+org/apache/spark/streaming/examples/JavaNetworkWordCount.java).\n+<br>\n+\n+</div>\n+</div>\n+\n+If you have already [downloaded](index.html#downloading) and [built](index.html#building) Spark,\n+you can run this example as follows. You will first need to run Netcat\n+(a small utility found in most Unix-like systems) as a data server by using\n+\n+{% highlight bash %}\n+$ nc -lk 9999\n+{% endhighlight %}\n+\n+Then, in a different terminal, you can start the example by using\n+\n+<div class=\"codetabs\">\n+<div data-lang=\"scala\" markdown=\"1\">\n+{% highlight bash %}\n+$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999\n+{% endhighlight %}\n+</div>\n+<div data-lang=\"java\" markdown=\"1\">\n+{% highlight bash %}\n+$ ./bin/run-example org.apache.spark.streaming.examples.JavaNetworkWordCount local[2] localhost 9999\n+{% endhighlight %}\n+</div>\n+</div>\n \n-# Linking with Spark Streaming\n \n-Add the following SBT or Maven dependency to your project to use Spark Streaming:\n+Then, any lines typed in the terminal running the netcat server will be counted and printed on\n+screen every second. It will look something like this.\n+\n+<table width=\"100%\">\n+    <td>\n+{% highlight bash %}\n+# TERMINAL 1:\n+# Running Netcat\n+\n+$ nc -lk 9999\n+\n+hello world\n+\n+\n+\n+...\n+{% endhighlight %}\n+    </td>\n+    <td width=\"2%\"></td>\n+    <td>\n+{% highlight bash %}\n+# TERMINAL 2: RUNNING NetworkWordCount or JavaNetworkWordCount\n+\n+$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999\n+...\n+-------------------------------------------\n+Time: 1357008430000 ms\n+-------------------------------------------\n+(hello,1)\n+(world,1)\n+...\n+{% endhighlight %}\n+    </td>\n+</table>\n+\n+***************************************************************************************************  \n+\n+# Basics\n+\n+Next, we move beyond the simple example and elaborate on the basics of Spark Streaming that you\n+need to know to write your streaming applications.\n+\n+## Linking\n+\n+To write your own Spark Streaming program, you will have to add the following dependency to your\n+ SBT or Maven project:\n \n     groupId = org.apache.spark\n     artifactId = spark-streaming_{{site.SCALA_VERSION}}\n     version = {{site.SPARK_VERSION}}\n \n-For ingesting data from sources like Kafka and Flume, add the corresponding artifact `spark-streaming-xyz_{{site.SCALA_VERSION}}` to the dependencies. For example, `spark-streaming-kafka_{{site.SCALA_VERSION}}` for Kafka, `spark-streaming-flume_{{site.SCALA_VERSION}}`, etc.  Please refer to the [Apache repository](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.spark%22%20AND%20v%3A%22{{site.SPARK_VERSION}}%22) for the full list of supported sources / artifacts.\n+For ingesting data from sources like Kafka and Flume that are not covered in Spark Streaming core\n+ API, you will have to add the corresponding\n+artifact `spark-streaming-xyz_{{site.SCALA_VERSION}}` to the dependencies. For example,\n+some of the prominent ones are as follows.\n \n-# Initializing Spark Streaming\n-The first thing a Spark Streaming program must do is create a `StreamingContext` object, which tells Spark how to access a cluster. A `StreamingContext` can be created by using\n+\n+<table class=\"table\">\n+<tr><th>Source</th><th>Artifact</th></tr>\n+<tr><td> Kafka </td><td> spark-streaming-kafka_{{site.SCALA_VERSION}} </td></tr>\n+<tr><td> Flume </td><td> spark-streaming-flume_{{site.SCALA_VERSION}} </td></tr>\n+<tr><td> Twitter </td><td> spark-streaming-twitter_{{site.SCALA_VERSION}} </td></tr>\n+<tr><td> ZeroMQ </td><td> spark-streaming-zeromq_{{site.SCALA_VERSION}} </td></tr>\n+<tr><td> MQTT </td><td> spark-streaming-mqtt_{{site.SCALA_VERSION}} </td></tr>\n+<tr><td> </td><td></td></tr>\n+</table>\n+\n+For a more a more updated list, please refer to the\n+[Apache repository](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.spark%22%20AND%20v%3A%22{{site.SPARK_VERSION}}%22)\n+for the full list of supported sources and artifacts.\n+\n+## Initializing\n+\n+<div class=\"codetabs\">\n+<div data-lang=\"scala\" markdown=\"1\">\n+\n+To initialize a Spark Streaming program in Scala, a\n+[`StreamingContext`](api/streaming/index.html#org.apache.spark.streaming.StreamingContext)\n+object has to be created, which is the main entry point of all Spark Streaming functionality.\n+A `StreamingContext` object can be created by using\n \n {% highlight scala %}\n new StreamingContext(master, appName, batchDuration, [sparkHome], [jars])\n {% endhighlight %}\n+</div>\n+<div data-lang=\"java\" markdown=\"1\">\n \n-The `master` parameter is a standard [Spark cluster URL](scala-programming-guide.html#master-urls) and can be \"local\" for local testing. The `appName` is a name of your program, which will be shown on your cluster's web UI. The `batchDuration` is the size of the batches (as explained earlier). This must be set carefully such that the cluster can keep up with the processing of the data streams. Start with something conservative like 5 seconds. See the [Performance Tuning](#setting-the-right-batch-size) section for a detailed discussion. Finally, `sparkHome` and `jars` are optional parameters, which need to be set when running on a cluster to specify the location of your code, as described in the [Spark programming guide](scala-programming-guide.html#deploying-code-on-a-cluster).\n+To initialize a Spark Streaming program in Java, a\n+[`JavaStreamingContext`](api/streaming/index.html#org.apache.spark.streaming.api\n+.java.JavaStreamingContext)\n+object has to be created, which is the main entry point of all Spark Streaming functionality.\n+A `JavaStreamingContext` object can be created by using\n \n {% highlight scala %}\n-new SparkConf(conf, batchDuration)\n+new JavaStreamingContext(master, appName, batchDuration, [sparkHome], [jars])\n+{% endhighlight %}\n+</div>\n+</div>\n+\n+The `master` parameter is a standard [Spark cluster URL](scala-programming-guide.html#master-urls)\n+and can be \"local\" for local testing. The `appName` is a name of your program,\n+which will be shown on your cluster's web UI. The `batchDuration` is the size of the batches\n+(as explained earlier). This must be set carefully such that the cluster can keep up with the\n+processing of the data streams. Start with something conservative like 5 seconds. See the\n+[Performance Tuning](#setting-the-right-batch-size) section for a detailed discussion.\n+Finally, `sparkHome` and `jars` are optional parameters, which need to be set when running\n+on a cluster to specify the location of your code, as described in the\n+[Spark programming guide](scala-programming-guide.html#deploying-code-on-a-cluster).\n+\n+The underlying SparkContext can be accessed as `context.sparkContext`.\n+\n+## DStreams\n+*Discretized Stream* or *DStream* is the basic abstraction provided by Spark Streaming.\n+It represents a continuous stream of data, either the raw data stream received from source,","path":"docs/streaming-programming-guide.md","position":353,"original_position":353,"commit_id":"d5b6196b532b5746e019b959a79ea0cc013a8fc3","original_commit_id":"d5b6196b532b5746e019b959a79ea0cc013a8fc3","user":{"login":"tdas","id":663212,"avatar_url":"https://gravatar.com/avatar/dd4cdf7510ba6e04e617f6c3a3fc62f7?d=https%3A%2F%2Fidenticons.github.com%2Fba3ac35029cf61336ffcc8484e3f4f93.png&r=x","gravatar_id":"dd4cdf7510ba6e04e617f6c3a3fc62f7","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"body":"Changed.","created_at":"2014-01-27T21:00:44Z","updated_at":"2014-01-27T21:00:44Z","html_url":"https://github.com/apache/incubator-spark/pull/497#discussion_r9205587","pull_request_url":"https://api.github.com/repos/apache/incubator-spark/pulls/497","_links":{"self":{"href":"https://api.github.com/repos/apache/incubator-spark/pulls/comments/9205587"},"html":{"href":"https://github.com/apache/incubator-spark/pull/497#discussion_r9205587"},"pull_request":{"href":"https://api.github.com/repos/apache/incubator-spark/pulls/497"}}}}