====== MapReduce Tutorial: Implementing iterative MapReduce jobs faster using All-Reduce ======

Implementing an iterative computation by running a separate Hadoop job for every iteration is usually not very efficient (although it is fault tolerant).

If we have enough machines that all input data fits into memory, we can implement iterative computation like this:
  - start several machines, each reads a portion of input data
  - when all the machines are ready, for each iteration
    - locally compute as much as is possible
    - communicate with other machines and combine the results to complete this iteration

There are many ways how the machines communicate with one another to finish an iteration. On of the possible implementations is //AllReduce// operation.

An AllReduce operation does the following:
  * every machine provides a value
  * values from all the machines are combined using specified operation (e.g., addition, minimum, maximum) into one resulting value
  * the resulting value is distributed to all the machines

It is possible for each machine to provide more than value. In that case, all machines must provide the same number of values, the corresponding values get reduced in parallel and every machine gets the result of all the reductions.

===== Hadoop AllReduce implementation =====
An implementation of AllReduce operation is provided in our Hadoop framework. The implementation is in theory independent of Hadoop (it could be implemented using e.g. an SGE array job), but we use Hadoop framework as it provides features we would have to implement ourselves.

A Hadoop AllReduce job is implemented using a mapper only. The mapper should read the input data during ''map'' operation without providing any output. After all input data is read, ''cooperate'' function is called. As the name suggests, it is executed by all the machines in a synchronized fashion. In this method, the following ''allReduce'' methods can be called:
  * ''allReduce($number_ref, $reduce_op)'' -- each machine provides one value, which is reduced across all the machines and the result is returned to all machines.
  * ''allReduce($array_ref, $reduce_op)'' -- each machine provides multiple values. The ''$array_ref'' can contain numbers and more array references. The reduced values are stored in the original array in every machine.
The ''$reduce_op'' can currently be one on:
  * ''$self%%->%%REDUCE_ADD''
  * ''$self%%->%%REDUCE_MIN''
  * ''$self%%->%%REDUCE_MAX''

It is crucial that all the mappers run simultaneously. This can be achieved using the ''/net/projects/hadoop/bin/compute-splitsize'' script: for given Hadoop input and requested number of mappers, it computes the appropriate splitsize.

When the computation finishes, only one of the mappers should print the results, as all of them have the same results. For simplicity, the ''cooperate'' method has ''$writeResults'' argument, which is set in exactly one mapper.

===== Example =====
This example reads the keys of ''/net/projects/hadoop/examples/inputs/numbers-small'', computes the sum of all the keys and print it:
<code perl sum.pl>
package My::AllReduceMapper;
use Moose;
with 'Hadoop::AllReduceMapper';

use List::Util qw(sum);

has 'keys' => (is => 'rw', isa => 'ArrayRef[Num]', default => sub { [] });

sub map {
  my ($self, $key, $value, $context) = @_;

  push @{$self->keys}, $key;
}

sub cooperate {
  my ($self, $context, $writeResults) = @_;

  my $sum = sum @{$self->keys};
  $self->allReduce(\$sum, $self->REDUCE_ADD);

  $context->write($sum) if $writeResults;
}

package main;
use Hadoop::Runner;

my $runner = Hadoop::Runner->new(
  mapper => My::AllReduceMapper->new(),
  input_format => 'KeyValueTextInputFormat');

$runner->run();
</code>

You can run the example locally using:
  wget --no-check-certificate 'https://wiki.ufal.ms.mff.cuni.cz/_export/code/courses:mapreduce-tutorial:step-16?codeblock=0' -O sum.pl
  rm -rf step-16-out; perl sum.pl /net/projects/hadoop/examples/inputs/numbers-small step-16-out
  less step-16-out/part-*
 
To run on a cluster with //C// machines using //C// mappers:
  rm -rf step-16-out; M=#of_machines; INPUT=/net/projects/hadoop/examples/inputs/numbers-small; perl sum.pl -c $M `/net/projects/hadoop/bin/compute-splitsize $INPUT $M` $INPUT step-16-out
  less step-16-out/part-*

===== Exercise 1 =====

Implement an AllReduce job on ''/net/projects/hadoop/examples/inputs/numbers-small'', which computes
  * number of keys
  * mean of the keys
  * variance of the keys
  * minimum of the keys
  * maximum of the keys
