Combiner
| Wiki Markup |
|---|
Combiner in Hama BSP model is used to combine messages during [barrier sync|SyncService] stage. This can reduce messages sent to other peers because sending messages somehow incurs overhead\[1\]. |
An example
Combiner's usage can be seen in source code - CombinerExample.java, in which an user defined combiner, SumCombiner, sums up messages to be sent and adds the result to BSPMessageBundle class. This decreases the total messages to just one IntegerMessage with a summary of all valuesCombiners are used for performing message aggregation to reduce communication overhead in cases when messages can be summarized arithmetically e.g., min, max, sum, and average at the sender side. Suppose that you want to send the integer messages to a specific processor from 0 to 1000 and sum all received the integer messages from all processors.
| No Format |
|---|
public static class SumCombiner extends Combiner {
@Override
void bsp(BSPPeer<NullWritable, NullWritable, NullWritable, NullWritable> peer) throws IOException,
public BSPMessageBundle combine(Iterable<BSPMessage> messages)SyncException, InterruptedException {
for BSPMessageBundle(int bundlei = new BSPMessageBundle();
0; i < 1000; i++) {
int sum = 0;
peer.send(masterTask, new IntegerMessage(peer.getPeerName(), i));
}
Iterator<BSPMessage> it = messagespeer.iteratorsync();
whileif (it.hasNext(peer.getPeerName().equals(masterTask)) {
sum + IntegerMessage received;
while ((received = ((IntegerMessage) itpeer.nextgetCurrentMessage()).getData(); != null) {
}
sum += bundle.addMessage(new IntegerMessage("Sum", sum))received.getData();
return bundle;}
}
}
|
If you follow the previous example, Each bsp processor will send a bundle of thousand Integer messages to a masterTask. Instead, you could use a Combiners in your BSP program to perform a sum Integer messages as below, that is what the Combiners are useful.Combiner actually gets called in sync function,
| No Format |
|---|
public static class publicSumCombiner voidextends sync() throws IOException, SyncException, InterruptedException {Combiner {
@Override
...
public BSPMessageBundle combine(Iterable<BSPMessage> messages) {
final BSPMessageBundle bundle = combineMessagesnew BSPMessageBundle(messages);
int ...
sum = }
|
and in combineMessages function the system checks if Combiner needs to apply.
| No Format |
|---|
private BSPMessageBundle combineMessages(Iterable<BSPMessage> messages) {
if (!conf.getClass("bsp.combiner.class", Combiner.class).equals(0;
Iterator<BSPMessage> it = messages.iterator();
while Combiner.class(it.hasNext()) {
Combiner combinersum += ((CombinerIntegerMessage) ReflectionUtilsit.newInstancenext(conf)).getClassgetData();
}
"bsp.combiner.class bundle.addMessage(new IntegerMessage("Sum", Combiner.class), confsum));
return combiner.combine(messages)bundle;
} else {
...
}
}
|
...
