title: How to filter array of objects by timestamp attribute (code alternative to SQL lag over partition by - command)
body: <p>So I'm writing the code for apache spark which executes on the java virtual machine. It has Spark SQL which let the user write SQL expressions on collections of objects. </p>
  
  <p>SQL command I am trying to query is too complicated for spark's engine so I am trying to find alternative using code not SQL.</p>
  
  <p>I am writing a script to check if my data streamers streamed every 12 hours </p>
  
  <p>So example data looks like this:</p>
  
  <pre><code>sourceId          streamedAt
  ----------------------------------
  station_1      2019-06-02 09:01:10
  station_1      2019-06-02 09:01:13
  station_1      2019-06-02 21:05:10
  station_1      2019-06-02 21:08:15
  station_2      2019-06-02 09:30:10
  station_2      2019-06-02 09:31:10
  station_2      2019-06-02 21:40:11
  station_2      2019-06-02 21:40:12
  </code></pre>
  
  <p>You can think of this as an array of StreamerEvent objects:</p>
  
  <pre><code>class StreamerEvent {
      String sourceId;
      Date streamedAt;
  }
  </code></pre>
  
  <p>I need to search through the collection and for each streamer and stream which will last for a few minutes(this happens every 12 hours +- some time) get only the first one out of that particular streaming interval. So my array now looks like this:</p>
  
  <pre><code>sourceId          streamedAt
  ----------------------------------
  station_1      2019-06-02 09:01:10
  station_1      2019-06-02 21:05:10
  station_2      2019-06-02 09:30:10
  station_2      2019-06-02 21:40:11
  </code></pre>
  
  <p>So my SQL query for this was:</p>
  
  <pre><code>spark.sql("SELECT * FROM (select sourceId, streamedAt, lag(streamedAt, 1) over(partition by sourceId order by streamedAt) as prev_streamedAt from streaming_data) streaming_data where prev_streamedAt is null or prev_streamedAt &lt; streamedAt - interval 11 hour")
  </code></pre>
  
  <p>But it doesn't work on streaming data, only on static.</p>
  
  <p>Can somebody give some hint on how to approach this using java/scala collections?</p>