Динамический GroupBy и подсчет с использованием Spark Dataframes / Datasets - PullRequest
Новая
       43

Динамический GroupBy и подсчет с использованием Spark Dataframes / Datasets

0 голосов
/ 01 ноября 2019

Вариант использования - группировать по каждому столбцу в данном наборе данных и получать счетчик этого столбца. Результирующий набор представляет собой (ключ, значение) карту и, наконец, объединяет их все. Например, students = {(age, firstname, lastname)(12, "FN", "LN"), (13, "df", "gh")} groupby age => (12, 1), (13, 1) groupby firstname => etc

Я знаю, что метод грубой силы состоит в том, чтобы сделать карту и вести карту для подсчета для каждого столбца, но я хотел посмотреть, есть ли что-то еще, что мы можем сделатьможет быть, foldLeft и функция Windows. Я попытался использовать свертку и куб, но это объединяет все столбцы вместе, а не по отдельности

Ответы [ 2 ]

1 голос
/ 01 ноября 2019

Предполагая, что вам необходимо указать ключ, значение, имя столбца группировки в качестве трех столбцов в выходных данных, вам придется использовать приведенный ниже код, чтобы можно было понять взаимосвязи ключа и столбца группировки.

Код

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)
1 голос
/ 01 ноября 2019

Предполагая, что вам требуется объединение сгруппированных кадров данных, я смог решить его следующим образом:

Код 

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    |
+------+-----+
...