Spark有效地找到了一组列的最常见值

Question

我想知道是否有一种更有效的方法来找到一组列的最常见值，而不是使用 rank()来将其用作缺失值的插补。p>例如。在Spark-Sql中，我可以制定类似的东西如何选择最经常出现的值？每列。该解决方案适用于使用等级的单列。我正在寻找的是a）更有效的变体（作为第一个答案）和b）比使用for a loop和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)

是结果，但对于列栏 ->首先，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
)

您可以调整分组列或密钥功能以处理更复杂的方案。

Answer 2

 // find most frequent value in the grouped columns
  def mode(df: DataFrame, valueColumnName: String, groupByColumns: Seq[String]): DataFrame = {
    df.groupBy(valueColumnName, groupByColumns: _*).count()
      .withColumn(
        "rn",
        row_number().over(Window.partitionBy(groupByColumns.head, groupByColumns.tail: _*).orderBy(col("count").desc))
      )
      .where(col("rn") === 1)
      .select(valueColumnName, groupByColumns: _*)
  }