title: spark windowing function VS group by performance issue
body: <p>I have a dataframe done like</p>
  
  <pre><code>| id | date      |  KPI_1 | ... | KPI_n
  | 1  |2012-12-12 |   0.1  | ... |  0.5
  | 2  |2012-12-12 |   0.2  | ... |  0.4
  | 3  |2012-12-12 |   0.66 | ... |  0.66 
  | 1  |2012-12-13 |   0.2  | ... |  0.46
  | 4  |2012-12-14 |   0.2  | ... |  0.45 
  | ...
  | 55| 2013-03-15 |  0.5  | ... |  0.55
  </code></pre>
  
  <p>we have </p>
  
  <ul>
  <li>X identifiers</li>
  <li>a row for every identifier for a given date</li>
  <li>n KPIs</li>
  </ul>
  
  <p>I have to calculate some derived KPI for every row, and this KPI depends on the previous values of every ID.
  Let's say my derived KPI is a diff, it would be:</p>
  
  <pre><code>| id | date      |  KPI_1 | ... | KPI_n | KPI_1_diff | KPI_n_diff
  | 1  |2012-12-12 |   0.1  | ... |  0.5  |   0.1      | 0.5
  | 2  |2012-12-12 |   0.2  | ... |  0.4  |   0.2      |0.4
  | 3  |2012-12-12 |   0.66 | ... |  0.66 |   0.66     | 0.66 
  | 1  |2012-12-13 |   0.2  | ... |  0.46 |   0.2-0.1  | 0.46 - 0.66
  | 4  |2012-12-13 |   0.2  | ... |  0.45  ...
  | ...
  | 55| 2013-03-15 |  0.5  | ... |  0.55
  </code></pre>
  
  <p>Now: what I would do is:</p>
  
  <pre><code>val groupedDF = myDF.groupBy("id").agg(
      collect_list(struct(col("date",col("KPI_1"))).as("wrapped_KPI_1"),
      collect_list(struct(col("date",col("KPI_2"))).as("wrapped_KPI_2")
      // up until nth KPI
  )
  </code></pre>
  
  <p>I would get aggregated data such as:</p>
  
  <pre><code>[("2012-12-12",0.1),("2012-12-12",0.2) ...
  </code></pre>
  
  <p>Then I would sort these wrapped data, unwrap and map over these aggregated result with some UDF and produce the output (compute diffs and other statistics).</p>
  
  <p><strong>Another approach</strong> is to use the window functions such as:</p>
  
  <pre><code>val window = Window.partitionBy(col("id")).orderBy(col("date")).rowsBetween(Window.unboundedPreceding,0L) 
  </code></pre>
  
  <p>and do :</p>
  
  <pre><code>val windowedDF = df.select (
    col("id"),
    col("date"),
    col("KPI_1"),
    collect_list(struct(col("date"),col("KPI_1"))).over(window),  
    collect_list(struct(col("date"),col("KPI_2"))).over(window)
     )   
  </code></pre>
  
  <p>This way I get:</p>
  
  <pre><code>[("2012-12-12",0.1)]
  [("2012-12-12",0.1), ("2012-12-13",0.1)]
  ...
  </code></pre>
  
  <p>That look nicer to process, but I suspect that repeating the window would produce unnecessary grouping and sorting for every KPI.</p>
  
  <p>So here are the questions:</p>
  
  <ol>
  <li>I'd rather go for the grouping approach?</li>
  <li>Would I go for the window? If so what is the most efficient approach to do it?</li>
  </ol>