аюсь агрегировать количество записей каждые 10 секунд, используя структурированную потоковую передачу для следующих входящих данных Кафки

{ 
  "ts2" : "2018/05/01 00:02:50.041", 
  "serviceGroupId" : "123", 
  "userId" : "avv-0", 
  "stream" : "", 
  "lastUserActivity" : "00:02:50", 
  "lastUserActivityCount" : "0" 
} 
{ 
  "ts2" : "2018/05/01 00:09:02.079", 
  "serviceGroupId" : "123", 
  "userId" : "avv-0", 
  "stream" : "", 
  "lastUserActivity" : "00:09:02", 
  "lastUserActivityCount" : "0" 
} 
{ 
  "ts2" : "2018/05/01 00:09:02.086", 
  "serviceGroupId" : "123", 
  "userId" : "avv-2", 
  "stream" : "", 
  "lastUserActivity" : "00:09:02", 
  "lastUserActivityCount" : "0" 
} 

Со следующей логикой

val sdvTuneInsAgg1 = df 
  .withWatermark("ts2", "10 seconds") 
  .groupBy(window(col("ts2"),"10 seconds")) 
  .agg(count("*") as "count") 
  .as[CountMetric1] 

val query1 = sdvTuneInsAgg1.writeStream
      .format("console")
      .foreach(writer)
      .start()

и я не вижу никаких записей внутри писателя. Но единственная аномалия в том, что текущая дата - 2018/05/24, но запись, которую я обрабатываю (ts2), имеет старые даты. Будет ли агрегация / подсчет работать в этом сценарии?