{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/26088814","id":26088814,"diff_hunk":"@@ -2,58 +2,154 @@\n layout: global\n title: Spark Streaming + Kafka Integration Guide\n ---\n-[Apache Kafka](http://kafka.apache.org/) is publish-subscribe messaging rethought as a distributed, partitioned, replicated commit log service.  Here we explain how to configure Spark Streaming to receive data from Kafka.\n+[Apache Kafka](http://kafka.apache.org/) is publish-subscribe messaging rethought as a distributed, partitioned, replicated commit log service.  Here we explain how to configure Spark Streaming to receive data from Kafka. There are two approaches to this - the old approach using Receivers and Kafka's high-level API, and a new experimental approach (introduced in Spark 1.3) without using Receivers. They have different programming models, performance characteristics, and semantics guarantees, so read on for more details.  \n \n-1. **Linking:** In your SBT/Maven project definition, link your streaming application against the following artifact (see [Linking section](streaming-programming-guide.html#linking) in the main programming guide for further information).\n+## Approach 1: Receiver-based Approach\n+This approach uses a Receiver to receive the data. The Received is implemented using the Kafka high-level consumer API. As with all receivers, the data received from Kafka through a Receiver is stored in Spark executors, and then jobs launched by Spark Streaming processes the data. \n+\n+However, under default configuration, this approach can loose data under failures (see [receiver reliability](streaming-programming-guide.html#receiver-reliability). To ensure zero-data loss, you have to additionally enable Write Ahead Logs in Spark Streaming. To ensure zero-data loss, enable the Write Ahead Logs (introduced in Spark 1.2) . This synchronously saves all the received Kafka data into write ahead logs on a distributed file system (e.g HDFS), so that all the data can be recovered on failure. Ssee [Deploying section](streaming-programming-guide.html#deploying-applications) in the streaming programming guide for more details on Write Ahead Logs.\n+\n+Next, we discuss how to use this approach in your streaming application.\n+\n+1. **Linking:** For Scala/Java applications using SBT/Maven project definitions, link your streaming application with the following artifact (see [Linking section](streaming-programming-guide.html#linking) in the main programming guide for further information).\n \n \t\tgroupId = org.apache.spark\n \t\tartifactId = spark-streaming-kafka_{{site.SCALA_BINARY_VERSION}}\n \t\tversion = {{site.SPARK_VERSION_SHORT}}\n \n-2. **Programming:** In the streaming application code, import `KafkaUtils` and create input DStream as follows.\n+\tFor Python applications, you will have to add this above library and its dependencies when deploying your application. See the *Deploying* subsection below.\n+\n+2. **Programming:** In the streaming application code, import `KafkaUtils` and create an input DStream as follows.\n \n \t<div class=\"codetabs\">\n \t<div data-lang=\"scala\" markdown=\"1\">\n \t\timport org.apache.spark.streaming.kafka._\n \n-\t\tval kafkaStream = KafkaUtils.createStream(\n-        \tstreamingContext, [zookeeperQuorum], [group id of the consumer], [per-topic number of Kafka partitions to consume])\n+\t\tval kafkaStream = KafkaUtils.createStream(streamingContext, \n+            [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])\n \n-\tSee the [API docs](api/scala/index.html#org.apache.spark.streaming.kafka.KafkaUtils$)\n+    You can also specify the key and value classes and their corresponding decoder classes using variations of `createStream`. See the [API docs](api/scala/index.html#org.apache.spark.streaming.kafka.KafkaUtils$)\n \tand the [example]({{site.SPARK_GITHUB_URL}}/blob/master/examples/scala-2.10/src/main/scala/org/apache/spark/examples/streaming/KafkaWordCount.scala).\n \t</div>\n \t<div data-lang=\"java\" markdown=\"1\">\n \t\timport org.apache.spark.streaming.kafka.*;\n \n-\t\tJavaPairReceiverInputDStream<String, String> kafkaStream = KafkaUtils.createStream(\n-        \tstreamingContext, [zookeeperQuorum], [group id of the consumer], [per-topic number of Kafka partitions to consume]);\n+\t\tJavaPairReceiverInputDStream<String, String> kafkaStream = \n+\t\t\tKafkaUtils.createStream(streamingContext,\n+            [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume]);\n \n-\tSee the [API docs](api/java/index.html?org/apache/spark/streaming/kafka/KafkaUtils.html)\n+    You can also specify the key and value classes and their corresponding decoder classes using variations of `createStream`. See the [API docs](api/java/index.html?org/apache/spark/streaming/kafka/KafkaUtils.html)\n \tand the [example]({{site.SPARK_GITHUB_URL}}/blob/master/examples/scala-2.10/src/main/java/org/apache/spark/examples/streaming/JavaKafkaWordCount.java).\n+\n+\t</div>\n+\t<div data-lang=\"python\" markdown=\"1\">\n+\t\tfrom pyspark.streaming.kafka import KafkaUtils\n+\n+\t\tkafkaStream = KafkaUtils.createStream(streamingContext, \\\n+\t\t\t[ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])\n+\n+\tBy default, the Python API will decode Kafka data as UTF8 encoded strings. You can specify your custom decoding function to decode the byte arrays in Kafka records to any arbitrary data type. See the [API docs](api/python/pyspark.streaming.html#pyspark.streaming.kafka.KafkaUtils)\n+\tand the [example]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/python/streaming/kafka_wordcount.py).\t\n \t</div>\n \t</div>\n \n-\t*Points to remember:*\n+\t**Points to remember:**\n \n \t- Topic partitions in Kafka does not correlate to partitions of RDDs generated in Spark Streaming. So increasing the number of topic-specific partitions in the `KafkaUtils.createStream()` only increases the number of threads using which topics that are consumed within a single receiver. It does not increase the parallelism of Spark in processing the data. Refer to the main document for more information on that.\n \n \t- Multiple Kafka input DStreams can be created with different groups and topics for parallel receiving of data using multiple receivers.\n \n-3. **Deploying:** Package `spark-streaming-kafka_{{site.SCALA_BINARY_VERSION}}` and its dependencies (except `spark-core_{{site.SCALA_BINARY_VERSION}}` and `spark-streaming_{{site.SCALA_BINARY_VERSION}}` which are provided by `spark-submit`) into the application JAR. Then use `spark-submit` to launch your application (see [Deploying section](streaming-programming-guide.html#deploying-applications) in the main programming guide).\n-\n-Note that the Kafka receiver used by default is an\n-[*unreliable* receiver](streaming-programming-guide.html#receiver-reliability) section in the\n-programming guide). In Spark 1.2, we have added an experimental *reliable* Kafka receiver that\n-provides stronger\n-[fault-tolerance guarantees](streaming-programming-guide.html#fault-tolerance-semantics) of zero\n-data loss on failures. This receiver is automatically used when the write ahead log\n-(also introduced in Spark 1.2) is enabled\n-(see [Deployment](#deploying-applications.html) section in the programming guide). This\n-may reduce the receiving throughput of individual Kafka receivers compared to the unreliable\n-receivers, but this can be corrected by running\n-[more receivers in parallel](streaming-programming-guide.html#level-of-parallelism-in-data-receiving)\n-to increase aggregate throughput. Additionally, it is recommended that the replication of the\n-received data within Spark be disabled when the write ahead log is enabled as the log is already stored\n-in a replicated storage system. This can be done by setting the storage level for the input\n-stream to `StorageLevel.MEMORY_AND_DISK_SER` (that is, use\n+\t- If you have enabled Write Ahead Logs with a replicated file system like HDFS, the received data is already being replicated in the log. Hence, the storage level in storage level for the input stream to `StorageLevel.MEMORY_AND_DISK_SER` (that is, use\n `KafkaUtils.createStream(..., StorageLevel.MEMORY_AND_DISK_SER)`).\n+asdasd","path":"docs/streaming-kafka-integration.md","position":90,"original_position":90,"commit_id":"86c4c2ad23d1aa02d995361211870f12ff812caa","original_commit_id":"86c4c2ad23d1aa02d995361211870f12ff812caa","user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"body":"Probably didn't mean to include this :)","created_at":"2015-03-10T00:01:32Z","updated_at":"2015-03-10T00:01:32Z","html_url":"https://github.com/apache/spark/pull/4956#discussion_r26088814","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/4956","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/26088814"},"html":{"href":"https://github.com/apache/spark/pull/4956#discussion_r26088814"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/4956"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/4956","id":30808553,"html_url":"https://github.com/apache/spark/pull/4956","diff_url":"https://github.com/apache/spark/pull/4956.diff","patch_url":"https://github.com/apache/spark/pull/4956.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/4956","number":4956,"state":"open","locked":false,"title":"[SPARK-6128][Streaming][Documentation] Updates to Spark Streaming Programming Guide","user":{"login":"tdas","id":663212,"avatar_url":"https://avatars.githubusercontent.com/u/663212?v=3","gravatar_id":"","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":"Updates to the documentation are as follows: \r\n\r\n- Added information on Kafka Direct API and Kafka Python API\r\n- Added joins to the main streaming guide\r\n- Improved details on the fault-tolerance semantics\r\n\r\nGenerated docs located here\r\nhttp://people.apache.org/~tdas/spark-1.3.0-temp-docs/streaming-programming-guide.html#fault-tolerance-semantics","created_at":"2015-03-09T23:00:21Z","updated_at":"2015-03-10T00:01:32Z","closed_at":null,"merged_at":null,"merge_commit_sha":"2c4dfb4f5f00a5e308d2d42807deece7b9572272","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/4956/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/4956/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/4956/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/86c4c2ad23d1aa02d995361211870f12ff812caa","head":{"label":"tdas:streaming-guide-update-1.3","ref":"streaming-guide-update-1.3","sha":"86c4c2ad23d1aa02d995361211870f12ff812caa","user":{"login":"tdas","id":663212,"avatar_url":"https://avatars.githubusercontent.com/u/663212?v=3","gravatar_id":"","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},"repo":{"id":17603343,"name":"spark","full_name":"tdas/spark","owner":{"login":"tdas","id":663212,"avatar_url":"https://avatars.githubusercontent.com/u/663212?v=3","gravatar_id":"","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},"private":false,"html_url":"https://github.com/tdas/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/tdas/spark","forks_url":"https://api.github.com/repos/tdas/spark/forks","keys_url":"https://api.github.com/repos/tdas/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tdas/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tdas/spark/teams","hooks_url":"https://api.github.com/repos/tdas/spark/hooks","issue_events_url":"https://api.github.com/repos/tdas/spark/issues/events{/number}","events_url":"https://api.github.com/repos/tdas/spark/events","assignees_url":"https://api.github.com/repos/tdas/spark/assignees{/user}","branches_url":"https://api.github.com/repos/tdas/spark/branches{/branch}","tags_url":"https://api.github.com/repos/tdas/spark/tags","blobs_url":"https://api.github.com/repos/tdas/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tdas/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tdas/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/tdas/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tdas/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/tdas/spark/languages","stargazers_url":"https://api.github.com/repos/tdas/spark/stargazers","contributors_url":"https://api.github.com/repos/tdas/spark/contributors","subscribers_url":"https://api.github.com/repos/tdas/spark/subscribers","subscription_url":"https://api.github.com/repos/tdas/spark/subscription","commits_url":"https://api.github.com/repos/tdas/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/tdas/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/tdas/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/tdas/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/tdas/spark/contents/{+path}","compare_url":"https://api.github.com/repos/tdas/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tdas/spark/merges","archive_url":"https://api.github.com/repos/tdas/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tdas/spark/downloads","issues_url":"https://api.github.com/repos/tdas/spark/issues{/number}","pulls_url":"https://api.github.com/repos/tdas/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/tdas/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/tdas/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tdas/spark/labels{/name}","releases_url":"https://api.github.com/repos/tdas/spark/releases{/id}","created_at":"2014-03-10T18:19:48Z","updated_at":"2014-10-09T07:43:39Z","pushed_at":"2015-03-09T22:50:21Z","git_url":"git://github.com/tdas/spark.git","ssh_url":"git@github.com:tdas/spark.git","clone_url":"https://github.com/tdas/spark.git","svn_url":"https://github.com/tdas/spark","homepage":null,"size":94385,"stargazers_count":1,"watchers_count":1,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":1,"forks":1,"open_issues":1,"watchers":1,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"70f88148bb04161a1a4968230d8e3fc7e3f8321a","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-03-09T23:24:28Z","pushed_at":"2015-03-09T23:24:28Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":1356185,"stargazers_count":3122,"watchers_count":3122,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2707,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":280,"forks":2707,"open_issues":280,"watchers":3122,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/4956"},"html":{"href":"https://github.com/apache/spark/pull/4956"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/4956"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/4956/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/4956/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/4956/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/86c4c2ad23d1aa02d995361211870f12ff812caa"}}}}