public class DefaultPartitionCoalescer extends Object implements PartitionCoalescer
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, where 1.0 is all locality, 0 is all balance) 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 |
DefaultPartitionCoalescer.PartitionLocations |
| Constructor and Description |
|---|
DefaultPartitionCoalescer(double balanceSlack) |
| Modifier and Type | Method and Description |
|---|---|
boolean |
addPartToPGroup(Partition part,
PartitionGroup pgroup) |
double |
balanceSlack() |
PartitionGroup[] |
coalesce(int maxPartitions,
RDD<?> prev)
Runs the packing algorithm and returns an array of PartitionGroups that if possible are
load balanced and grouped by locality
|
boolean |
compare(scala.Option<PartitionGroup> o1,
scala.Option<PartitionGroup> o2) |
boolean |
compare(PartitionGroup o1,
PartitionGroup o2) |
scala.collection.Seq<String> |
currPrefLocs(Partition part,
RDD<?> prev) |
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,
RDD<?> prev,
double balanceSlack,
DefaultPartitionCoalescer.PartitionLocations partitionLocs)
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 using the balanceSlack variable
|
scala.util.Random |
rnd() |
void |
setupGroups(int targetLen,
DefaultPartitionCoalescer.PartitionLocations partitionLocs)
Initializes targetLen partition groups.
|
void |
throwBalls(int maxPartitions,
RDD<?> prev,
double balanceSlack,
DefaultPartitionCoalescer.PartitionLocations partitionLocs) |
public 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 boolean noLocality()
public scala.Option<PartitionGroup> getLeastGroupHash(String key)
key - string representing a partitioned group on preferred machine keyPartitionGroup that has least elements for keypublic boolean addPartToPGroup(Partition part, PartitionGroup pgroup)
public void setupGroups(int targetLen,
DefaultPartitionCoalescer.PartitionLocations partitionLocs)
targetLen - partitionLocs - (undocumented)public PartitionGroup pickBin(Partition p, RDD<?> prev, double balanceSlack, DefaultPartitionCoalescer.PartitionLocations partitionLocs)
p - partition (ball to be thrown)balanceSlack - determines the trade-off between load-balancing the partitions sizes and
their locality. e.g., balanceSlack=0.10 means that it allows up to 10%
imbalance in favor of localityprev - (undocumented)partitionLocs - (undocumented)public void throwBalls(int maxPartitions,
RDD<?> prev,
double balanceSlack,
DefaultPartitionCoalescer.PartitionLocations partitionLocs)
public PartitionGroup[] getPartitions()
public PartitionGroup[] coalesce(int maxPartitions, RDD<?> prev)
coalesce in interface PartitionCoalescermaxPartitions - (undocumented)prev - (undocumented)