apache-spark - Spark聚合函数——aggregateByKey是如何工作的？

Question

假设我有一个分布在 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 工作节点都会从文件中读取一部分。现在假设每个节点将存储:

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

我的问题是，假设我想对这些数据进行计算，并且有一个步骤我需要将键分组在一起，因此键值对将是 [k1 [{k1 c1} { k1 c2} {k1 c3}]].. 等等。

有一个函数叫做groupByKey()，使用起来非常昂贵，推荐使用aggregateByKey()。所以我想知道 groupByKey() 和 aggregateByKey() 在幕后是如何工作的？有人可以用我上面提供的例子来解释一下吗？打乱后，行驻留在每个节点上的什么位置？

Answer 1

aggregateByKey()与reduceByKey有很大不同。所发生的情况是，reduceByKey 是aggregateByKey 的一种特殊情况。

aggregateByKey() 将组合特定键的值，这种组合的结果可以是您指定的任何对象。您必须指定如何在一个分区(在同一节点中执行)内组合(“添加”)值以及如何组合来自不同分区(可能位于不同节点中)的结果。 reduceByKey 是一种特殊情况，因为组合的结果(例如求和)与值的类型相同，并且从不同分区组合时的操作也与组合内部值时的操作相同。分区。

一个例子:想象一下你有一个配对列表。您将其并行化:

val pairs = sc.parallelize(Array(("a", 3), ("a", 1), ("b", 7), ("a", 5)))

现在您想通过按键“组合”它们以产生总和。在这种情况下，reduceByKey 和aggregateByKey 是相同的:

val resReduce = pairs.reduceByKey(_ + _) //the same operation for everything
resReduce.collect
res3: Array[(String, Int)] = Array((b,7), (a,9))

//0 is initial value, _+_ inside partition, _+_ between partitions
val resAgg = pairs.aggregateByKey(0)(_+_,_+_)
resAgg.collect
res4: Array[(String, Int)] = Array((b,7), (a,9))

现在，假设您希望聚合是一组值，这是与整数不同的类型(整数之和也是整数):

import scala.collection.mutable.HashSet
//the initial value is a void Set. Adding an element to a set is the first
//_+_ Join two sets is the  _++_
val sets = pairs.aggregateByKey(new HashSet[Int])(_+_, _++_)
sets.collect
res5: Array[(String, scala.collection.mutable.HashSet[Int])]  =Array((b,Set(7)), (a,Set(1, 5, 3)))