SparkReadStream.scala

spark.readStream
  .format("kafka")
  .option("subscribe", "input")
  .load()
  .withWatermark(eventTime="timestampColumn", delayThreshold="10 seconds")
  .groupBy('value.cast("string") as 'key)
  .groupBy(window(col("time"), "1 day").as("time")) // tumbling window
  .groupBy(window(col("time"), "10 minutes", "5 minutes").as("time")) // sliding window
  .groupBy( 'keyCol, window("timestamp","10 mins"))
  .agg(count("*") as 'value)
  .writeStream
  .format("kafka")
  .option("topic", "output")
  .trigger("1 minute")
  .outputMode("update")
  .option("checkpointLocation", "/cp/")
  .start()
  .awaitTermination()