title: ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast
body: <p>I am using an Aggregator to apply some custom merge on a DataFrame after grouping its records by their primary key:</p>
  <pre><code>case class Player(
    pk: String, 
    ts: String, 
    first_name: String, 
    date_of_birth: String
  )
  
  case class PlayerProcessed(
    var ts: String, 
    var first_name: String, 
    var date_of_birth: String
  )
  
  // Cutomer Aggregator -This just for the example, actual one is more complex
  object BatchDedupe extends Aggregator[Player, PlayerProcessed, PlayerProcessed] {
  
    def zero: PlayerProcessed = PlayerProcessed(&quot;0&quot;, null, null)
  
    def reduce(bf: PlayerProcessed, in : Player): PlayerProcessed = {
      bf.ts = in.ts
      bf.first_name = in.first_name
      bf.date_of_birth = in.date_of_birth
      bf
    }
  
    def merge(bf1: PlayerProcessed, bf2: PlayerProcessed): PlayerProcessed = {
      bf1.ts = bf2.ts
      bf1.first_name = bf2.first_name
      bf1.date_of_birth = bf2.date_of_birth
      bf1
    }
  
    def finish(reduction: PlayerProcessed): PlayerProcessed = reduction
    def bufferEncoder: Encoder[PlayerProcessed] = Encoders.product
    def outputEncoder: Encoder[PlayerProcessed] = Encoders.product
  }
  
  
  val ply1 = Player(&quot;12121212121212&quot;, &quot;10000001&quot;, &quot;Rogger&quot;, &quot;1980-01-02&quot;)
  val ply2 = Player(&quot;12121212121212&quot;, &quot;10000002&quot;, &quot;Rogg&quot;, null)
  val ply3 = Player(&quot;12121212121212&quot;, &quot;10000004&quot;, null, &quot;1985-01-02&quot;)
  val ply4 = Player(&quot;12121212121212&quot;, &quot;10000003&quot;, &quot;Roggelio&quot;, &quot;1982-01-02&quot;)
  
  val seq_users = sc.parallelize(Seq(ply1, ply2, ply3, ply4)).toDF.as[Player]
  
  val grouped = seq_users.groupByKey(_.pk)
  
  val non_sorted = grouped.agg(BatchDedupe.toColumn.name(&quot;deduped&quot;))
  non_sorted.show(false)
  </code></pre>
  <p>This returns:</p>
  <pre><code>+--------------+--------------------------------+
  |key           |deduped                         |
  +--------------+--------------------------------+
  |12121212121212|{10000003, Roggelio, 1982-01-02}|
  +--------------+--------------------------------+
  </code></pre>
  <p>Now, I would like to order the records based on <code>ts</code> before aggregating them. From <a href="https://stackoverflow.com/q/39505599/1773841">here</a> I understand that <code>.sortBy(&quot;ts&quot;)</code> do not guarantee the order <strong>after</strong> the <code>.groupByKey(_.pk)</code>. So I was trying to apply the <code>.sortBy</code> between the <code>.groupByKey</code> and the <code>.agg</code></p>
  <p>The output of the <code>.groupByKey(_.pk)</code> is a <code>KeyValueGroupedDataset[String,Player]</code>, being the second element an <code>Iterator</code>. So to apply some sorting logic there I convert it into a <code>Seq</code>:</p>
  <pre><code>val sorted = grouped.mapGroups{case(k, iter) =&gt; (k, iter.toSeq.sortBy(_.ts))}.agg(BatchDedupe.toColumn.name(&quot;deduped&quot;))
  sorted.show(false)
  </code></pre>
  <p>However, the output of <code>.mapGroups</code> after adding the sorting logic is a <code>Dataset[(String, Seq[Player])]</code>. So when I try to invoke the <code>.agg</code> function on it I am getting the following exception:</p>
  <pre><code>Caused by: ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to $line050e0d37885948cd91f7f7dd9e3b4da9311.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$Player
  </code></pre>
  <p>How could I convert back the output of my <code>.mapGroups(...)</code> into a <code>KeyValueGroupedDataset[String,Player]</code>?</p>
  <p>I tried to cast back to Iterator as follows:</p>
  <pre><code>val sorted = grouped.mapGroups{case(k, iter) =&gt; (k, iter.toSeq.sortBy(_.ts).toIterator)}.agg(BatchDedupe.toColumn.name(&quot;deduped&quot;))
  </code></pre>
  <p>But this approach produced the following exception:</p>
  <pre><code>UnsupportedOperationException: No Encoder found for Iterator[Player]
  - field (class: &quot;scala.collection.Iterator&quot;, name: &quot;_2&quot;)
  - root class: &quot;scala.Tuple2&quot;
  </code></pre>
  <p>How else can I add the sort logic between the <code>.groupByKey</code> and <code>.agg</code> methods?</p>