public class PartitionCoalescer
extends Object
prev) into fewer partitions, so that each partition of
this RDD computes one or more of the parent ones. It will produce exactly maxPartitions if the
parent had more than maxPartitions, or fewer if the parent had fewer.
This transformation is useful when an RDD with many partitions gets filtered into a smaller one, or to avoid having a large number of small tasks when processing a directory with many files.
If there is no locality information (no preferredLocations) in the parent, then the coalescing is very simple: chunk parents that are close in the Array in chunks. If there is locality information, it proceeds to pack them with the following four goals:
(1) Balance the groups so they roughly have the same number of parent partitions (2) Achieve locality per partition, i.e. find one machine which most parent partitions prefer (3) Be efficient, i.e. O(n) algorithm for n parent partitions (problem is likely NP-hard) (4) Balance preferred machines, i.e. avoid as much as possible picking the same preferred machine
Furthermore, it is assumed that the parent RDD may have many partitions, e.g. 100 000. We assume the final number of desired partitions is small, e.g. less than 1000.
The algorithm tries to assign unique preferred machines to each partition. If the number of desired partitions is greater than the number of preferred machines (can happen), it needs to start picking duplicate preferred machines. This is determined using coupon collector estimation (2n log(n)). The load balancing is done using power-of-two randomized bins-balls with one twist: it tries to also achieve locality. This is done by allowing a slack (balanceSlack) between two bins. If two bins are within the slack in terms of balance, the algorithm will assign partitions according to locality. (contact alig for questions)
| Modifier and Type | Class and Description |
|---|---|
class |
PartitionCoalescer.LocationIterator |
| Constructor and Description |
|---|
PartitionCoalescer(int maxPartitions,
RDD<?> prev,
double balanceSlack) |
| Modifier and Type | Method and Description |
|---|---|
boolean |
addPartToPGroup(Partition part,
PartitionGroup pgroup) |
boolean |
compare(scala.Option<PartitionGroup> o1,
scala.Option<PartitionGroup> o2) |
boolean |
compare(PartitionGroup o1,
PartitionGroup o2) |
scala.collection.Seq<String> |
currPrefLocs(Partition part) |
scala.Option<PartitionGroup> |
getLeastGroupHash(String key)
Sorts and gets the least element of the list associated with key in groupHash
The returned PartitionGroup is the least loaded of all groups that represent the machine "key"
|
PartitionGroup[] |
getPartitions() |
scala.collection.mutable.ArrayBuffer<PartitionGroup> |
groupArr() |
scala.collection.mutable.Map<String,scala.collection.mutable.ArrayBuffer<PartitionGroup>> |
groupHash() |
scala.collection.mutable.Set<Partition> |
initialHash() |
boolean |
noLocality() |
PartitionGroup |
pickBin(Partition p)
Takes a parent RDD partition and decides which of the partition groups to put it in
Takes locality into account, but also uses power of 2 choices to load balance
It strikes a balance between the two use the balanceSlack variable
|
scala.util.Random |
rnd() |
PartitionGroup[] |
run()
Runs the packing algorithm and returns an array of PartitionGroups that if possible are
load balanced and grouped by locality
|
void |
setupGroups(int targetLen)
Initializes targetLen partition groups and assigns a preferredLocation
This uses coupon collector to estimate how many preferredLocations it must rotate through
until it has seen most of the preferred locations (2 * n log(n))
|
int |
slack() |
void |
throwBalls() |
public PartitionCoalescer(int maxPartitions,
RDD<?> prev,
double balanceSlack)
public boolean compare(PartitionGroup o1, PartitionGroup o2)
public boolean compare(scala.Option<PartitionGroup> o1, scala.Option<PartitionGroup> o2)
public scala.util.Random rnd()
public scala.collection.mutable.ArrayBuffer<PartitionGroup> groupArr()
public scala.collection.mutable.Map<String,scala.collection.mutable.ArrayBuffer<PartitionGroup>> groupHash()
public scala.collection.mutable.Set<Partition> initialHash()
public int slack()
public boolean noLocality()
public scala.collection.Seq<String> currPrefLocs(Partition part)
public scala.Option<PartitionGroup> getLeastGroupHash(String key)
key - string representing a partitioned group on preferred machine keypublic boolean addPartToPGroup(Partition part, PartitionGroup pgroup)
public void setupGroups(int targetLen)
targetLen - public PartitionGroup pickBin(Partition p)
p - partition (ball to be thrown)public void throwBalls()
public PartitionGroup[] getPartitions()
public PartitionGroup[] run()