Skip to content

[SPARK-19304] [Streaming] [Kinesis] fix kinesis slow checkpoint recovery #16842

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
Gauravshah wants to merge 6 commits into from
Closed

[SPARK-19304] [Streaming] [Kinesis] fix kinesis slow checkpoint recovery #16842

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b5e544a
added limit to kinesis checkpoint backed rdd to reduce number of reco…
Gauravshah Feb 7, 2017
d3b8c62
fixing test cases on b5e544a8ec326149b7d03773dd7abf8703ee44a2
Gauravshah Feb 8, 2017
274ee27
limiting max records limit to 10k for aws limitations
Gauravshah Feb 22, 2017
4f5edd3
limiting max records limit to 10k for aws limitations
Gauravshah Feb 28, 2017
82499bc
scala coding style fixes, one line per param
Gauravshah Mar 1, 2017
c8efdcf
scala coding style fixes, one line per param
Gauravshah Mar 4, 2017
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
19 changes: 13 additions & 6 deletions ...kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ import org.apache.spark.util.NextIterator
/** Class representing a range of Kinesis sequence numbers. Both sequence numbers are inclusive. */
private[kinesis]
case class SequenceNumberRange(
streamName: String, shardId: String, fromSeqNumber: String, toSeqNumber: String)
streamName: String, shardId: String, fromSeqNumber: String, toSeqNumber: String,
Copy link
Contributor

@brkyvz brkyvz Feb 28, 2017

Choose a reason for hiding this comment

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

one parameter per line:

  streamName: String,
  shardId: String,
  ...

recordCount: Int)
Copy link
Member

@srowen srowen Feb 7, 2017

Choose a reason for hiding this comment

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

Why is this a property of a range -- or when would it not equal (from - to + 1)?

Copy link
Contributor Author

@Gauravshah Gauravshah Feb 7, 2017

Choose a reason for hiding this comment

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

Not sure of a better place to put.
from - to != count. Kinesis seqNumber are in order but are not sequential

Copy link
Member

@srowen srowen Feb 7, 2017

Choose a reason for hiding this comment

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

OK, but is it an 'input' or 'output'? the usage below makes it look like the caller dictates how many records are in the range, but it doesn't know that ahead of time? I probably misunderstand this.

Copy link
Contributor Author

@Gauravshah Gauravshah Feb 7, 2017

Choose a reason for hiding this comment

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

http://docs.aws.amazon.com/streams/latest/dev/key-concepts.html#sequence-number

Copy link
Contributor Author

@Gauravshah Gauravshah Feb 7, 2017

Choose a reason for hiding this comment

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

its an input to spark checkpoint metadata. On streaming KinesisReceiver receives records creates blocks & knows about seqNumber, count. When recovering from checkpoint we read back this information from checkpoint and make aws kinesis getRecords call with fromSeqNumber & limit

Copy link
Contributor

@brkyvz brkyvz Feb 21, 2017

Choose a reason for hiding this comment

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

I'm worried this change will break checkpoint recovery, because we use Java serialization, and be a barrier to users from upgrading.

Copy link
Contributor Author

@Gauravshah Gauravshah Feb 22, 2017

Choose a reason for hiding this comment

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

Not sure on upgrading, since for code upgrade we need to delete the checkpoint directory and start afresh. I did run this patch and was able to serialize the limit into checkpoint, ( not a scala pro though)


