[SPARK-8390][Streaming][Kafka] fix docs related to HasOffsetRanges #6863

koeninger · 2015-06-17T22:10:22Z

No description provided.

…f the existing java direct stream api

…leanup

SparkQA · 2015-06-17T22:26:37Z

Test build #35065 has finished for PR 6863 at commit bb4336b.

This patch fails some tests.
This patch merges cleanly.
This patch adds no public classes.

koeninger · 2015-06-17T23:24:30Z

I don't think a doc change caused a python test failure.
On Jun 17, 2015 5:27 PM, "UCB AMPLab" notifications@github.com wrote:

Merged build finished. Test FAILed.

—
Reply to this email directly or view it on GitHub
#6863 (comment).

tdas · 2015-06-18T02:15:39Z

docs/streaming-kafka-integration.md


- *Efficiency:* Achieving zero-data loss in the first approach required the data to be stored in a Write Ahead Log, which further replicated the data. This is actually inefficient as the data effectively gets replicated twice - once by Kafka, and a second time by the Write Ahead Log. This second approach eliminate the problem as there is no receiver, and hence no need for Write Ahead Logs.
+- *Efficiency:* Achieving zero-data loss in the first approach required the data to be stored in a Write Ahead Log, which further replicated the data. This is actually inefficient as the data effectively gets replicated twice - once by Kafka, and a second time by the Write Ahead Log. This second approach eliminates the problem as there is no receiver, and hence no need for Write Ahead Logs.  As long as you have sufficient Kafka retention, messages can be recovered from Kafka.


Are there two spaces before "As long as"

tdas · 2015-06-18T02:20:30Z

If the java example is updated to the access the offsets, then the scala example should also be updated.

koeninger · 2015-06-18T03:05:14Z

Yeah, the spacing in that document in general is a mess (mix of tabs and spaces, some 2 spaces between sentences, etc). I cleaned it up somewhat. Also further fixed the scala / java offset ranges examples, the java one is pretty much a C&P from the test at this point.

SparkQA · 2015-06-18T04:02:11Z

Test build #35087 has finished for PR 6863 at commit b108c9d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2015-06-18T23:08:51Z

docs/streaming-kafka-integration.md


- *Exactly-once semantics:* The first approach uses Kafka's high level API to store consumed offsets in Zookeeper. This is traditionally the way to consume data from Kafka. While this approach (in combination with write ahead logs) can ensure zero data loss (i.e. at-least once semantics), there is a small chance some records may get consumed twice under some failures. This occurs because of inconsistencies between data reliably received by Spark Streaming and offsets tracked by Zookeeper. Hence, in this second approach, we use simple Kafka API that does not use Zookeeper and offsets tracked only by Spark Streaming within its checkpoints. This eliminates inconsistencies between Spark Streaming and Zookeeper/Kafka, and so each record is received by Spark Streaming effectively exactly once despite failures.
+- *Exactly-once semantics:* The first approach uses Kafka's high level API to store consumed offsets in Zookeeper. This is traditionally the way to consume data from Kafka. While this approach (in combination with write ahead logs) can ensure zero data loss (i.e. at-least once semantics), there is a small chance some records may get consumed twice under some failures. This occurs because of inconsistencies between data reliably received by Spark Streaming and offsets tracked by Zookeeper. Hence, in this second approach, we use simple Kafka API that does not use Zookeeper. Offsets are tracked by Spark Streaming within its checkpoints. This eliminates inconsistencies between Spark Streaming and Zookeeper/Kafka, and so each record is received by Spark Streaming effectively exactly once despite failures. In order to achieve exactly-once semantics for output of your results, your save method must be either idempotent, or an atomic transaction that saves results and offsets in your own data store.


"save method" --> "output operation that saves the data to external data stores"

Also refer to the fault-tolerance section in the main programming guide. This stuff is common for everything.
*nit: Can you capitalize Write Ahead Logs

tdas · 2015-06-18T23:13:50Z

Could you update the example to something that accesses the offsets in the foreachRDD?

var offsetRanges: OffsetRange[] = _
...
kafkaStream.transform { rdd => 
   offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
   rdd
}.map { ... }.reduceByKey { ... }.foreachRDD { rdd => 
   // use offsetRanges to do something, maybe print the ranges
}

Do the same for Java and Scala

…approach shown in docs actually compiles + runs

SparkQA · 2015-06-19T17:19:42Z

Test build #35279 has finished for PR 6863 at commit 3744492.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-06-19T17:21:14Z

Test build #35280 has finished for PR 6863 at commit 26a06bd.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2015-06-19T20:15:22Z

