根据迭代逻辑，有几种可能的方法：

使用数据集api

鉴于

val df =
      Seq(("a", 0, "foo"), ("b", 1, "foo"), ("a", 1, "foobar"))
        .toDF("A", "B", "C")

先预处理一下

df.select($"A", struct($"B", $"C") as $"S").show()

得到

+---+-----------+
|  A|          S|
+---+-----------+
|  a|   [0, foo]|
|  b|   [1, foo]|
|  a|[1, foobar]|
+---+-----------+

现在我们可以将任何scala代码应用于元组序列，包括排序：

df.select($"A", struct($"B", $"C") as $"S")
      .groupBy("A")
      .agg(collect_list("S"))
      .as[(String, Seq[(Int, String)])]
      .map {
        case (a, l) => (a, l.sortBy(_._1).map(_._2).maxBy(_.length))
      }
      .show()

使用udafs

实现自定义udaf：

class MyAgg extends Aggregator[
      (Int, String),
      mutable.ListBuffer[(Int, String)],
      /* any output type here */] {
...

并用它进行聚合：

val myagg = udaf(new MyAgg())
df.select($"A", struct($"B", $"C") as "S").groupBy($"A").agg(myagg($"S"))