Dear Wiki user, You have subscribed to a wiki page or wiki category on "Hama Wiki" for change notification.
The "Combiner" page has been changed by edwardyoon: http://wiki.apache.org/hama/Combiner?action=diff&rev1=2&rev2=3 Comment: Refactor. <<TableOfContents(2)>> == Combiner == - Combiner in Hama BSP model is used to combine messages during [[SyncService|barrier sync]] 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 values. + Combiners 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. + + {{{ + public void bsp(BSPPeer<NullWritable, NullWritable, NullWritable, NullWritable> peer) throws IOException, + SyncException, InterruptedException { + + for (int i = 0; i < 1000; i++) { + peer.send(masterTask, new IntegerMessage(peer.getPeerName(), i)); + } + peer.sync(); + + if (peer.getPeerName().equals(masterTask)) { + IntegerMessage received; + while ((received = (IntegerMessage) peer.getCurrentMessage()) != null) { + sum += received.getData(); + } + } + } + }}} + + 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. {{{ public static class SumCombiner extends Combiner { @@ -29, +47 @@ }}} - Combiner actually gets called in sync function, - - {{{ - - public void sync() throws IOException, SyncException, InterruptedException { - ... - final BSPMessageBundle bundle = combineMessages(messages); - ... - } - }}} - - and in combineMessages function the system checks if Combiner needs to apply. - - {{{ - private BSPMessageBundle combineMessages(Iterable<BSPMessage> messages) { - if (!conf.getClass("bsp.combiner.class", Combiner.class).equals( - Combiner.class)) { - Combiner combiner = (Combiner) ReflectionUtils.newInstance(conf.getClass( - "bsp.combiner.class", Combiner.class), conf); - - return combiner.combine(messages); - } else { - ... - } - } - }}} - - [1]. MALEWICZ, G., AUSTERN, M. H., BIK, A. J., DEHNERT, J. C., HORN, I., LEISER, N., AND CZAJKOWSKI, G. Pregel: A system for large-scale graph processing. In Proceedings of SIGMOD (2010) -