So the JIRA was about updating the examples actually. Its great that you have updated the docs AND the tests, but it would ideal if the examples DirectKafkaWordCount and JavaDirectKafkaWordCount are updated to show how the offset ranges can be accessed. Since you have updated the tests, mind updating the examples as well?

koeninger · 2015-06-19T20:28:44Z

The word count examples don't have any need of accessing offsets.

Wouldn't it be better to have separate examples? I don't want someone
thinking they need to do all this typecast hoop jumping just to get a word
count

On Fri, Jun 19, 2015 at 3:16 PM, Tathagata Das notifications@github.com
wrote:

So the JIRA was about updating the examples actually. Its great that you
have updated the docs AND the tests, but it would ideal if the examples
DirectKafkaWordCount and JavaDirectKafkaWordCount are updated to show how
the offset ranges can be accessed. Since you have updated the tests, mind
updating the examples as well?

—
Reply to this email directly or view it on GitHub
#6863 (comment).

tdas · 2015-06-19T21:23:29Z

That is a good point. Then make two different examples one for Java and
Scala.
BTW, I would like to have this for 1.4.1, so gotta merge soon. So I think I
will merge this, and you can make another PR with the same JIRA to add the
new examples.

On Fri, Jun 19, 2015 at 1:29 PM, Cody Koeninger notifications@github.com
wrote:

The word count examples don't have any need of accessing offsets.

Wouldn't it be better to have separate examples? I don't want someone
thinking they need to do all this typecast hoop jumping just to get a word
count

On Fri, Jun 19, 2015 at 3:16 PM, Tathagata Das notifications@github.com
wrote:

So the JIRA was about updating the examples actually. Its great that you
have updated the docs AND the tests, but it would ideal if the examples
DirectKafkaWordCount and JavaDirectKafkaWordCount are updated to show how
the offset ranges can be accessed. Since you have updated the tests, mind
updating the examples as well?

—
Reply to this email directly or view it on GitHub
#6863 (comment).

—
Reply to this email directly or view it on GitHub
#6863 (comment).

tdas · 2015-06-19T21:25:21Z

Could you update title to get the ordering right? [JIRA][Streaming][Kafka]. Otherwise LGTM.

koeninger · 2015-06-19T22:43:11Z

Title should be right now
On Jun 19, 2015 4:26 PM, "Tathagata Das" notifications@github.com wrote:

Could you update title to get the ordering right?
[JIRA][Streaming][Kafka]. Otherwise LGTM.

—
Reply to this email directly or view it on GitHub
#6863 (comment).

Author: cody koeninger <cody@koeninger.org> Closes #6863 from koeninger/SPARK-8390 and squashes the following commits: 26a06bd [cody koeninger] Merge branch 'master' into SPARK-8390 3744492 [cody koeninger] [Streaming][Kafka][SPARK-8390] doc changes per TD, test to make sure approach shown in docs actually compiles + runs b108c9d [cody koeninger] [Streaming][Kafka][SPARK-8390] further doc fixes, clean up spacing bb4336b [cody koeninger] [Streaming][Kafka][SPARK-8390] fix docs related to HasOffsetRanges, cleanup 3f3c57a [cody koeninger] [Streaming][Kafka][SPARK-8389] Example of getting offset ranges out of the existing java direct stream api

tdas · 2015-06-20T00:45:38Z

I have merged this to master and branch-1.4, thanks man!

Author: cody koeninger <cody@koeninger.org> Closes apache#6863 from koeninger/SPARK-8390 and squashes the following commits: 26a06bd [cody koeninger] Merge branch 'master' into SPARK-8390 3744492 [cody koeninger] [Streaming][Kafka][SPARK-8390] doc changes per TD, test to make sure approach shown in docs actually compiles + runs b108c9d [cody koeninger] [Streaming][Kafka][SPARK-8390] further doc fixes, clean up spacing bb4336b [cody koeninger] [Streaming][Kafka][SPARK-8390] fix docs related to HasOffsetRanges, cleanup 3f3c57a [cody koeninger] [Streaming][Kafka][SPARK-8389] Example of getting offset ranges out of the existing java direct stream api

koeninger added 2 commits June 16, 2015 10:18

[Streaming][Kafka][SPARK-8389] Example of getting offset ranges out o…

3f3c57a

…f the existing java direct stream api

[Streaming][Kafka][SPARK-8390] fix docs related to HasOffsetRanges, c…

bb4336b

…leanup

tdas reviewed Jun 18, 2015
View reviewed changes

[Streaming][Kafka][SPARK-8390] further doc fixes, clean up spacing

b108c9d

tdas reviewed Jun 18, 2015
View reviewed changes

koeninger added 2 commits June 19, 2015 11:12

[Streaming][Kafka][SPARK-8390] doc changes per TD, test to make sure …

