Technical Depth Verification Questions

import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout}
case class Event(userId: String, ts: Long, value: Double)
case class Bucket(tsSec: Long, sum: Double, count: Long)
case class UserState(buckets: Map[Long, (Double, Long)]) // tsSec -> (sum,count)

def updateUser(userId: String, events: Iterator[Event], state: GroupState[UserState]) = {
  val allowedLatenessMs = 2*60*1000L   // e.g. 2 min watermark
  val windowMs = 60*1000L
  val nowWatermark = state.getCurrentWatermarkMs() // conceptual; Spark uses watermark externally

  val s = state.getOption.getOrElse(UserState(Map.empty))
  var buckets = s.buckets

  // ingest events into per-second buckets
  events.foreach { e =>
    val sec = e.ts / 1000
    val (sum,cnt) = buckets.getOrElse(sec, (0.0,0L))
    buckets = buckets.updated(sec, (sum + e.value, cnt + 1))
  }

  // evict buckets older than watermark or older than window
  val minAllowedSec = ((nowWatermark - windowMs) / 1000).max(0)
  buckets = buckets.filter { case (sec,_) => sec >= minAllowedSec }

  // compute rolling 1-minute avg
  val (totalSum, totalCount) = buckets.values.foldLeft((0.0,0L)){case ((s,c),(ss,cc)) => (s+ss,c+cc)}
  val avg = if (totalCount>0) totalSum/totalCount else null

  if (buckets.nonEmpty) {
    state.update(UserState(buckets))
    state.setTimeoutTimestamp((minAllowedSec + windowMs/1000 + allowedLatenessMs/1000)*1000) // schedule cleanup
  } else {
    state.remove()
  }

  Iterator((userId, avg, /*output event-time = current watermark or max ts in buckets*/))
}

val ds = spark.readStream
  .format("kafka")... // parse Event with timestamp column "ts"
  .withWatermark("ts", "2 minutes")
  .as[Event]
  .groupByKey(_.userId)
  .flatMapGroupsWithState(OutputMode.Update(), GroupStateTimeout.EventTimeTimeout())(updateUser)