在保留多个值的同时计数键的出现次数

Question

我在计算一个键的出现次数时遇到了一些问题，同时也保留了几个值。

通常我只会：

val a = file1.map(x => (x, 1)).reduceByKey(_ + _)

它给出了每个关键点的出现次数。

但是，我还希望在计算键的出现次数的同时，保留键每次出现的值。类似这样：

val a = file1.map(x => (x(1), (x(2), 1)).reduceByKey{case (x,y) => (x._1, y._1, x._2+y._2)}

例如：如果键x（1）是一个国家，而x（2）是一个城市，那么我想保留一个国家的所有城市，并知道一个国家有多少个城市。

Answer 1

我建议您使用dataframes以及dataframes比rdds优化且易于使用。

但是，如果您想了解reduceByKey的功能（即按照您所说的城市信息保存其他信息），那么您可以执行以下操作

假设您有一个rdd

val rdd = sc.parallelize(Seq(
  ("country1", "city1"),
  ("country1", "city2"),
  ("country1", "city3"),
  ("country1", "city3"),
  ("country2", "city1"),
  ("country2", "city2")
))

您尝试的ReduceByKey需要进行一些修改，因为

rdd.map(x => (x._1, (Set(x._2), 1)))  //I have used Set to get distinct cities (you can use list or arrays or any other collection
  .reduceByKey((x,y)=> (x._1 ++ y._1, x._2 + y._2))  //cities are also summed and counts are also summed

这应该给你

(country2,(Set(city1, city2),2))
(country1,(Set(city1, city2, city3),4))

我希望答案有帮助

如果你想详细学习ReporteByKey可以查看我的详细解答

Answer 2

在这种情况下，我建议使用数据帧而不是RDD，并使用groupBy和agg方法。

您可以使用toDF函数轻松地将RDD转换为数据帧，只需确保首先导入隐式。示例假设RDD有两列：

val spark = SparkSession.builder.getOrCreate()
import spark.implicits._

val df = rdd.toDF("country", "city")

然后使用分组依据（groupBy）并聚合所需的值。

df.groupBy("country").agg(collect_set($"city").as("cities"), count($"city").as("count"))

Answer 3

将城市的数量和列表放在一起是复杂和多余的。您可以收集所有城市，并在末尾添加大小：

如果您使用数据帧接口，它当然会更容易（假设数据帧（key： Int， City： String））

import org.apache.spark.sql.{ functions => f}
import spark.implicits._
df.groupBy($"key").
   agg(f.collect_set($"city").as("cities")).
   withColumn("ncities", f.size($"cities"))

但您可以对原始rdd执行类似的操作（我假设在输入元组中为（id，city））

rdd.map{ x => (x(0),Set(x(1)))}.
   reduceByKey{ case(x,y) => x ++ y }.
   map { case(x,y:Set[_]) => (x,y, y.size)}