You can download the template {{:courses:mapreduce-tutorial:step-16-exercise1.txt|statistics.pl}} and execute it using:
  wget --no-check-certificate 'https://wiki.ufal.ms.mff.cuni.cz/_media/courses:mapreduce-tutorial:step-16-exercise1.txt' -O statistics.pl
  # NOW VIEW THE FILE
  # $EDITOR statistics.pl
  rm -rf step-16-out; M=#of_machines; INPUT=/net/projects/hadoop/examples/inputs/numbers-small; perl statistics.pl -c $M `/net/projects/hadoop/bin/compute-splitsize $INPUT $M` $INPUT step-16-out
  less step-16-out/part-*

===== Exercise 2 =====

Implement an AllReduce job on ''/net/projects/hadoop/examples/inputs/numbers-small'', which computes median of the input data. You can use the following iterative algorithm:
  * At the beginning, set //min<sub>1</sub>// = ''Integer.MIN_VALUE'', //max<sub>1</sub>// = ''Integer.MAX_VALUE'', //index_to_find// = number_of_input_data / 2.
  * In step //i//, do the following:
    - Consider only input keys in range <//min<sub>i</sub>//, //max<sub>i</sub>//>.
    - Compute //split// = ceiling of mean of the keys.
    - If the //index_to_find// is in range <1+number of keys less than //split//, number of keys less or equal to //split//>, then ''split'' is median.
    - Else, if //index_to_find// is at most the number of keys less than //split//, set //max<sub>i+1</sub>// = //split//-1.
    - Else, set //min<sub>i+1</sub>// = //split//+1 and subtract from //index_to_find// the number of keys less or equal to //split//.

You can download the template {{:courses:mapreduce-tutorial:step-16-exercise2.txt|median.pl}} and execute it using:
  wget --no-check-certificate 'https://wiki.ufal.ms.mff.cuni.cz/_media/courses:mapreduce-tutorial:step-16-exercise2.txt' -O median.pl
  # NOW VIEW THE FILE
  # $EDITOR median.pl
  rm -rf step-16-out; M=#of_machines; INPUT=/net/projects/hadoop/examples/inputs/numbers-small; perl median.pl -c $M `/net/projects/hadoop/bin/compute-splitsize $INPUT $M` $INPUT step-16-out
  less step-16-out/part-*

Solution: {{:courses:mapreduce-tutorial:step-16-solution2.txt|median.pl}}.

===== Exercise 3 =====

Implement an AllReduce job on ''/net/projects/hadoop/examples/inputs/points-small'', which implements the [[http://en.wikipedia.org/wiki/K-means_clustering#Standard_algorithm|K-means clustering algorithm]]. See [[.:step-15|K-means clustering exercise]] for description of input data.

You can download the template {{:courses:mapreduce-tutorial:step-16-exercise3.txt|kmeans.pl}}. This template uses two environment variables:
  * ''CLUSTERS_NUM'' -- number of clusters
  * ''CLUSTERS_FILE'' -- file where to read the initial clusters from
You can download and compile it using:
  wget --no-check-certificate 'https://wiki.ufal.ms.mff.cuni.cz/_media/courses:mapreduce-tutorial:step-16-exercise3.txt' -O kmeans.pl
  # NOW VIEW THE FILE
  # $EDITOR kmeans.pl
You can run it using specified number of machines on the following input data:
  * ''/net/projects/hadoop/examples/inputs/points-small'':
<code>M=#of_machines; export CLUSTERS_NUM=50 CLUSTERS_FILE=/net/projects/hadoop/examples/inputs/points-small/points.txt
rm -rf step-16-out; perl kmeans.pl -c $M `/net/projects/hadoop/bin/compute-splitsize $CLUSTERS_FILE $M` $CLUSTERS_FILE step-16-out</code>
  * ''/net/projects/hadoop/examples/inputs/points-medium'':
<code>M=#of_machines; export CLUSTERS_NUM=100 CLUSTERS_FILE=/net/projects/hadoop/examples/inputs/points-medium/points.txt
rm -rf step-16-out; perl kmeans.pl -c $M `/net/projects/hadoop/bin/compute-splitsize $CLUSTERS_FILE $M` $CLUSTERS_FILE step-16-out</code>
  * ''/net/projects/hadoop/examples/inputs/points-large'':
<code>M=#of_machines; export CLUSTERS_NUM=200 CLUSTERS_FILE=/net/projects/hadoop/examples/inputs/points-large/points.txt
rm -rf step-16-out; perl kmeans.pl -c $M `/net/projects/hadoop/bin/compute-splitsize $CLUSTERS_FILE $M` $CLUSTERS_FILE step-16-out</code>

Solution: {{:courses:mapreduce-tutorial:step-16-solution3.txt|kmeans.pl}}, much faster solution with distance computations written in C: {{:courses:mapreduce-tutorial:step-16-solution3_c.txt|kmeans_C.pl}}.