Skip to content

[SPARK-11681][Streaming] Correctly update state timestamp even when state is not updated #9648

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
tdas wants to merge 3 commits into from
Closed

[SPARK-11681][Streaming] Correctly update state timestamp even when state is not updated #9648

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
92694c7
Fixed bug and added unit tests
tdas Nov 12, 2015
53a7c2c
Remove bad changes
tdas Nov 12, 2015
7b345cc
Merge remote-tracking branch 'apache-github/master' into SPARK-11681
tdas Nov 13, 2015
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
Next Next commit
Remove bad changes
  • Loading branch information
@tdas
tdas committed Nov 12, 2015
commit 53a7c2cd8769032006067ed2da5a893e0cabe850
2 changes: 1 addition & 1 deletion streaming/src/main/scala/org/apache/spark/streaming/rdd/TrackStateRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ import org.apache.spark._
private[streaming] case class TrackStateRDDRecord[K, S, E](
var stateMap: StateMap[K, S], var emittedRecords: Seq[E])

object TrackStateRDDRecord {
private[streaming] object TrackStateRDDRecord {
def updateRecordWithData[K: ClassTag, V: ClassTag, S: ClassTag, E: ClassTag](
prevRecord: Option[TrackStateRDDRecord[K, S, E]],
dataIterator: Iterator[(K, V)],
Expand Down
5 changes: 1 addition & 4 deletions streaming/src/test/scala/org/apache/spark/streaming/rdd/TrackStateRDDSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,18 +226,16 @@ class TrackStateRDDSuite extends SparkFunSuite with BeforeAndAfterAll {
// Assert that the function was called only for the keys present in the data
assert(TrackStateRDDSuite.touchedStateKeys.size === testData.size,
"More number of keys are being touched than that is expected")

assert(TrackStateRDDSuite.touchedStateKeys.toSet === testData.toMap.keys,
"Keys not in the data are being touched unexpectedly")

// Assert that the test RDD's data has not changed
//assertRDD(initStateRDD, initStateWthTime, Set.empty)
assertRDD(initStateRDD, initStateWthTime, Set.empty)
newStateRDD
}

// Test no-op, no state should change
testStateUpdates(initStateRDD, Seq(), initStateWthTime) // should not scan any state

testStateUpdates(
initStateRDD, Seq(("k1", 0)), initStateWthTime) // should not update existing state
testStateUpdates(
Expand All @@ -251,7 +249,6 @@ class TrackStateRDDSuite extends SparkFunSuite with BeforeAndAfterAll {
Set(("k1", 0, initTime), ("k2", 0, initTime), ("k3", 0, updateTime), ("k4", 0, updateTime)))

// Test updating of state
println("---------------")
val rdd3 = testStateUpdates(
initStateRDD, Seq(("k1", 1)), // should increment k1's state 0 -> 1
Set(("k1", 1, updateTime), ("k2", 0, initTime)))
Expand Down