[SPARK-17813][SQL][KAFKA] Maximum data per trigger by koeninger · Pull Request #15527 · apache/spark

koeninger · 2016-10-18T05:28:24Z

What changes were proposed in this pull request?

maxOffsetsPerTrigger option for rate limiting, proportionally based on volume of different topicpartitions.

How was this patch tested?

Added unit test

…fsets

Testing maxOffsetsPerTrigger requires the per-partition sendMessages testing added in SPARK-17812

…ation

SparkQA · 2016-10-18T06:02:07Z

Test build #67113 has finished for PR 15527 at commit 6c8d459.

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

SparkQA · 2016-10-24T16:56:15Z

Test build #67455 has finished for PR 15527 at commit 76ecafc.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

…alClock

SparkQA · 2016-10-24T17:53:36Z

Test build #67459 has finished for PR 15527 at commit fde4e33.

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

zsxwing

Thanks for your work. The major issue is currentPartitionOffsets should be set when recovering from failure.

zsxwing · 2016-10-27T00:08:49Z

@@ -153,11 +201,7 @@ private[kafka010] case class KafkaSource(



currentPartitionOffsets should be set to untilPartitionOffsets if it's empty. It means recovery from a failure.

Shouldn't it be set to the highest available offset in the streaming metadata log, not the highest available offset in kafka?

currentPartitionOffsets is the last processed offsets. Right? When recovering from the failure, getBatch will be called firstly, then getOffset.

Ok, I see what you were saying, thought you were talking about resetting it in getOffset, not the end passed to getBatch.

zsxwing · 2016-10-27T00:14:21Z

+ StartStream(ProcessingTime(100), clock),
+ AdvanceManualClock(100),
+ // 1 from smallest, 1 from middle, 8 from biggest
+ CheckAnswer(1, 10, 100, 101, 102, 103, 104, 105, 106, 107),


There is a race condition here. The batch may be still running. I figured out the following codes to cover the recovery and fix the race condition finally.
test("maxOffsetsPerTrigger") { valtopic= newTopic() testUtils.createTopic(topic, partitions =3) testUtils.sendMessages(topic, (100 to 200).map(_.toString).toArray, Some(0)) testUtils.sendMessages(topic, (10 to 20).map(_.toString).toArray, Some(1)) testUtils.sendMessages(topic, Array("1"), Some(2)) valreader= spark .readStream .format("kafka") .option("kafka.bootstrap.servers", testUtils.brokerAddress) .option("kafka.metadata.max.age.ms", "1") .option("maxOffsetsPerTrigger", 10) .option("subscribe", topic) .option("startingOffsets", "earliest") valkafka= reader.load() .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") .as[(String, String)] valmapped: org.apache.spark.sql.Dataset[_] = kafka.map(kv => kv._2.toInt) valclock=newStreamManualClockvalwaitUntilBatchProcessed=AssertOnQuery { q => eventually(Timeout(streamingTimeout)) { if (!q.exception.isDefined) { assert(clock.isStreamWaitingAt(clock.getTimeMillis())) } } if (q.exception.isDefined) { throw q.exception.get } true } testStream(mapped)( StartStream(ProcessingTime(100), clock), waitUntilBatchProcessed, // 1 from smallest, 1 from middle, 8 from biggestCheckAnswer(1, 10, 100, 101, 102, 103, 104, 105, 106, 107), AdvanceManualClock(100), waitUntilBatchProcessed, // smallest now empty, 1 more from middle, 9 more from biggestCheckAnswer(1, 10, 100, 101, 102, 103, 104, 105, 106, 107, 11, 108, 109, 110, 111, 112, 113, 114, 115, 116 ), StopStream, StartStream(ProcessingTime(100), clock), waitUntilBatchProcessed, AdvanceManualClock(100), waitUntilBatchProcessed, // smallest now empty, 1 more from middle, 9 more from biggestCheckAnswer(1, 10, 100, 101, 102, 103, 104, 105, 106, 107, 11, 108, 109, 110, 111, 112, 113, 114, 115, 116, 12, 117, 118, 119, 120, 121, 122, 123, 124, 125 ), AdvanceManualClock(100), waitUntilBatchProcessed, // smallest now empty, 1 more from middle, 9 more from biggestCheckAnswer(1, 10, 100, 101, 102, 103, 104, 105, 106, 107, 11, 108, 109, 110, 111, 112, 113, 114, 115, 116, 12, 117, 118, 119, 120, 121, 122, 123, 124, 125, 13, 126, 127, 128, 129, 130, 131, 132, 133, 134 ) ) }
This test fails now because of an issue being fixed in #14553.

FYI, #14553 got merged.

SparkQA · 2016-10-27T03:58:46Z

Test build #67616 has finished for PR 15527 at commit 6a7ff24.

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

zsxwing

LGTM except some style nits.

zsxwing · 2016-10-27T04:42:21Z

+ until: Map[TopicPartition, Long]): Map[TopicPartition, Long] = {
+ val fromNew = fetchNewPartitionEarliestOffsets(until.keySet.diff(from.keySet).toSeq)
+ val sizes = until.flatMap { case (tp, end) =>
+ // If begin isn't defined, something's wrong, but let alert logic in getBatch handle it


nit: use 2 spaces

zsxwing · 2016-10-27T04:42:25Z

+ until
+ } else {
+ until.map { case (tp, end) =>
+ tp -> sizes.get(tp).map { size =>


nit: use 2 spaces

SparkQA · 2016-10-27T14:45:04Z

Test build #67645 has finished for PR 15527 at commit 5e4b468.

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

zsxwing · 2016-10-27T17:30:29Z

LGTM. Thanks! Merging to master and 2.0.

## What changes were proposed in this pull request? maxOffsetsPerTrigger option for rate limiting, proportionally based on volume of different topicpartitions. ## How was this patch tested? Added unit test Author: cody koeninger <cody@koeninger.org> Closes#15527 from koeninger/SPARK-17813. (cherry picked from commit 1042325) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

## What changes were proposed in this pull request? maxOffsetsPerTrigger option for rate limiting, proportionally based on volume of different topicpartitions. ## How was this patch tested? Added unit test Author: cody koeninger <cody@koeninger.org> Closesapache#15527 from koeninger/SPARK-17813.

koeninger added 8 commits October 13, 2016 23:23

[SPARK-17812][SQL][KAFKA] parse json for topicpartitions and offsets
c45ded7

Merge branch 'master' into SPARK-17812
12d3988

[SPARK-17812][SQL][KAFKA] implement specified offsets and assign
3120fd8

[SPARK-17812][SQL][KAFKA] doc and test updates
35bb8c3

[SPARK-17812][SQL][KAFKA] style fixes
2e53e5a

[SPARK-17812][SQL][KAFKA] additional paranoia on reset of starting of…
5e4511f
…fsets

Merge branch 'SPARK-17812' into SPARK-17813
cae967c
Testing maxOffsetsPerTrigger requires the per-partition sendMessages testing added in SPARK-17812

[SPARK-17813][SQL][KAFKA] maxOffsetsPerTrigger proportional implement…
6c8d459
…ation

Merge branch 'master' into SPARK-17813
76ecafc

[SPARK-17813][SQL][KAFKA] fix test for upstream changes to StreamManu…
fde4e33
…alClock

zsxwing requested changes Oct 27, 2016
View reviewed changes

zsxwing reviewed Oct 27, 2016
View reviewed changes

koeninger added 2 commits October 26, 2016 20:42

Merge branch 'master' into SPARK-17813
64fca67

[SPARK-17813][SQL][KAFKA] set current offsets on recovery
6a7ff24

zsxwing requested changes Oct 27, 2016
View reviewed changes

[SPARK-17813][SQL][KAFKA] spacing
5e4b468

asfgit closed this in 1042325Oct 27, 2016

		@@ -153,11 +201,7 @@ private[kafka010] case class KafkaSource(

Uh oh!

Conversation

koeninger commented Oct 18, 2016• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Oct 18, 2016

Uh oh!

SparkQA commented Oct 24, 2016

Uh oh!

SparkQA commented Oct 24, 2016

Uh oh!

zsxwing 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!

SparkQA commented Oct 27, 2016

Uh oh!

zsxwing 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!

SparkQA commented Oct 27, 2016

Uh oh!

zsxwing commented Oct 27, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

koeninger commented Oct 18, 2016•
edited
Loading