ntento agregar el recuento de registros cada 10 segundos utilizando la transmisión estructurada para los siguientes datos entrantes de Kafk

{ 
  "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" 
} 

Con la siguiente lógica

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()

y no veo ningún registro dentro del escritor. Pero, la única anomalía es que la fecha actual es 2018/05/24 pero el registro que estoy procesando (ts2) tiene fechas antiguas. ¿La agregación / conteo funcionará en este escenario?