Skip to content

[SPARK-17813][SQL][KAFKA] Maximum data per trigger #15527

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
koeninger wants to merge 13 commits into from
Closed

[SPARK-17813][SQL][KAFKA] Maximum data per trigger #15527

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
c45ded7
[SPARK-17812][SQL][KAFKA] parse json for topicpartitions and offsets
koeninger Oct 14, 2016
12d3988
Merge branch 'master' into SPARK-17812
koeninger Oct 14, 2016
3120fd8
[SPARK-17812][SQL][KAFKA] implement specified offsets and assign
koeninger Oct 14, 2016
35bb8c3
[SPARK-17812][SQL][KAFKA] doc and test updates
koeninger Oct 16, 2016
2e53e5a
[SPARK-17812][SQL][KAFKA] style fixes
koeninger Oct 16, 2016
5e4511f
[SPARK-17812][SQL][KAFKA] additional paranoia on reset of starting of…
koeninger Oct 16, 2016
cae967c
Merge branch 'SPARK-17812' into SPARK-17813
koeninger Oct 18, 2016
6c8d459
[SPARK-17813][SQL][KAFKA] maxOffsetsPerTrigger proportional implement…
koeninger Oct 18, 2016
76ecafc
Merge branch 'master' into SPARK-17813
koeninger Oct 24, 2016
fde4e33
[SPARK-17813][SQL][KAFKA] fix test for upstream changes to StreamManu…
koeninger Oct 24, 2016
64fca67
Merge branch 'master' into SPARK-17813
koeninger Oct 27, 2016
6a7ff24
[SPARK-17813][SQL][KAFKA] set current offsets on recovery
koeninger Oct 27, 2016
5e4b468
[SPARK-17813][SQL][KAFKA] spacing
koeninger Oct 27, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
[SPARK-17813][SQL][KAFKA] spacing
  • Loading branch information
@koeninger
koeninger committed Oct 27, 2016
commit 5e4b468111ec20f11fc352de4075637f12e3a499
6 changes: 4 additions & 2 deletions external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,8 @@ private[kafka010] case class KafkaSource(
from: Map[TopicPartition, Long],
until: Map[TopicPartition, Long]): Map[TopicPartition, Long] = {
val fromNew = fetchNewPartitionEarliestOffsets(until.keySet.diff(from.keySet).toSeq)
val sizes = until.flatMap { case (tp, end) =>
val sizes = until.flatMap {
case (tp, end) =>
// If begin isn't defined, something's wrong, but let alert logic in getBatch handle it
Copy link
Member

@zsxwing zsxwing Oct 27, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: use 2 spaces

from.get(tp).orElse(fromNew.get(tp)).flatMap { begin =>
val size = end - begin
Expand All @@ -166,7 +167,8 @@ private[kafka010] case class KafkaSource(
if (total < 1) {
until
} else {
until.map { case (tp, end) =>
until.map {
case (tp, end) =>
tp -> sizes.get(tp).map { size =>
Copy link
Member

@zsxwing zsxwing Oct 27, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: use 2 spaces

val begin = from.get(tp).getOrElse(fromNew(tp))
val prorate = limit * (size / total)
Expand Down