Skip to content

[SPARK-18894][SS] Fix event time watermark delay threshold specified in months or years #16304

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 11 commits into from
Closed

[SPARK-18894][SS] Fix event time watermark delay threshold specified in months or years #16304

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
7fc3411
No event time stats when no watermark is set
tdas Dec 16, 2016
33a7d1b
Disallow dfelay threshold in months or years
tdas Dec 16, 2016
8c603f4
Minor
tdas Dec 16, 2016
67ad327
Changed fix
tdas Dec 16, 2016
736c903
Undo comments
tdas Dec 16, 2016
0a42fab
Updated test
tdas Dec 16, 2016
9d3175a
Minor
tdas Dec 16, 2016
40c200f
Minor
tdas Dec 16, 2016
b9a37bf
Merge remote-tracking branch 'apache-github/master' into SPARK-18834-1
tdas Dec 20, 2016
debc748
Fixed deadlock
tdas Dec 20, 2016
29f0037
Allow negative watermark
tdas Dec 21, 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
7 changes: 6 additions & 1 deletion ...core/src/main/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,11 @@ case class EventTimeWatermarkExec(
child: SparkPlan) extends SparkPlan {

val eventTimeStats = new EventTimeStatsAccum()
val delayMs = {
val millisPerMonth = CalendarInterval.MICROS_PER_DAY / 1000 * 31
delay.milliseconds + delay.months * millisPerMonth
}

sparkContext.register(eventTimeStats)

override protected def doExecute(): RDD[InternalRow] = {
Expand All @@ -101,7 +106,7 @@ case class EventTimeWatermarkExec(
if (a semanticEquals eventTime) {
val updatedMetadata = new MetadataBuilder()
.withMetadata(a.metadata)
.putLong(EventTimeWatermark.delayKey, delay.milliseconds)
.putLong(EventTimeWatermark.delayKey, delayMs)
.build()

a.withMetadata(updatedMetadata)
Expand Down
8 changes: 6 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ import scala.collection.JavaConverters._

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.streaming._
import org.apache.spark.sql.streaming.StreamingQueryListener.QueryProgressEvent
Expand Down Expand Up @@ -84,6 +84,8 @@ trait ProgressReporter extends Logging {
private val timestampFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'") // ISO8601
timestampFormat.setTimeZone(TimeZone.getTimeZone("UTC"))

private val hasEventTime = logicalPlan.collect { case e: EventTimeWatermark => e }.nonEmpty

@volatile
protected var currentStatus: StreamingQueryStatus = {
new StreamingQueryStatus(
Expand Down Expand Up @@ -182,7 +184,9 @@ trait ProgressReporter extends Logging {

/** Extracts statistics from the most recent query execution. */
private def extractExecutionStats(hasNewData: Boolean): ExecutionStats = {
val watermarkTimestamp = Map("watermark" -> formatTimestamp(offsetSeqMetadata.batchWatermarkMs))
val watermarkTimestamp =
if (hasEventTime) Map("watermark" -> formatTimestamp(offsetSeqMetadata.batchWatermarkMs))
else Map.empty[String, String]

if (!hasNewData) {
return ExecutionStats(Map.empty, Seq.empty, watermarkTimestamp)
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,7 +387,7 @@ class StreamExecution(
lastExecution.executedPlan.collect {
case e: EventTimeWatermarkExec if e.eventTimeStats.value.count > 0 =>
logDebug(s"Observed event time stats: ${e.eventTimeStats.value}")
e.eventTimeStats.value.max - e.delay.milliseconds
math.max(0, e.eventTimeStats.value.max - e.delayMs)
Copy link
Member

@zsxwing zsxwing Dec 20, 2016

Choose a reason for hiding this comment

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

newWatermarkMs can be negative. Image the user is processing some very old data which is before 1970.

Copy link
Contributor Author

@tdas tdas Dec 21, 2016

Choose a reason for hiding this comment

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

i think a lot of things are going to break for that usecase. I dont think our sql functions, Java time format, etc are even designed to handle negative millis. The way I found this issue is that when i tried for convert -ve watermark to formatted string, it gave a very weird date. So I dont think we should add complexity for that use case.
@marmbrus any thoughts?

Copy link
Contributor

@marmbrus marmbrus Dec 21, 2016

Choose a reason for hiding this comment

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

Im not totally sure, but it does seem likely that we don't handle negative dates (and that is probably okay).

Copy link
Member

@zsxwing zsxwing Dec 21, 2016

Choose a reason for hiding this comment

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

It sounds like a SQL bug. IMO, a negative timestamp is valid, it's not a negative date, just a date before 1970.

Copy link
Contributor Author

@tdas tdas Dec 21, 2016

Choose a reason for hiding this comment

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

okay, I think I was doing something to goof up the conversion using SimpleDateFormat. I am fixing that.

}.headOption.foreach { newWatermarkMs =>
if (newWatermarkMs > offsetSeqMetadata.batchWatermarkMs) {
logInfo(s"Updating eventTime watermark to: $newWatermarkMs ms")
Expand Down
71 changes: 57 additions & 14 deletions .../spark/sql/streaming/WatermarkSuite.scala → ...l/streaming/EventTimeWatermarkSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,17 @@ package org.apache.spark.sql.streaming

import java.{util => ju}
import java.text.SimpleDateFormat
import java.util.{Calendar, Date}

import org.scalatest.BeforeAndAfter

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.functions.{count, window}
import org.apache.spark.sql.InternalOutputModes.Complete

class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
class EventTimeWatermarkSuite extends StreamTest with BeforeAndAfter with Logging {

import testImplicits._

Expand All @@ -52,48 +54,59 @@ class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
assert(e.getMessage contains "int")
}


test("event time and watermark metrics") {
val inputData = MemoryStream[Int]
// No event time metrics when there is no watermarking
val inputData1 = MemoryStream[Int]
val aggWithoutWatermark = inputData1.toDF()
.withColumn("eventTime", $"value".cast("timestamp"))
.groupBy(window($"eventTime", "5 seconds") as 'window)
.agg(count("*") as 'count)
.select($"window".getField("start").cast("long").as[Long], $"count".as[Long])

val windowedAggregation = inputData.toDF()
testStream(aggWithoutWatermark, outputMode = Complete)(
AddData(inputData1, 15),
CheckAnswer((15, 1)),
assertEventStats { e => assert(e.isEmpty) },
AddData(inputData1, 10, 12, 14),
CheckAnswer((10, 3), (15, 1)),
assertEventStats { e => assert(e.isEmpty) }
)

// All event time metrics where watermarking is set
val inputData2 = MemoryStream[Int]
val aggWithWatermark = inputData2.toDF()
.withColumn("eventTime", $"value".cast("timestamp"))
.withWatermark("eventTime", "10 seconds")
.groupBy(window($"eventTime", "5 seconds") as 'window)
.agg(count("*") as 'count)
.select($"window".getField("start").cast("long").as[Long], $"count".as[Long])

def assertEventStats(body: ju.Map[String, String] => Unit): AssertOnQuery = AssertOnQuery { q =>
body(q.recentProgress.filter(_.numInputRows > 0).lastOption.get.eventTime)
true
}

testStream(windowedAggregation)(
AddData(inputData, 15),
testStream(aggWithWatermark)(
AddData(inputData2, 15),
CheckAnswer(),
assertEventStats { e =>
assert(e.get("max") === formatTimestamp(15))
assert(e.get("min") === formatTimestamp(15))
assert(e.get("avg") === formatTimestamp(15))
assert(e.get("watermark") === formatTimestamp(0))
},
AddData(inputData, 10, 12, 14),
AddData(inputData2, 10, 12, 14),
CheckAnswer(),
assertEventStats { e =>
assert(e.get("max") === formatTimestamp(14))
assert(e.get("min") === formatTimestamp(10))
assert(e.get("avg") === formatTimestamp(12))
assert(e.get("watermark") === formatTimestamp(5))
},
AddData(inputData, 25),
AddData(inputData2, 25),
CheckAnswer(),
assertEventStats { e =>
assert(e.get("max") === formatTimestamp(25))
assert(e.get("min") === formatTimestamp(25))
assert(e.get("avg") === formatTimestamp(25))
assert(e.get("watermark") === formatTimestamp(5))
},
AddData(inputData, 25),
AddData(inputData2, 25),
CheckAnswer((10, 3)),
assertEventStats { e =>
assert(e.get("max") === formatTimestamp(25))
Expand Down Expand Up @@ -124,6 +137,29 @@ class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
)
}

test("delay in years handled correctly") {
val input = MemoryStream[Long]
val aggWithWatermark = input.toDF()
.withColumn("eventTime", $"value".cast("timestamp"))
.withWatermark("eventTime", "1 month")
Copy link
Contributor

@marmbrus marmbrus Dec 16, 2016

Choose a reason for hiding this comment

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

This seems to be testing months, but the title says years?

Copy link
Contributor Author

@tdas tdas Dec 16, 2016

Choose a reason for hiding this comment

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

yeah. sorry .. i pushed too early. i updated the test now.

Copy link
Contributor

@marmbrus marmbrus Dec 16, 2016

Choose a reason for hiding this comment

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

Just wanted to make sure I wasn't missing something :)

.groupBy(window($"eventTime", "5 seconds") as 'window)
.agg(count("*") as 'count)
.select($"window".getField("start").cast("long").as[Long], $"count".as[Long])

val currentTimeMs = System.currentTimeMillis
val millisPerYear = 1000L * 60 * 60 * 24 * 366 // assume leap year

testStream(aggWithWatermark)(
AddData(input, currentTimeMs / 1000),
CheckAnswer(),
assertEventStats { e =>
assert(timestampFormat.parse(e.get("max")).getTime === (currentTimeMs / 1000) * 1000)
val watermarkTime = timestampFormat.parse(e.get("watermark"))
assert(currentTimeMs - watermarkTime.getTime >= 2 * millisPerYear)
}
)
}

test("recovery") {
val inputData = MemoryStream[Int]
val df = inputData.toDF()
Expand Down Expand Up @@ -231,6 +267,13 @@ class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
)
}

private def assertEventStats(body: ju.Map[String, String] => Unit): AssertOnQuery = {
AssertOnQuery { q =>
body(q.recentProgress.filter(_.numInputRows > 0).lastOption.get.eventTime)
true
}
}

private val timestampFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'") // ISO8601
timestampFormat.setTimeZone(ju.TimeZone.getTimeZone("UTC"))

Expand Down