java - wrong value class: class org.apache.hadoop.io.Text is not class org.apache.hadoop.io.IntWritable -

i have used 1 mapper,one reducer , 1 combiner class getting error below:

java.io.ioexception: wrong value class: class org.apache.hadoop.io.text not class org.apache.hadoop.io.intwritable @ org.apache.hadoop.mapred.ifile$writer.append(ifile.java:199) @ org.apache.hadoop.mapred.task$combineoutputcollector.collect(task.java:1307) @ org.apache.hadoop.mapred.task$newcombinerrunner$outputconverter.write(task.java:1623) @ org.apache.hadoop.mapreduce.task.taskinputoutputcontextimpl.write(taskinputoutputcontextimpl.java:89) @ org.apache.hadoop.mapreduce.lib.reduce.wrappedreducer$context.write(wrappedreducer.java:105) @ bookpublished1$combine.reduce(bookpublished1.java:47) @ bookpublished1$combine.reduce(bookpublished1.java:1) @ org.apache.hadoop.mapreduce.reducer.run(reducer.java:171) @ org.apache.hadoop.mapred.task$newcombinerrunner.combine(task.java:1644) @ org.apache.hadoop.mapred.maptask$mapoutputbuffer.sortandspill(maptask.java:1618) @ org.apache.hadoop.mapred.maptask$mapoutputbuffer.flush(maptask.java:1467) @ org.apache.hadoop.mapred.maptask$newoutputcollector.close(maptask.java:699) @ org.apache.hadoop.mapred.maptask.runnewmapper(maptask.java:769) @ org.apache.hadoop.mapred.maptask.run(maptask.java:339) @ org.apache.hadoop.mapred.yarnchild$2.run(yarnchild.java:162) @ java.security.accesscontroller.doprivileged(native method) @ javax.security.auth.subject.doas(subject.java:415) @ org.apache.hadoop.security.usergroupinformation.doas(usergroupinformation.java:1491) @ org.apache.hadoop.mapred.yarnchild.main(yarnchild.java:157) 

my entire program looks below:

import java.io.ioexception;  import org.apache.hadoop.io.floatwritable; import org.apache.hadoop.io.intwritable; import org.apache.hadoop.io.longwritable; import org.apache.hadoop.io.text; import org.apache.hadoop.mapreduce.mapper; import org.apache.hadoop.mapreduce.reducer; import org.apache.hadoop.conf.configuration; import org.apache.hadoop.mapreduce.job; import org.apache.hadoop.mapreduce.lib.input.textinputformat; import org.apache.hadoop.mapreduce.lib.output.textoutputformat; import org.apache.hadoop.mapreduce.lib.input.fileinputformat; import org.apache.hadoop.mapreduce.lib.output.fileoutputformat; import org.apache.hadoop.fs.path;   public class bookpublished1 {      public static class map extends mapper<longwritable,text,text,intwritable>{          public void map(longwritable key, text value,context context)                 throws ioexception,interruptedexception {              string line = value.tostring();             string [] stryear = line.split(";");             context.write(new text(stryear[3]), new intwritable(1));             }           }       public static class combine extends reducer<text,intwritable,text,text>{          public void reduce(text key, iterable<intwritable> values,context context)                 throws ioexception,interruptedexception {             int sum=0;             // todo auto-generated method stub             for(intwritable x: values)             {                 sum+=x.get();             }                context.write(new text("booksummary"), new text(key + "_"+ sum));          }      }  public static class reduce extends reducer<text,text,text,floatwritable>{          public void reduce(text key, iterable<text> values,context context)throws ioexception,interruptedexception              {             long publishyear =0l, max=long.max_value;                             text publishyear1 = null,maxyear=null;                             long publishvalue= 0l;             string compositestring;             string compositestringarray[];             // todo auto-generated method stub             for(text  x: values)             {                                                 compositestring = x.tostring();                 compositestringarray = compositestring.split("_");                 publishyear1=new text(compositestringarray[0]);                 publishvalue=new long(compositestringarray[1]);                 if(publishvalue > max){                 max=publishvalue;                 maxyear=publishyear1;              }             }         text keytext= new text("max" + " ( " + maxyear.tostring() + ") : ");              context.write(keytext, new floatwritable(max));        }  }       public static void main(string[] args) throws exception {         configuration conf= new configuration();         job job = new job(conf,"bookpublished");          job.setjarbyclass(bookpublished1.class);         job.setmapperclass(map.class);         job.setreducerclass(reduce.class);         job.setcombinerclass(combine.class);         job.setmapoutputkeyclass(text.class);         job.setmapoutputvalueclass(intwritable.class);         job.setoutputkeyclass(text.class);         job.setoutputvalueclass(floatwritable.class);         job.setinputformatclass(textinputformat.class);         job.setoutputformatclass(textoutputformat.class);            path outputpath = new path(args[1]);                     fileinputformat.addinputpath(job, new path(args[0]));                 fileoutputformat.setoutputpath(job, new path(args[1]));           outputpath.getfilesystem(conf).delete(outputpath);         system.exit(job.waitforcompletion(true) ? 0 : 1);     }  } 

please me resolution.

output types of combiner must match output types of mapper. hadoop makes no guarantees on how many times combiner applied, or applied @ all. , that's happens in case.

values map (<text, intwritable>) go directly reduce types <text, text> expected.