如何使用mapGroups在scala spark中的groupby之后计算列中的不同值答案

【问题标题】：how to count distinct values in a column after groupby in scala spark using mapGroups如何使用mapGroups在scala spark中的groupby之后计算列中的不同值
【发布时间】：2018-10-02 18:31:52
【问题描述】：

我是 scala spark 的新手。我有一个文本文件数据作为

001,delhi,india
002,chennai,india
003,hyderabad,india
004,newyork,us
005,chicago,us
006,lasvegas,us
007,seattle,us

我想计算每个国家/地区不同城市的数量，所以我应用了 groupBy 和 mapGroups。我不确定如何计算 mapGroups 中的值。请在下面找到我的代码

val ds1 = sparkSession.read.textFile("samplefile.txt").map(x => x.split(","))
  val ds2 = ds1.groupByKey(x => x(2)).mapGroups{case(k,iter) => (k,iter.map(x => x(2)).toArray)}

请帮助我了解语法。我知道可以通过 spark-sql 轻松完成，但我想通过 scala 来完成

【问题讨论】：

如果您想按城市计数，只需将toArray 替换为size。
@leo 如果我的城市不是不同的，并且我想计算每个国家/地区的不同城市数量，那么语法会是什么？？
@LeoC 请尝试回答我上面发布的问题
请看我发布的答案。
@LeoC 假设我有一个额外的列（未在我的示例文件中给出），我想按国家/地区计算平均分组，那么语法应该是什么。提前致谢

标签： scala apache-spark

【解决方案1】：

要计算每个国家/地区的不同城市，您可以将按国家/地区列表映射到city 的数组并计算不同城市的数量：

val ds1 = spark.read.textFile("/path/to/textfile").map(_.split(","))
val ds2 = ds1.
  groupByKey(_(2)).mapGroups{ case (k, iter) =>
    (k, iter.map(_(1)).toList.distinct.size)
  }

[更新]

要计算每个国家/地区的平均值，例如，从第 4 个数字列开始，可以使用以下方法：

val ds3 = ds1.
  groupByKey(_(2)).mapGroups{ case (k, iter) =>
    val numList = iter.map(_(3).toDouble).toList
    (k, numList.sum / numList.size)
  }

如果您需要各种类型的数字聚合，我认为 Spark DataFrame API 会是更高效的工具（例如，它内置了 avg()）。

【讨论】：

【解决方案2】：

正确的方法是像 df 成为您的源数据框，

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

val df: DataFrame = ???

val result = df.groupBy("country col name").agg(countDistinct("city column name").alias("city_count"))

希望这会有所帮助。

【讨论】：

它只计算城市。不同的城市
这是scala，使用dataframe api。 spark-sql 类似于spark.sql("query here")