工人之间的平衡RDD划分-火花

Question

我正在使用x:key，y:set（values）的RDD，并将其调用为 。

#values: RDD of tuples (key, val)    
file = values.groupByKey().mapValues(set).cache()
info_file = array(file.map(lambda (x,y): len(y)).collect())
var = np.var(info_file) #extremely high
def f():
     ...
file.foreachPartition(f)

len（y）的方差非常高，以至于大约1%的对集（用百分位数方法验证）使得集合中的值总数的20%总数=np.sum（info_file）。如果Spark随机使用shuffle进行分区，那么很有可能会有1%的数据落入同一分区，从而导致工作人员之间的负载不平衡。

有没有办法确保“重”元组在分区中正常分布？我实际上将文件分成两个分区，重和轻，基于阈值 = np.percentile（info_file，99.9） 给出的 len（y） 阈值，以便分离这组元组，然后重新分区。

light = file.filter(lambda (x,y): len(y) < threshold).cache()
heavy = file.filter(lambda (x,y): len(y) >= threshold).cache()

light.foreachPartition(f)
heavy.foreachPartition(f)

但获得几乎相同的运行时间。负载可能已经优化，只是想检查我是否可以做更多/更好的事情。

Answer 1

您可以使用Ganglia来监控集群负载。这将为您提供一个很好的指示，表明可能导致集群负载不均衡的任何数据倾斜。

如果您确实有一个不幸的数据偏差，有一些方法可以解决它，例如重组数据或对键加盐。例如，参见StackOverflow Q

请注意，您也可以像现在这样，将数据分为< code >重分区和< code >轻分区，但在这种情况下，您希望< code >缓存文件，而不是< code >重和< code >轻，因为您需要多次处理< code >文件。像这样:

cachedFile = file.cache()

light = cachedFile.filter(lambda (x,y): len(y) < threshold)
heavy = cachedFile.filter(lambda (x,y): len(y) >= threshold)

light.foreachPartition(f)
heavy.foreachPartition(f)

希望有帮助。