Эквивалент sql запроса для потоковой передачи withWatermark Spark

Question

У меня есть метод, который принимает искровой sql запрос в качестве аргумента, запускает его в потоковом наборе данных, мне нужно обрабатывать оконную функцию и withWatermark. функция окна кажется возможной, но я не могу найти эквивалентный синтаксис sql для водяного знака.

  val levels = spark.sparkContext.parallelize(Seq(
  // (year, month, dayOfMonth, hour, minute, second)
  ((2012, 12, 12, 12, 12, 12), 5),
  ((2012, 12, 12, 12, 12, 14), 9),
  ((2012, 12, 12, 13, 13, 14), 4),
  ((2016, 8, 13, 0, 0, 0), 10),
  ((2017, 5, 27, 0, 0, 0), 15))).
  map { case ((yy, mm, dd, h, m, s), a) => (LocalDateTime.of(yy, mm, dd, h, m, s), a) }.
  map { case (ts, a) => (Timestamp.valueOf(ts), a) }.
  toDF("time", "level")

  levels.createGlobalTempView("test")

  val apiResult = levels.select(window($"time", "5 seconds","2 seconds"), $"level")
  //same window function using sql 
  val sqlResult=  spark.sql("select window(time, '5 seconds','2 seconds'), level from  global_temp.test");

  apiResult.show(false)
  sqlResult.show(false) // both apiResult, sqlResult produces same result

  //need equivalent sql for below watermark-window function 
  val windowedCountsApi = levels
  .withWatermark("time", "10 seconds")
  .groupBy(window($"time", "5 minutes", "2 minutes"), $"level")
  .count()

  windowedCountsApi.show(truncate = false)

для простоты я тестировал с обычным набором данных, ваша помощь очень ценится