[SPARK-16312][Follow-up][STREAMING][KAFKA][DOC] Add java code snippet for Kafka 0.10 integration doc #15679

lw-lin · 2016-10-29T01:20:07Z

What changes were proposed in this pull request?

added java code snippet for Kafka 0.10 integration doc

How was this patch tested?

SKIP_API=1 jekyll build

Screenshot

lw-lin · 2016-10-29T01:21:26Z

@koeninger @zsxwing it'd be great if you can take a look, thanks!

lw-lin · 2016-10-29T01:26:21Z

docs/streaming-kafka-0-10-integration.md

+	    OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+
+	    // some time later, after outputs have completed
+	    ((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);


i personally feel it'd be strange while we can stream.asInstanceOf[CanCommitOffsets].commitAsync(...) in scala, we must ((CanCommitOffsets) stream**.inputDStream()**).commitAsync(...) in java? I can open a pr to fix this when needed. @koeninger @zsxwing options please?

I think it's far too late to fix those issues at this point. DStreams return an RDD, not a parameterized type. KafkaUtils methods return DStreams and RDDs, not an implementation specific type.

thanks Cody. sorry for not being clear, but my point was that the Java kafka input stream does not implements CanCommitOffsets, thus it has to delegate commitAsync(...) explicitly to stream.inputDStream(), which is a scala input stream which implements CanCommitOffsets.

should createDirectStream() return a java kafka inputdstream that also implements CanCommitOffsets? so people can write:

((CanCommitOffsets) stream).commitAsync(offsetRanges);

rather than

((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);

I understood your point. My point is that you have to do the same kind of delegation to get access to HasOffsetRanges on a java rdd, and you're unlikely to be able to fix that kind of thing at this point without either changing the interfaces for dstream, or exposing implementation classes, which Spark is historically very much against.

SparkQA · 2016-10-29T01:39:40Z

Test build #67735 has finished for PR 15679 at commit 9dc8e0f.

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

lw-lin · 2016-10-29T01:46:22Z

Oh there's a on-going 2.0.2 RC cut. It'd be great if we can make this into 2.0.2. So also cc @srowen to take a look, thanks!

srowen

Minor style things

srowen · 2016-10-29T11:39:54Z

docs/streaming-kafka-0-10-integration.md

@@ -85,6 +121,20 @@ If you have a use case that is better suited to batch processing, you can create

 </div>
 <div data-lang="java" markdown="1">
+	// Import dependencies and create kafka params as in Create Direct Stream above
+
+	OffsetRange[] offsetRanges = new OffsetRange[]{


Nit: you can just write ... = { ... } in Java for a static array declaration.

done -- thanks!

srowen · 2016-10-29T11:40:19Z

docs/streaming-kafka-0-10-integration.md

+	  OffsetRange.create("test", 1, 0, 100)
+	};
+
+	JavaRDD<ConsumerRecord<String, String>> rdd = KafkaUtils.<String, String>createRDD(


I don't think <String, String> is needed after KafkaUtils in Java 7?

srowen · 2016-10-29T11:40:43Z

docs/streaming-kafka-0-10-integration.md

+	    final OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+	    rdd.foreachPartition(new VoidFunction<Iterator<ConsumerRecord<String, String>>>() {
+	      @Override
+	      public void call(Iterator<ConsumerRecord<String, String>> consumerRecords) throws Exception {


Can this actually throw Exception?

done -- my bad

srowen · 2016-10-29T11:41:08Z

docs/streaming-kafka-0-10-integration.md

+
+	// begin from the the offsets committed to the database
+	Map<TopicPartition, Long> fromOffsets = new HashMap<>();
+	for (resultSet: selectOffsetsFromYourDatabase)


Space before colon

srowen · 2016-10-29T11:41:25Z

docs/streaming-kafka-0-10-integration.md

+
+	    Object results = yourCalculation(rdd);
+
+	    yourTransactionBlock {


What is this syntax? don't think this is Java

oh, just to keep it consistent with the scala counterpart

I agree with Sean, this would probably be clearer if it was changed to a comment like
// begin your transaction
...
// end your transaction

well received; thanks!

srowen · 2016-10-29T11:42:17Z

docs/streaming-kafka-0-10-integration.md

+	import org.apache.kafka.common.serialization.StringDeserializer;
+	import scala.Tuple2;
+
+	Map<String, Object> kafkaParams = new HashMap<String, Object>();


Q: is the map value type really Object for the Kafka API?

A: yes -- please see https://github.com/apache/kafka/blob/0.10.0/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L539

SparkQA · 2016-10-29T13:49:53Z

Test build #67760 has finished for PR 15679 at commit 400ae12.

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

koeninger · 2016-10-29T14:36:22Z

Were these extracted from compiled example projects, or just written up?

lw-lin · 2016-10-29T14:40:24Z

@koeninger thanks. These were extracted from my local compiled example projects -- no compiled kafka010 examples yet in the Spark repo, right?

koeninger · 2016-10-29T15:00:47Z

docs/streaming-kafka-0-10-integration.md

 	}.toMap

+	import org.apache.spark.streaming.kafka010.ConsumerStrategies.Assign


This import is in an awkward place, omit it or at least move it to the top of this example.

koeninger · 2016-10-29T15:07:30Z

Thanks for working on this, couple minor things to fix but otherwise looks good.

SparkQA · 2016-10-29T15:54:45Z

Test build #67766 has finished for PR 15679 at commit bd622e1.

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

SparkQA · 2016-10-29T15:59:36Z

Test build #67767 has finished for PR 15679 at commit 8be3c8b.

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

koeninger · 2016-10-29T16:18:01Z

LGTM

srowen · 2016-10-30T09:32:51Z

merged to master/2.0

… for Kafka 0.10 integration doc ## What changes were proposed in this pull request? added java code snippet for Kafka 0.10 integration doc ## How was this patch tested? SKIP_API=1 jekyll build ## Screenshot ![kafka-doc](https://cloud.githubusercontent.com/assets/15843379/19826272/bf0d8a4c-9db8-11e6-9e40-1396723df4bc.png) Author: Liwei Lin <lwlin7@gmail.com> Closes #15679 from lw-lin/kafka-010-examples. (cherry picked from commit 505b927) Signed-off-by: Sean Owen <sowen@cloudera.com>

lw-lin · 2016-10-30T11:05:18Z

@srowen @koeninger thanks for the review!

… for Kafka 0.10 integration doc ## What changes were proposed in this pull request? added java code snippet for Kafka 0.10 integration doc ## How was this patch tested? SKIP_API=1 jekyll build ## Screenshot ![kafka-doc](https://cloud.githubusercontent.com/assets/15843379/19826272/bf0d8a4c-9db8-11e6-9e40-1396723df4bc.png) Author: Liwei Lin <lwlin7@gmail.com> Closes apache#15679 from lw-lin/kafka-010-examples.

Add Java code snippet for Kafka guide

9dc8e0f

lw-lin commented Oct 29, 2016

View reviewed changes

srowen requested changes Oct 29, 2016

View reviewed changes

Address comments

400ae12

koeninger reviewed Oct 29, 2016

View reviewed changes

lw-lin added 2 commits October 29, 2016 23:34

Address Cody's comments

bd622e1

Fix space

8be3c8b

asfgit closed this in 505b927 Oct 30, 2016

lw-lin deleted the kafka-010-examples branch October 30, 2016 11:05


		Object results = yourCalculation(rdd);

		yourTransactionBlock {

		}.toMap

		import org.apache.spark.streaming.kafka010.ConsumerStrategies.Assign

[SPARK-16312][Follow-up][STREAMING][KAFKA][DOC] Add java code snippet for Kafka 0.10 integration doc #15679

[SPARK-16312][Follow-up][STREAMING][KAFKA][DOC] Add java code snippet for Kafka 0.10 integration doc #15679

Uh oh!

Conversation

lw-lin commented Oct 29, 2016

What changes were proposed in this pull request?

How was this patch tested?

Screenshot

Uh oh!

lw-lin commented Oct 29, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lw-lin Oct 29, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Oct 29, 2016

Uh oh!

lw-lin commented Oct 29, 2016

Uh oh!

srowen left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Oct 29, 2016

Uh oh!

koeninger commented Oct 29, 2016

Uh oh!

lw-lin commented Oct 29, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

koeninger commented Oct 29, 2016

Uh oh!

SparkQA commented Oct 29, 2016

Uh oh!

SparkQA commented Oct 29, 2016

Uh oh!

koeninger commented Oct 29, 2016

Uh oh!

srowen commented Oct 30, 2016

Uh oh!

lw-lin commented Oct 30, 2016

Uh oh!

Uh oh!

lw-lin Oct 29, 2016 •

edited

Loading