apache-spark - spark 有效地找到一组列的最频繁值

Question

我想知道在 spark 中是否有比使用 rank() 更有效的方法来查找一组列的最频繁值，以便将其用作缺失值的插补。

例如在 spark-sql 中我可以制定类似的东西 how to select the most frequently appearing values? 每列。 此解决方案适用于使用排名的单个列。我正在寻找的是 a) 更有效的变体(如第一个答案概述)和 b) 比使用 for 循环和 a) 的解决方案更优化的东西来申请多列。

您认为有可能在 spark 中对此进行优化吗？

编辑

一个例子。这是一个小数据集

case class FooBarGG(foo: Int, bar: String, baz: String, dropme: String)
val df = Seq((0, "first", "A", "dropme"), (1, "second", "A", "dropme2"),
    (0, "first", "B", "foo"),
    (1, "first", "C", "foo"))
    .toDF("foo", "bar", "baz", "dropme").as[FooBarGG]
val columnsFactor = Seq("bar", "baz")
val columnsToDrop = Seq("dropme")
val factorCol= (columnsFactor ++ columnsToDrop).map(c => col(c))

从答案中查询

df.groupBy(factorCol: _*).count.agg(max(struct($"count" +: factorCol: _*)).alias("mostFrequent")).show
+--------------------+
|        mostFrequent|
+--------------------+
|[1,second,A,dropme2]|
+--------------------+
|-- mostFrequent: struct (nullable = true)
 |    |-- count: long (nullable = false)
 |    |-- bar: string (nullable = true)
 |    |-- baz: string (nullable = true)
 |    |-- dropme: string (nullable = true)

结果除了column bar -> first, baz -> A 和drompe -> foo 是单个top1最频繁的值，与返回结果不同。

Answer 1

只要您的字段可以排序并且计数是领先的，您就可以使用简单聚合:

import org.apache.spark.sql.functions._

val df = Seq("John", "Jane", "Eve", "Joe", "Eve").toDF("name")
val grouping = Seq($"name")

df.groupBy(grouping: _*).count.agg(max(struct($"count" +: grouping: _*)))

也可以使用静态类型的 Dataset:

import org.apache.spark.sql.catalyst.encoders.RowEncoder

df.groupByKey(x => x)(RowEncoder(df.schema)).count.reduce(
  (x, y) => if (x._2 > y._2) x else y
)

您可以调整分组列或键函数以处理更复杂的场景。