title: Spark Overlap algorithm using Dataframes
body: <p>Given a data-source with fields: <code>product_id</code> - <code>product</code> - <code>start_time</code> - <code>end_time</code></p>
  
  <p>I am trying to build the logic for catching <strong>Overlapping Records</strong> for the same <code>product</code> (based on <code>start_time</code> and <code>end_time</code>) using the Dataframe functions. </p>
  
  <pre><code>------------------------------------------------
  | product_id | product | start_time | end_time |
  ------------------------------------------------
  |      1     | bottle  |     2      |    4     |
  |      2     | bottle  |     3      |    5     |
  |      3     | bottle  |     2      |    3     |
  |      4     | bottle  |     6      |    7     |
  |      1     |   can   |     2      |    4     |
  |      2     |   can   |     5      |    6     |
  |      3     |   can   |     2      |    4     |
  </code></pre>
  
  <p>I'd like to receive in output</p>
  
  <pre><code>-------------------------------------------------------------------------------------------------
  | product_id_a | product_id_b | product | start_time_a | end_time_a | start_time_b | end_time_b |
  -------------------------------------------------------------------------------------------------
  |       1      |       2      | bottle  |      2       |     4      |      3       |     5      |
  |       1      |       3      | bottle  |      2       |     4      |      2       |     3      |
  </code></pre>
  
  <p>Because <code>bottle_1</code> has overlapping times with <code>bottle_2</code> and <code>bottle_3</code>, where 2 records are overlapping if the following conditions are satisfied:</p>
  
  <ul>
  <li><code>max(a.start_time, b.start_time) &lt; min(a.end_time, b.end_time)</code></li>
  <li><code>!(a.start_time == b.start_time &amp;&amp; a.end_time == b.end_time)</code></li>
  <li><code>a.start_time != b.start_time || a.end_time != b.end_time</code></li>
  </ul>
  
  <p>where the last 2 conditions just specify that I am not interested in cases in which <code>start_time</code> and <code>end_time</code> are equal (e.g. <code>can_1</code> and <code>can_3</code> are not in the expected result even if they have same <code>start_time</code> and <code>end_time</code>).</p>
  
  <p>For how the problem is structured is easy to think to a solution with MapReduce using RDD, but I am interested to a <strong>solution with Dataframes</strong>.</p>
  
  <p><strong>HINT</strong>: is there any chance that with <code>groupBy().agg()</code> is possible to specify an interesting condition that reaches the described logic?</p>
  
  <p>For any further explanation feel free to ask</p>
  
  <p><strong>NOT DUPLICATE</strong> of <a href="https://stackoverflow.com/questions/41711716/how-to-aggregate-over-rolling-time-window-with-groups-in-spark">How to aggregate over rolling time window with groups in Spark</a></p>
  
  <p>Unfortunately in the reported answer has been used <code>F.lag</code>, which in my case is not a good enough condition: F.lag uses a comparison with the previous record only, but in the reported example would not work as desired since that <code>bottle_1</code> would not be reported as overlapping with <code>bottle_3</code> because they are not consecutive records </p>