title: Hadoop MapReduce - Euler's Totient / Sum of Totient (and other Mathematical operation)
body: <p>As part of my studies I am working on an implementation of a Sum of Totient (Euler's Totient) with different parallel computing languages, and I am quite struggling with the MapReduce one to be honest. 
  The main goal is to do kind of a benchmark with the runtime, efficiency and so on ...</p>
  
  <p>My code is running right now, and I got the right output, but it's pretty slow, and I am wondering why. </p>
  
  <p>Is it because of my implementation or is it because Hadoop MadReduce isn't made for this purpose.
  I also implemented a combiner because from what I read it was supposed to optimize the code, but it's not.
  Sorry if the question seems stupid but I didn't find anything on internet and I am tired of trying everything without any result.</p>
  
  <p>My input file is a range of value from 1 to 15000</p>
  
  <pre><code>1 2 3 4 5 6 ... 14998 14999 15000
  </code></pre>
  
  <p>And I am working on a cluster of 32 nodes, my goal was to have each node doing the calculation of a portion of my range (combiner), and then summing all the "sub sums" of the combiner in the reducer.</p>
  
  <pre><code>import java.io.IOException;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  
  public class NewTotient {
  
    public static long hcf(long x, long y)
    {
      long t;
  
      while (y != 0) {
        t = x % y;
        x = y;
        y = t;
      }
      return x;
    }
  
    public static boolean relprime(long x, long y)
    {
      return hcf(x, y) == 1;
    }
  
    public static long euler(long n)
    {
      long length, i;
  
      length = 0;
      for (i = 1; i &lt; n; i++)
        if (relprime(n, i))
          length++;
      return length;
    }
  
    public static class TotientMapper extends Mapper&lt;LongWritable, Text, Text, IntWritable&gt;{
  
      private final static IntWritable one = new IntWritable(1);
      private Text word = new Text();
      protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
          for (String val : value.toString().split(" ")) {
              context.write(new Text(), new IntWritable(Integer.valueOf(val)));
          }
      }
    }
  
    public static class TotientCombiner extends Reducer&lt;Text,IntWritable,Text,IntWritable&gt; {
      //private IntWritable result = new IntWritable();
  
      protected void reduce(Text key, Iterable&lt;IntWritable&gt; values, Context context)throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += NewTotient.euler(val.get());
            }
        }
    }
  
    public static class TotientReducer extends Reducer&lt;Text,IntWritable,Text,IntWritable&gt; {
      //private IntWritable result = new IntWritable();
  
      protected void reduce(Text key, Iterable&lt;IntWritable&gt; values, Context context)throws IOException, InterruptedException {
            int sum = 1;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(null, new IntWritable(sum));
        }
    }
  
    public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
      System.out.println("\n\n__________________________________________________________\n"+"Starting Job\n"+"__________________________________________________________\n\n");
      final long startTime = System.currentTimeMillis();
  
      Job job = Job.getInstance(conf, "Sum of Totient");
      job.setJarByClass(NewTotient.class);
      job.setMapperClass(TotientMapper.class);
      job.setCombinerClass(TotientCombiner.class);
      job.setReducerClass(TotientReducer.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);
      //job.setOutputKeyClass(Text.class);
      //job.setOutputValueClass(Text.class);
      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
      job.waitForCompletion(true);
      final double duration = (System.currentTimeMillis() - startTime)/1000.0;
      System.out.println("\n\n__________________________________________________________\n"+"Job Finished in " + duration + " seconds\n"+"__________________________________________________________\n\n");
      System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
  }
  </code></pre>
  
  <p>And if it can help here is my output for a Dataset going from 0 to 10 (so basically I am just calculating the sum of the 10 first Totient :</p>
  
  <pre><code>__________________________________________________________
  Starting Job
  __________________________________________________________
  
  
  2018-04-02 06:09:27,583 INFO client.RMProxy: Connecting to ResourceManager at bwlf32/137.195.143.132:33312
  2018-04-02 06:09:28,377 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
  2018-04-02 06:09:28,423 INFO mapreduce.JobResourceUploader: Disabling Erasure Coding for path: /tmp/hadoop-yarn/staging/jo20/.staging/job_1522471222360_0016
  2018-04-02 06:09:28,775 INFO input.FileInputFormat: Total input files to process : 1
  2018-04-02 06:09:29,029 INFO mapreduce.JobSubmitter: number of splits:1
  2018-04-02 06:09:29,101 INFO Configuration.deprecation: yarn.resourcemanager.system-metrics-publisher.enabled is deprecated. Instead, use yarn.system-metrics-publisher.enabled
  2018-04-02 06:09:29,288 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1522471222360_0016
  2018-04-02 06:09:29,290 INFO mapreduce.JobSubmitter: Executing with tokens: []
  2018-04-02 06:09:29,538 INFO conf.Configuration: resource-types.xml not found
  2018-04-02 06:09:29,539 INFO resource.ResourceUtils: Unable to find 'resource-types.xml'.
  2018-04-02 06:09:29,628 INFO impl.YarnClientImpl: Submitted application application_1522471222360_0016
  2018-04-02 06:09:29,687 INFO mapreduce.Job: The url to track the job: http://bwlf32:33314/proxy/application_1522471222360_0016/
  2018-04-02 06:09:29,688 INFO mapreduce.Job: Running job: job_1522471222360_0016
  2018-04-02 06:09:37,849 INFO mapreduce.Job: Job job_1522471222360_0016 running in uber mode : false
  2018-04-02 06:09:37,852 INFO mapreduce.Job:  map 0% reduce 0%
  2018-04-02 06:09:44,960 INFO mapreduce.Job:  map 100% reduce 0%
  2018-04-02 06:09:52,008 INFO mapreduce.Job:  map 100% reduce 100%
  2018-04-02 06:09:52,022 INFO mapreduce.Job: Job job_1522471222360_0016 completed successfully
  2018-04-02 06:09:52,178 INFO mapreduce.Job: Counters: 53
      File System Counters
          FILE: Number of bytes read=6
          FILE: Number of bytes written=414497
          FILE: Number of read operations=0
          FILE: Number of large read operations=0
          FILE: Number of write operations=0
          HDFS: Number of bytes read=123
          HDFS: Number of bytes written=0
          HDFS: Number of read operations=8
          HDFS: Number of large read operations=0
          HDFS: Number of write operations=2
      Job Counters
          Launched map tasks=1
          Launched reduce tasks=1
          Rack-local map tasks=1
          Total time spent by all maps in occupied slots (ms)=9126
          Total time spent by all reduces in occupied slots (ms)=9688
          Total time spent by all map tasks (ms)=4563
          Total time spent by all reduce tasks (ms)=4844
          Total vcore-milliseconds taken by all map tasks=4563
          Total vcore-milliseconds taken by all reduce tasks=4844
          Total megabyte-milliseconds taken by all map tasks=1168128
          Total megabyte-milliseconds taken by all reduce tasks=1240064
      Map-Reduce Framework
          Map input records=1
          Map output records=10
          Map output bytes=50
          Map output materialized bytes=6
          Input split bytes=102
          Combine input records=10
          Combine output records=0
          Reduce input groups=0
          Reduce shuffle bytes=6
          Reduce input records=0
          Reduce output records=0
          Spilled Records=0
          Shuffled Maps =1
          Failed Shuffles=0
          Merged Map outputs=1
          GC time elapsed (ms)=157
          CPU time spent (ms)=2220
          Physical memory (bytes) snapshot=507772928
          Virtual memory (bytes) snapshot=3889602560
          Total committed heap usage (bytes)=347078656
          Peak Map Physical memory (bytes)=306073600
          Peak Map Virtual memory (bytes)=1945808896
          Peak Reduce Physical memory (bytes)=201699328
          Peak Reduce Virtual memory (bytes)=1943793664
      Shuffle Errors
          BAD_ID=0
          CONNECTION=0
          IO_ERROR=0
          WRONG_LENGTH=0
          WRONG_MAP=0
          WRONG_REDUCE=0
      File Input Format Counters
          Bytes Read=21
      File Output Format Counters
          Bytes Written=0
  
  
  __________________________________________________________
  Job Finished in 26.225 seconds
  __________________________________________________________
  
  
  2018-04-02 06:09:52,182 INFO mapreduce.Job: Running job: job_1522471222360_0016
  2018-04-02 06:09:52,188 INFO mapreduce.Job: Job job_1522471222360_0016 running in uber mode : false
  2018-04-02 06:09:52,188 INFO mapreduce.Job:  map 100% reduce 100%
  2018-04-02 06:09:52,193 INFO mapreduce.Job: Job job_1522471222360_0016 completed successfully
  2018-04-02 06:09:52,201 INFO mapreduce.Job: Counters: 53
      File System Counters
          FILE: Number of bytes read=6
          FILE: Number of bytes written=414497
          FILE: Number of read operations=0
          FILE: Number of large read operations=0
          FILE: Number of write operations=0
          HDFS: Number of bytes read=123
          HDFS: Number of bytes written=0
          HDFS: Number of read operations=8
          HDFS: Number of large read operations=0
          HDFS: Number of write operations=2
      Job Counters
          Launched map tasks=1
          Launched reduce tasks=1
          Rack-local map tasks=1
          Total time spent by all maps in occupied slots (ms)=9126
          Total time spent by all reduces in occupied slots (ms)=9688
          Total time spent by all map tasks (ms)=4563
          Total time spent by all reduce tasks (ms)=4844
          Total vcore-milliseconds taken by all map tasks=4563
          Total vcore-milliseconds taken by all reduce tasks=4844
          Total megabyte-milliseconds taken by all map tasks=1168128
          Total megabyte-milliseconds taken by all reduce tasks=1240064
      Map-Reduce Framework
          Map input records=1
          Map output records=10
          Map output bytes=50
          Map output materialized bytes=6
          Input split bytes=102
          Combine input records=10
          Combine output records=0
          Reduce input groups=0
          Reduce shuffle bytes=6
          Reduce input records=0
          Reduce output records=0
          Spilled Records=0
          Shuffled Maps =1
          Failed Shuffles=0
          Merged Map outputs=1
          GC time elapsed (ms)=157
          CPU time spent (ms)=2220
          Physical memory (bytes) snapshot=507772928
          Virtual memory (bytes) snapshot=3889602560
          Total committed heap usage (bytes)=347078656
          Peak Map Physical memory (bytes)=306073600
          Peak Map Virtual memory (bytes)=1945808896
          Peak Reduce Physical memory (bytes)=201699328
          Peak Reduce Virtual memory (bytes)=1943793664
      Shuffle Errors
          BAD_ID=0
          CONNECTION=0
          IO_ERROR=0
          WRONG_LENGTH=0
          WRONG_MAP=0
          WRONG_REDUCE=0
      File Input Format Counters
          Bytes Read=21
      File Output Format Counters
          Bytes Written=0
  </code></pre>
  
  <p>And using my sequential code in java it's way faster :</p>
  
  <pre><code>real    0m0.512s
  user    0m0.279s
  sys     0m0.142s
  </code></pre>
  
  <p>Just to be clear, I have to use this way of calculation, because it's slow enough to have interesting comparison between different system, and I can't improve the speed of my system using a smarter way of calculation even if I know there are the idea of counting all prime factors and their multiples and subtract this count from n to get the totient function value (Prime factors and multiples of prime factors won’t have gcd as 1).</p>