/** Class representing an array of Kinesis sequence number ranges */
private[kinesis]
Expand Down Expand Up @@ -138,6 +139,8 @@ class KinesisSequenceRangeIterator(
private val client = new AmazonKinesisClient(credentials)
private val streamName = range.streamName
private val shardId = range.shardId
// AWS limits to maximum of 10k records per get call
private val maxGetRecordsLimit = 10000

private var toSeqNumberReceived = false
private var lastSeqNumber: String = null
Expand All @@ -155,12 +158,14 @@ class KinesisSequenceRangeIterator(

// If the internal iterator has not been initialized,
// then fetch records from starting sequence number
internalIterator = getRecords(ShardIteratorType.AT_SEQUENCE_NUMBER, range.fromSeqNumber)
internalIterator = getRecords(ShardIteratorType.AT_SEQUENCE_NUMBER, range.fromSeqNumber,
range.recordCount)
} else if (!internalIterator.hasNext) {

// If the internal iterator does not have any more records,
// then fetch more records after the last consumed sequence number
internalIterator = getRecords(ShardIteratorType.AFTER_SEQUENCE_NUMBER, lastSeqNumber)
internalIterator = getRecords(ShardIteratorType.AFTER_SEQUENCE_NUMBER, lastSeqNumber,
range.recordCount)
}

if (!internalIterator.hasNext) {
Expand Down Expand Up @@ -193,9 +198,10 @@ class KinesisSequenceRangeIterator(
/**
* Get records starting from or after the given sequence number.
*/
private def getRecords(iteratorType: ShardIteratorType, seqNum: String): Iterator[Record] = {
private def getRecords(iteratorType: ShardIteratorType, seqNum: String,
Copy link
Contributor

@brkyvz brkyvz Mar 3, 2017

Choose a reason for hiding this comment

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

you forgot here

Copy link
Contributor Author

@Gauravshah Gauravshah Mar 4, 2017

Choose a reason for hiding this comment

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

👍

recordCount: Int): Iterator[Record] = {
val shardIterator = getKinesisIterator(iteratorType, seqNum)
val result = getRecordsAndNextKinesisIterator(shardIterator)
val result = getRecordsAndNextKinesisIterator(shardIterator, recordCount)
result._1
}

Expand All @@ -204,10 +210,11 @@ class KinesisSequenceRangeIterator(
* to get records from Kinesis), and get the next shard iterator for next consumption.
*/
private def getRecordsAndNextKinesisIterator(
shardIterator: String): (Iterator[Record], String) = {
shardIterator: String, recordCount: Int): (Iterator[Record], String) = {
Copy link
Contributor

@brkyvz brkyvz Feb 28, 2017

Choose a reason for hiding this comment

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

ditto, one param per line

val getRecordsRequest = new GetRecordsRequest
getRecordsRequest.setRequestCredentials(credentials)
getRecordsRequest.setShardIterator(shardIterator)
getRecordsRequest.setLimit(Math.min(recordCount, this.maxGetRecordsLimit))
val getRecordsResult = retryOrTimeout[GetRecordsResult](
s"getting records using shard iterator") {
client.getRecords(getRecordsRequest)
Expand Down
3 changes: 2 additions & 1 deletion external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,8 @@ private[kinesis] class KinesisReceiver[T](
if (records.size > 0) {
val dataIterator = records.iterator().asScala.map(messageHandler)
val metadata = SequenceNumberRange(streamName, shardId,
records.get(0).getSequenceNumber(), records.get(records.size() - 1).getSequenceNumber())
records.get(0).getSequenceNumber(), records.get(records.size() - 1).getSequenceNumber(),
records.size())
blockGenerator.addMultipleDataWithCallback(dataIterator, metadata)
}
}
Expand Down
4 changes: 2 additions & 2 deletions ...is-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDDSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ abstract class KinesisBackedBlockRDDTests(aggregateTestData: Boolean)
shardIdToSeqNumbers = shardIdToDataAndSeqNumbers.mapValues { _.map { _._2 }}
shardIdToRange = shardIdToSeqNumbers.map { case (shardId, seqNumbers) =>
val seqNumRange = SequenceNumberRange(
testUtils.streamName, shardId, seqNumbers.head, seqNumbers.last)
testUtils.streamName, shardId, seqNumbers.head, seqNumbers.last, seqNumbers.size)
(shardId, seqNumRange)
}
allRanges = shardIdToRange.values.toSeq
Expand Down Expand Up @@ -181,7 +181,7 @@ abstract class KinesisBackedBlockRDDTests(aggregateTestData: Boolean)

// Create the necessary ranges to use in the RDD
val fakeRanges = Array.fill(numPartitions - numPartitionsInKinesis)(
SequenceNumberRanges(SequenceNumberRange("fakeStream", "fakeShardId", "xxx", "yyy")))
SequenceNumberRanges(SequenceNumberRange("fakeStream", "fakeShardId", "xxx", "yyy", 1)))
val realRanges = Array.tabulate(numPartitionsInKinesis) { i =>
val range = shardIdToRange(shardIds(i + (numPartitions - numPartitionsInKinesis)))
SequenceNumberRanges(Array(range))
Expand Down
4 changes: 2 additions & 2 deletions ...al/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,13 +119,13 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun

// Generate block info data for testing
val seqNumRanges1 = SequenceNumberRanges(
SequenceNumberRange("fakeStream", "fakeShardId", "xxx", "yyy"))
SequenceNumberRange("fakeStream", "fakeShardId", "xxx", "yyy", 67))
val blockId1 = StreamBlockId(kinesisStream.id, 123)
val blockInfo1 = ReceivedBlockInfo(
0, None, Some(seqNumRanges1), new BlockManagerBasedStoreResult(blockId1, None))

val seqNumRanges2 = SequenceNumberRanges(
SequenceNumberRange("fakeStream", "fakeShardId", "aaa", "bbb"))
SequenceNumberRange("fakeStream", "fakeShardId", "aaa", "bbb", 89))
val blockId2 = StreamBlockId(kinesisStream.id, 345)
val blockInfo2 = ReceivedBlockInfo(
0, None, Some(seqNumRanges2), new BlockManagerBasedStoreResult(blockId2, None))
Expand Down