title: Sliding window in Spark Structured Streaming
body: <p>I have a stream of data coming from IoT devices that have an id (uuid) and a quantity (ie. temperature).</p>
  
  <p>I'd like to keep a count of events received in the last 15 minutes, with a sliding window of, say, 1 or 5 minutes.</p>
  
  <p>I have implemented the following in Spark but it generates all windows, but I'm interested only in the most recent one (and maybe zero if the device has not sent any data in the meantime):</p>
  
  <pre><code>import org.apache.spark.sql.functions._
  
  val agg15min = stream
    .withWatermark("createdAtTimestamp", "15 minutes")
    .where("device_uuid is not null")
    .groupBy($"device_uuid", window($"createdAtTimestamp", "15 minutes", "5 minutes"))
    .count()
  </code></pre>
  
  <p>I have tried filtering the data afterwards like this:</p>
  
  <pre><code>val query15min =
    agg15min
      .writeStream
      .format("memory")
      .queryName("query15min")
      .outputMode("complete")
      .start()
  </code></pre>
  
  <p>and then:</p>
  
  <pre><code>val df15min = spark.sql("""
  with cte as (
  select
      device_uuid,
      date_format(window.end, "MMM-dd HH:mm") as time,
      rank() over (partition by device_uuid order by window.end desc) as rank,
      count
    from query15min
  )
  select
    device_uuid,
    count
  from cte
  where rank = 1""")
  </code></pre>
  
  <p>But the documentation says <code>memory</code> is not for production use, and also it is rather inefficient.</p>
  
  <p>Is there an efficient way to implement this kind of logic in Spark Structured Streaming?</p>