使用 Spark Dataframes/Datasets 的动态 GroupBy 和计数答案

【问题标题】：Dynamic GroupBy and count using Spark Dataframes/Datasets使用 Spark Dataframes/Datasets 的动态 GroupBy 和计数
【发布时间】：2020-02-27 10:01:47
【问题描述】：

用例是按给定数据集中的每一列分组，并获取该列的计数。结果集是 (key, value) 映射，最后是它们的集合。例如 students = {(age, firstname, lastname)(12, "FN", "LN"), (13, "df", "gh")} groupby age => (12, 1), (13, 1) groupby firstname => etc

我知道蛮力方法是做一个地图并为每列维护一个地图计数但我想看看我们是否可以用 foldLeft 和 windows 功能做更多的事情。我尝试使用汇总和多维数据集，但这确实将所有列组合在一起而不是单独的

【问题讨论】：

能否分享示例数据

标签： scala apache-spark apache-spark-sql dataset rdd

【解决方案1】：

假设您需要分组数据帧的联合，我能够解决如下：

代码

val df = Seq(("12", "FN", "LN"),
      ("13", "FN", "gh")).toDF("age", "firstname", "lastname")
df.show(false)

val initialDF = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], StructType(
      Seq(StructField("column", StringType), StructField("count", IntegerType))
))

df.columns.foldLeft(initialDF)((df1, column) => df1.union(df.groupBy(column).count())).show(false)

输出

INPUT DF:
+---+---------+--------+
|age|firstname|lastname|
+---+---------+--------+
|12 |FN       |LN      |
|13 |FN       |gh      |
+---+---------+--------+

OUTPUT DF:
+------+-----+
|column|count|
+------+-----+
|12    |1    |
|13    |1    |
|FN    |2    |
|gh    |1    |
|LN    |1    |
+------+-----+

【讨论】：

是的！那绝对是我想要的。我尝试了 foldleft，但我认为它用错了。将尝试一下，看看它在正常计数和维护地图上的表现如何。老实说，我不太热衷于后者，所以寻找简洁的东西。谢谢！
@Patssay- 如果您完成了测试，请接受答案，或者如果您需要任何帮助，请告诉我。

【解决方案2】：

假设您需要 Key、Value、Grouping Column name 作为输出中的三列，则必须使用以下代码才能理解键和分组列的关系。

代码

val df = Seq(("12", "FN", "LN"),
    ("13", "FN", "gh")).toDF("age", "firstname", "lastname")
df.show(false)

val initialDF = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], StructType(
    Seq(StructField("Key", StringType), StructField("Value", IntegerType), 
    StructField("GroupColumn", StringType))
))

val resultantDf = df.columns.foldLeft(initialDF)((df1, column) => df1.union(
      df.groupBy(column).count().withColumn("GroupColumn", lit(column))
    ))
resultantDf.show(false)

resultantDf.collect().map { row =>
      (row.getString(0), row.getLong(1))
}.foreach(println)

输出

INPUT DF:
+---+---------+--------+
|age|firstname|lastname|
+---+---------+--------+
|12 |FN       |LN      |
|13 |FN       |gh      |
+---+---------+--------+

OUTPUT DF:
+---+-----+-----------+
|Key|Value|GroupColumn|
+---+-----+-----------+
|12 |1    |age        |
|13 |1    |age        |
|FN |2    |firstname  |
|gh |1    |lastname   |
|LN |1    |lastname   |
+---+-----+-----------+

OUTPUT LIST:
(12,1)
(13,1)
(FN,2)
(gh,1)
(LN,1)

【讨论】：