spark_trigger.scala

def updateFunction(alertStrings: Seq[String], stateOption: Option[(Boolean, Date, String)]): Option[(Boolean, Date, String)] = {
  (alertStrings, stateOption) match {
    case (_, Some((_, triggerTime, string))) => {
      val cal = Calendar.getInstance()
      cal.add(Calendar.Minute, -60)
      if (triggerTime.before(cal.getTime)) None else Some((false, triggerTime, string))
    }
    case (Nil, None) => None
    case (Seq(string, _*), None) => Some((true, Calendar.getInstance().getTime(), string))
  }
}
val alertState =
  alertStream
    .updateStateByKey(updateFunction)
    .filter {
      case (_, (bool, _, _)) => bool
    }.map {
      case (_, (_, _, string)) => string
    }

alertState.print