这里有一个稍微冗长的方法。你可以 map 每列到一个具有一行的Dataframe，该行对应于所需输出中的行。必要时添加na列。将列名转换为所需的列名，最后执行 unionAll 合并Dataframe（每行一个）。

import org.apache.spark.sql.expressions.Window

val top = 2

val result = ds.columns.map(
    c => ds.groupBy(c).count()
           .withColumn("rn", row_number().over(Window.orderBy(desc("count"))))
           .filter(s"rn <= $top")
           .groupBy().pivot("rn")
           .agg(first(col(c)), first(col("count")))
           .select(lit(c), col("*"))
).map(df => 
    if (df.columns.size != 1 + top*2)
        df.select(List(col("*")) ::: (1 to (top*2+1 - df.columns.size)).toList.map(x => lit("NA")): _*)
    else df
).map(df =>
    df.toDF(List("feature") ::: (1 to top).toList.flatMap(x => Seq(s"top$x name", s"top$x value")): _*)
).reduce(_ unionAll _)

result.show
+--------+---------+----------+---------+----------+
| feature|top1 name|top1 value|top2 name|top2 value|
+--------+---------+----------+---------+----------+
|emp_name|      ram|         3|     sham|         2|
|emp_area|      USA|         4|   Canada|         1|
|    dept|    Sales|         3|Marketing|         2|
|     zip|   805912|         5|       NA|        NA|
+--------+---------+----------+---------+----------+