下面是我使用writablecomparator的代码片段，但它不起作用

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class MovieComparator extends WritableComparator{

    public MovieComparator(){
        super(Movie.class);
    }
    @Override
    public int compare(WritableComparable o,WritableComparable o2){
        System.out.println("in compare");
        Movie m = (Movie)o;
        Movie m2 = (Movie)o2;
        System.out.println(m.compareTo(m2));
        return m.movieId.compareTo(m2.movieId);
    }

}

public class Movie implements WritableComparable {

    Text movieId;
    Text movieTitle;

    public Movie(Text movieId, Text movieTitle) {

        this.movieId = movieId;
        this.movieTitle = movieTitle;
    }

    public Movie(){

    }

    public String getMovieId() {
        return movieId.toString();
    }

    public void setMovieId(String movieId) {
        this.movieId = new Text(movieId);
    }

    public String getMovieTitle() {
        return movieTitle.toString();
    }

    public void setMovieTitle(String movieTitle) {
        this.movieTitle = new Text(movieTitle);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        //movieId = in.read;
        movieId.readFields(in);
        movieTitle.readFields(in);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        //out.writeUTF(movieId);
        //out.writeUTF(movieTitle);
        movieId.write(out);
        movieTitle.write(out);

    }

    @Override
    public int compareTo(Movie o) {

    //  System.out.println("in compareTo");
        int res=movieTitle.compareTo(o.movieTitle);
        return res;
    }

    @Override
    public int hashCode(){
        return movieId.hashCode();
    }

    @Override
    public boolean equals(Object o){
        Movie m=(Movie)o; 
        return movieId.equals(m.movieId);
    }

    @Override
    public String toString(){
        return movieTitle.toString();
    }

}

In driver class I am setting the comparator by below line

job.setSortComparatorClass(MovieComparator.class);

Can any body tell me where I am wrong in this at it gives exception below

14/09/08 14:17:03 WARN mapred.LocalJobRunner: job_local_0001
java.io.IOException: Spill failed
    at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1029)
    at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:691)
    at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
    at com.impetus.MovieMapper.map(MovieMapper.java:44)
    at com.impetus.MovieMapper.map(MovieMapper.java:1)
    at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:370)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:212)