Q: Map reduce example beside word count I followed step by step via example in here : https://www.tutorialspoint.com/hadoop/hadoop_mapreduce.htm
I want to find max of each year in file like the following:
1320 23
1221 60
1320 33
1221 66


And the result that I expected is:
1320 33
1221 66

And I did like the following in java:
import java.util.*; 

import java.io.IOException; 
import java.io.IOException; 

import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.conf.*; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapred.*; 
import org.apache.hadoop.util.*; 

public class ProcessUnits {
   //Mapper class 
   public static class E_EMapper extends MapReduceBase implements 
   Mapper<LongWritable ,/*Input key Type */ 
   Text,                /*Input value Type*/ 
   Text,                /*Output key Type*/ 
   IntWritable>        /*Output value Type*/ 
   {
      //Map function 
      public void map(LongWritable key, Text value, 
      OutputCollector<Text, IntWritable> output,   
      
      Reporter reporter) throws IOException { 
         String line = value.toString(); 
         String lasttoken = null; 
         StringTokenizer s = new StringTokenizer(line," "); 
         String year = s.nextToken(); 
         
         while(s.hasMoreTokens()) {
            lasttoken = s.nextToken();
         }
         int avgprice = Integer.parseInt(lasttoken); 
         output.collect(new Text(year), new IntWritable(avgprice)); 
      } 
   }
   
   //Reducer class 
   public static class E_EReduce extends MapReduceBase implements Reducer< Text, IntWritable, Text, IntWritable > {
   
      //Reduce function 
      public void reduce( Text key, Iterator <IntWritable> values, 
      OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { 
         int maxavg = 0 ; 
         int val = Integer.MIN_VALUE;
            
         while (values.hasNext()) { 
            val = values.next().get();
            if(val > maxavg) { 
                maxavg = val ;
                
            }
             
         }
         output.collect(key, new IntWritable(maxavg));
      } 
   }

   //Main function 
   public static void main(String args[])throws Exception { 
      JobConf conf = new JobConf(ProcessUnits.class); 
      
      conf.setJobName("max_eletricityunits"); 
      conf.setOutputKeyClass(Text.class);
      conf.setOutputValueClass(IntWritable.class); 
      conf.setMapperClass(E_EMapper.class); 
      conf.setCombinerClass(E_EReduce.class); 
      conf.setReducerClass(E_EReduce.class); 
      conf.setInputFormat(TextInputFormat.class); 
      conf.setOutputFormat(TextOutputFormat.class); 
      
      FileInputFormat.setInputPaths(conf, new Path(args[0])); 
      FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
      
      JobClient.runJob(conf); 
   } 
} 

The error I got when I execute this program is the following:

Error: java.util.NoSuchElementException

at java.util.StringTokenizer.nextToken(StringTokenizer.java:349)
    at ProcessUnits$E_EMapper.map(ProcessUnits.java:28)
    at ProcessUnits$E_EMapper.map(ProcessUnits.java:14)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:54)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:465)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:349)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:178)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:422)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1845)
    at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:172)

I know this problem is because my program can't maps line by line of file  , it maps entire file

    String line = value.toString(); 
    String lasttoken = null; 
    StringTokenizer s = new StringTokenizer(line," "); 
    String year = s.nextToken(); 
             
    while(s.hasMoreTokens()) {
       lasttoken = s.nextToken();
         }
    int avgprice = Integer.parseInt(lasttoken); 
    output.collect(new Text(year), new IntWritable(avgprice)); 


Any idea to solve this problem from you guys?

A: Try reading each line from file once and split the values. Map all the corresponding years and prices. Then using reduce function compare the price with some constant if greater assign the value.
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class E_Mapper extends Mapper<LongWritable, Text, Text, IntWritable> {
public void map(LongWritable ikey, Text ivalue, Context context)
        throws IOException, InterruptedException {
    
    String line= ivalue.toString();
    String [] values = line.splitBy(" ");
    for(String price:values)
    {
        context.write(new Text(year),price);          
}}}


public class E_Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {

public void reduce(Text key, Iterable<IntWritable> values, Context context)
        throws IOException, InterruptedException {
    int avg=0;
    for (IntWritable val : values) {
                   if(val.get()>avg){
             context.write(key,new IntWritable(sum));
}}}

