title: Spark Dataframes- Reducing By Key
body: <p>Let's say I have a data structure like this where ts is some timestamp</p>
  
  <pre><code>case class Record(ts: Long, id: Int, value: Int)
  </code></pre>
  
  <p>Given a large number of these records I want to end up with the record with the highest timestamp for each id.  Using the RDD api I think the following code gets the job done:</p>
  
  <pre><code>def findLatest(records: RDD[Record])(implicit spark: SparkSession) = {
    records.keyBy(_.id).reduceByKey{
      (x, y) =&gt; if(x.ts &gt; y.ts) x else y
    }.values
  }
  </code></pre>
  
  <p>Likewise this is my attempt with datasets:</p>
  
  <pre><code>def findLatest(records: Dataset[Record])(implicit spark: SparkSession) = {
    records.groupByKey(_.id).mapGroups{
      case(id, records) =&gt; {
        records.reduceLeft((x,y) =&gt; if (x.ts &gt; y.ts) x else y)
      }
    }
  }
  </code></pre>
  
  <p>I've being trying to work out how to achieve something similar with dataframes but to no avail- I realise I can do the grouping with:</p>
  
  <pre><code>records.groupBy($"id")
  </code></pre>
  
  <p>But that gives me a RelationGroupedDataSet and it's not clear to me what aggregation function I need to write to achieve what I want- all example aggregations I've seen appear to focus on returning just a single column being aggregated rather than the whole row.</p>
  
  <p>Is it possible to achieve this using dataframes?</p>