在 Apache Spark (Scala) 中使用 reduceByKey

我有一个类型为：（用户 ID、名称、计数）的元组列表。

例如，

val x = sc.parallelize(List(
    ("a", "b", 1),
    ("a", "b", 1),
    ("c", "b", 1),
    ("a", "d", 1))
)

我试图将这个集合简化为一种类型，其中每个 元素名称被计算在内。

所以上面的 val x 被转换为：

(a,ArrayBuffer((d,1), (b,2)))
(c,ArrayBuffer((b,1)))

这是我当前使用的代码：

val byKey = x.map({case (id,uri,count) => (id,uri)->count})

val grouped = byKey.groupByKey
val count = grouped.map{case ((id,uri),count) => ((id),(uri,count.sum))}
val grouped2: org.apache.spark.rdd.RDD[(String, Seq[(String, Int)])] = count.groupByKey

grouped2.foreach(println)

我正在尝试使用reduceByKey，因为它的执行速度比groupByKey 更快。

如何实现reduceByKey来代替上面的代码来提供 相同的映射？

按照您的代码：

val byKey = x.map({case (id,uri,count) => (id,uri)->count})

你可以这样做：

val reducedByKey = byKey.reduceByKey(_ + _)

scala> reducedByKey.collect.foreach(println)
((a,d),1)
((a,b),2)
((c,b),1)

PairRDDFunctions[K,V].reduceByKey采用可应用于 RDD[(K,V)] 的 V 类型的关联归约函数。换句话说，你需要一个函数f[V](e1:V, e2:V) : V。在这种特殊情况下，整数求和：(x:Int, y:Int) => x+y or _ + _用短下划线表示法。

作为记录：reduceByKey表现优于groupByKey因为它尝试在 shuffle/reduce 阶段之前本地应用 reduce 函数。groupByKey将在分组之前强制对所有元素进行洗牌。