【译】避免使用GroupByKey

by:leotse

原文:Avoid GroupByKey

译文

让我们来看两个wordcount的例子,一个使用了reduceByKey,而另一个使用groupByKey:

1
2
3
4
5
6
7
8
9
10
11
val words = Array("one", "two", "two", "three", "three", "three")
val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))

val wordCountsWithReduce = wordPairsRDD
.reduceByKey(_ + _)
.collect()

val wordCountsWithGroup = wordPairsRDD
.groupByKey()
.map(t => (t._1, t._2.sum))
.collect()

上面两个函数所得到的结果都是正确的,但是当数据集很大时,使用了reduceByKey的例子表现更佳。这是因为在shuffle输出的数据前,Spark会Combine每一个partition上具有相同key的输出结果。

看下图我们就能理解reduceByKey的工作流程。我们注意到同一台机器上数据shuffle之前,相同key的数据(通过调用传入reduceByKey的lambda函数)Combine在一起的,然后再一次调用这个lambda函数去reduce来自各个partition的全部值,从而得到最终的结果。

ReduceByKey

另一方面,当调用groupByKey的时候,所有的键值对都会进行shuffle,这将增加很多无谓的数据进行网络传输。

为了确定哪台机器将接受Shuffle后的键值对,Spark会针对该键值对数据的key调用一个分区函数。当某一台executor机器上的内存不足以保存过多的Shuffle后数据时,Spark就会溢写数据到磁盘上。然而,这种溢写磁盘会一次性将一个key的全部键值对数据写入磁盘,因此如果一个key拥有过多键值对数据——多到内存放不下时,将会抛出Out Of Memory异常。在之后发布的Spark中将会更加优雅地处理这种情况,使得这个job仍会继续运行,但是我们仍然需要避免(使用groupByKey)。当Spark需要溢写磁盘的时候,它的性能将受到严重影响

GroupByKey

如果你有一个非常大的数据集,那么reduceByKeygroupByKey进行shuffle的数据量之间的差异将会更加夸张。

下面是一些你可以用来替代groupByKey的函数:
1)当你在combine数据但是返回的数据类型因输入值的类型而异时,你可以使用combineByKey
2)如果key使用到结合函数和“零值”,你可以用foldByKey函数合并value;