title: Efficient spark dataset operations when partitioned by overlapping columns
body: <p>I have a dataset ("guid", "timestamp", "agt") like below</p>
  
  <pre><code>val df = List(Test("a", "1", null),
     Test("b", "2", "4"),
     Test("a", "1", "3"),
     Test("b", "2", "4"),
     Test("c", "1", "3"),
     Test("a", "6", "8"),
     Test("b", "2", "4"),
     Test("a", "1", "4")
  </code></pre>
  
  <p>I need to compute </p>
  
  <ul>
  <li>the minimum timestamp for each row when grouped by guid.   </li>
  <li>The count for each key when grouped by (guid, timestamp)  </li>
  <li>The agtM of row when grouped by guid and ordered by timestamp(desc) and then take first non empty agt else "" </li>
  <li>Drop duplicates</li>
  </ul>
  
  <p>So the output will be like below.</p>
  
  <pre><code>+----+---------+---+-------+-----+----+
  |guid|timestamp|agt|minimum|count|agtM|
  +----+---------+---+-------+-----+----+
  |   c|        1|  3|      1|    1|   3|
  |   b|        2|  4|      2|    3|   4|
  |   a|        1|   |      1|    3|   8|
  |   a|        6|  8|      1|    1|   8|
  +----+---------+---+-------+-----+----+
  </code></pre>
  
  <p>I have tried </p>
  
  <pre><code>val w = Window.partitionBy($"guid")
  
      val w1 = Window.partitionBy($"guid", $"timestamp")
      val w2 = Window.partitionBy($"guid").orderBy($"timestamp".desc).rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
  
      val gg = df.toDS()
        .withColumn("minimum", min("timestamp").over(w))
        .withColumn("count", count("*").over(w1))
        .withColumn("agtM", coalesce(first($"agt", true).over(w2), lit("")))
        .dropDuplicates("guid", "timestamp")
  </code></pre>
  
  <p>The agtM calculation I am not so confident though. My target is to achieve minimum shuffling as in this scenario we first group by guid and then group by (guid, timestamp) and logically the second grouping should happen in the first created partition. the output is then grouped by guid and joined with another table. Both data are pretty huge (in TBs) so wanted to achieve this with minimum shuffling and didn't want to move the computation inside mapGroups later (I could have done the agtM calculation simply by filtering the group with non-empty agenttime and then maxBy timestamp). Can you please suggest the the best way to achieve the above?</p>
  
  <p><strong>EDIT</strong></p>
  
  <p>The agtM calculation has been fixed. Just to give more context for the operations ahead, union of output and another dataset(one extra field, we kept it dummy in the output) will then need to be grouped by key to produce final results. I was also thinking of calculating these values(except window w) inside each partition (mapPartitions), then take the list inside each partition as another list and do further calculation. </p>