3744492

…approach shown in docs actually compiles + runs

Merge branch 'master' into SPARK-8390

26a06bd

koeninger changed the title ~~[Streaming][Kafka][SPARK-8390] fix docs related to HasOffsetRanges~~ [SPARK-8390][Streaming][Kafka] fix docs related to HasOffsetRanges Jun 19, 2015

asfgit closed this in b305e37 Jun 20, 2015

tdas mentioned this pull request Jun 20, 2015

[Streaming][Kafka][SPARK-8389] Example of getting offset ranges out o… #6846

Closed

isaacl mentioned this pull request Jul 26, 2016

Make offsetRanges variable volatile #14372

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-8390][Streaming][Kafka] fix docs related to HasOffsetRanges #6863

[SPARK-8390][Streaming][Kafka] fix docs related to HasOffsetRanges #6863

koeninger commented Jun 17, 2015

SparkQA commented Jun 17, 2015

koeninger commented Jun 17, 2015

tdas Jun 18, 2015

tdas commented Jun 18, 2015

koeninger commented Jun 18, 2015

SparkQA commented Jun 18, 2015

tdas Jun 18, 2015

tdas commented Jun 18, 2015

SparkQA commented Jun 19, 2015

SparkQA commented Jun 19, 2015

tdas commented Jun 19, 2015

koeninger commented Jun 19, 2015

tdas commented Jun 19, 2015

tdas commented Jun 19, 2015

koeninger commented Jun 19, 2015

tdas commented Jun 20, 2015


		- Efficiency: Achieving zero-data loss in the first approach required the data to be stored in a Write Ahead Log, which further replicated the data. This is actually inefficient as the data effectively gets replicated twice - once by Kafka, and a second time by the Write Ahead Log. This second approach eliminate the problem as there is no receiver, and hence no need for Write Ahead Logs.
		- Efficiency: Achieving zero-data loss in the first approach required the data to be stored in a Write Ahead Log, which further replicated the data. This is actually inefficient as the data effectively gets replicated twice - once by Kafka, and a second time by the Write Ahead Log. This second approach eliminates the problem as there is no receiver, and hence no need for Write Ahead Logs. As long as you have sufficient Kafka retention, messages can be recovered from Kafka.


		- Exactly-once semantics: The first approach uses Kafka's high level API to store consumed offsets in Zookeeper. This is traditionally the way to consume data from Kafka. While this approach (in combination with write ahead logs) can ensure zero data loss (i.e. at-least once semantics), there is a small chance some records may get consumed twice under some failures. This occurs because of inconsistencies between data reliably received by Spark Streaming and offsets tracked by Zookeeper. Hence, in this second approach, we use simple Kafka API that does not use Zookeeper and offsets tracked only by Spark Streaming within its checkpoints. This eliminates inconsistencies between Spark Streaming and Zookeeper/Kafka, and so each record is received by Spark Streaming effectively exactly once despite failures.
		- Exactly-once semantics: The first approach uses Kafka's high level API to store consumed offsets in Zookeeper. This is traditionally the way to consume data from Kafka. While this approach (in combination with write ahead logs) can ensure zero data loss (i.e. at-least once semantics), there is a small chance some records may get consumed twice under some failures. This occurs because of inconsistencies between data reliably received by Spark Streaming and offsets tracked by Zookeeper. Hence, in this second approach, we use simple Kafka API that does not use Zookeeper. Offsets are tracked by Spark Streaming within its checkpoints. This eliminates inconsistencies between Spark Streaming and Zookeeper/Kafka, and so each record is received by Spark Streaming effectively exactly once despite failures. In order to achieve exactly-once semantics for output of your results, your save method must be either idempotent, or an atomic transaction that saves results and offsets in your own data store.

[SPARK-8390][Streaming][Kafka] fix docs related to HasOffsetRanges #6863

[SPARK-8390][Streaming][Kafka] fix docs related to HasOffsetRanges #6863

Conversation

koeninger commented Jun 17, 2015

SparkQA commented Jun 17, 2015

koeninger commented Jun 17, 2015

tdas Jun 18, 2015

Choose a reason for hiding this comment

tdas commented Jun 18, 2015

koeninger commented Jun 18, 2015

SparkQA commented Jun 18, 2015

tdas Jun 18, 2015

Choose a reason for hiding this comment

tdas commented Jun 18, 2015

SparkQA commented Jun 19, 2015

SparkQA commented Jun 19, 2015

tdas commented Jun 19, 2015

koeninger commented Jun 19, 2015

tdas commented Jun 19, 2015

tdas commented Jun 19, 2015

koeninger commented Jun 19, 2015

tdas commented Jun 20, 2015