相关疑难解决方法(0)

假设我在3个节点上有一个分发系统,我的数据分布在这些节点中.例如,我有一个test.csv文件,它存在于所有3个节点上,它包含2列:

**row   | id,  c.**
---------------
row1  | k1 , c1  
row2  | k1 , c2  
row3  | k1 , c3  
row4  | k2 , c4  
row5  | k2 , c5  
row6  | k2 , c6  
row7  | k3 , c7  
row8  | k3 , c8  
row9  | k3 , c9  
row10 | k4 , c10   
row11 | k4 , c11  
row12 | k4 , c12 

然后我使用SparkContext.textFile将文件读出为rdd等等.据我所知,每个spark worker节点都会读取文件中的一部分.所以现在让我们说每个节点都会存储:

• 节点1:行1~4
• 节点2:第5~8行
• 节点3:第9~12行

我的问题是,假设我想对这些数据进行计算,我需要将一个键组合在一起,因此键值对将是[k1 [{k1 c1} {k1 c2} {k1 c3}]]..等等. …

我有一个类型的元组列表:(用户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而不是上面的代码来提供相同的映射？