我有一个类型为:(用户 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)] 的 to 类型 V 的关联归约函数。换句话说,你需要一个函数 f[V](e1:V, e2:V) : V
.在这种特殊情况下,整数总和:(x:Int, y:Int) => x+y
或 _ + _
简而言之,下划线表示法。备案:
reduceByKey
性能优于 groupByKey
因为它试图在 shuffle/reduce 阶段之前在本地应用 reduce 函数。 groupByKey
将在分组之前强制对所有元素进行洗牌。
关于scala - 在 Apache Spark (Scala) 中使用 reduceByKey,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/24071560/