title: Hadoop does not understand Composite Keys are equal
body: <p>My data set has this format:</p>
  
  <pre><code>userID mediaID rating
  </code></pre>
  
  <p>I'd like to find the co-occurence of any pair of mediaIDs that receive a rating higher than a threshold across all users. To do this, I followed several examples to implement a composite key. I wrote a PairKey class which stores a unique pair, implemented compareTo and overridden hashCode and equals... </p>
  
  <pre><code>  public static class PairKey implements WritableComparable&lt;PairKey&gt; {
  
      private Integer lowID;
      private Integer highID;
  
  
      public PairKey() {
  
          this.lowID = -1;
          this.highID = -1;
  
      }
  
      public PairKey(Integer one, Integer two) {
          //should be impossible
          if (one.equals(two)) {
              throw new IllegalArgumentException("Cannot have a pair key with identical IDs");
          }
          if (one &lt; two) {
              lowID = one;
              highID = two;
          }
          else {
              lowID = two;
              highID = one;
          }
      }
  
      public Integer getLowID() {
          return lowID;
      }
  
      public Integer getHighID() {
          return highID;
      }
  
      public void setLowID(Integer _lowID) {
          lowID = _lowID;
      }
  
      public void setHighID(Integer _highID) {
          highID = _highID;
      }
  
      @Override
      public int compareTo(PairKey other) {
          int _lowCompare = lowID.compareTo(other.getLowID());
          if (_lowCompare != 0) {
              return _lowCompare;
          }
          int _highCompare = highID.compareTo(other.getHighID());
          return _highCompare;
      }
  
      @Override
      public void write(DataOutput dataOutput) throws IOException {
          dataOutput.writeInt(lowID.intValue());
          dataOutput.writeInt(highID.intValue());
      }
  
      @Override
      public void readFields(DataInput dataInput) throws IOException {
          lowID = new Integer(dataInput.readInt());
          highID = new Integer(dataInput.readInt());
      }
  
      @Override
      public String toString() {
          return "&lt;" + lowID + ", " + highID + "&gt;";
      }
  
      @Override
      public boolean equals(Object o) {
  
          if (this == o) {
              return true;
          }
          if ( o == null || this.getClass() != o.getClass()) {
              return false;
          }
  
          PairKey other = (PairKey) o;
  
          //compare fields
          if (this.lowID != null ?    this.lowID.equals(other.getLowID()) == false  : other.getLowID() != null) return false;
          if (this.highID != null ?   this.highID.equals(other.getHighID()) == false : other.getHighID() != null) return false;
  
          return true;
      }
  
  
      @Override
      public int hashCode() {
          int _lowHash = this.lowID.hashCode();
          int _highHash = this.highID.hashCode();
          return 163 * (_lowHash ) + _highHash;
      }
  }
  </code></pre>
  
  <p>Here is my mapper code, I store all movieIDs that passed the threshold in a set per user, then emit all the possible pairs in this set:</p>
  
  <pre><code>    public static class PairMapper extends Mapper&lt;Text, Text, PairKey, IntWritable&gt; {
  
      private Map&lt;Integer, SortedSet&lt;Integer&gt;&gt; temp = new HashMap&lt;Integer, SortedSet&lt;Integer&gt;&gt;();
      private IntWritable one = new IntWritable(1);
      private PairKey _key = new PairKey();
  
      public void map(Text key, Text value, Context context) throws IOException, InterruptedException {
          Integer userID = new Integer(key.toString());
          String[] vals = value.toString().split("\t");
          String _movieID = vals[0];
          String _rating = vals[1];
          Integer movieID = new Integer(_movieID);
          Integer rating = new Integer(_rating);
          if (rating &gt; 3) {
              SortedSet candidates  = temp.get(userID);
              if (candidates == null) {
                  candidates = new TreeSet&lt;Integer&gt;();
              }
              candidates.add(movieID);
              temp.put(userID, candidates);
  
          }
      }//map
  
      public void cleanup(Context context) throws IOException, InterruptedException {
  
          for (Map.Entry&lt;Integer, SortedSet&lt;Integer&gt;&gt; e : temp.entrySet()) {
  
              SortedSet&lt;Integer&gt; _set = e.getValue();
              Integer [] arr = _set.toArray(new Integer[_set.size()]);
              for (int i = 0 ; i &lt; arr.length-1 ; i++) {
                  for (int j = i+1 ; j &lt; arr.length ; j++) {
                      _key.setLowID(arr[i]);
                      _key.setHighID(arr[j]);
                      context.write(_key, one);
                  }//for j
  
              }//for i
  
  
  
  
          }
  
  
  
      }//cleanup
  
  
  
  }//PairMapper
  </code></pre>
  
  <p>This is my reducer:</p>
  
  <pre><code>   public static class PairReducer extends Reducer&lt;PairKey, Iterable&lt;IntWritable&gt;, Text, IntWritable&gt; {
  
      public void reduce(PairKey key, Iterable&lt;IntWritable&gt; vals, Context context) throws IOException, InterruptedException {
          int sum = 0;
          for (IntWritable val : vals) {
              sum+= val.get();
          }//for
          IntWritable result = new IntWritable(sum);
          context.write(new Text(key.toString()), result);
      } //reduce
  
  }
  </code></pre>
  
  <p>and this is my driver main method:</p>
  
  <pre><code> public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
      String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
  
      if (otherArgs.length != 2) {
          System.err.println("Usage: moviepairs &lt;in&gt; &lt;out&gt;");
          System.exit(2);
      }
  
      //CONFIGURE THE JOB
      Job job = new Job(conf, "movie pairs");
  
      job.setJarByClass(MoviePairs.class);
  
     job.setSortComparatorClass(CompositeKeyComparator.class);
     job.setPartitionerClass(NaturalKeyPartitioner.class);
     job.setGroupingComparatorClass(NaturalKeyGroupingComparator.class);
  
      //map-reduce classes
      job.setMapperClass(PairMapper.class);
      job.setCombinerClass(PairReducer.class);
      job.setReducerClass(PairReducer.class);
  
  
      //key-val classes
      job.setMapOutputKeyClass(PairKey.class);
      job.setMapOutputValueClass(IntWritable.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);
  
  
      job.setInputFormatClass(KeyValueTextInputFormat.class);
      FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
      FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
  
      System.exit(job.waitForCompletion(true)? 0 :1);
  
  }
  </code></pre>
  
  <p>I expect to get this in my reducer:</p>
  
  <pre><code>pair &lt;1,2&gt;: [1,1,1]
  </code></pre>
  
  <p>but instead the reducer does not seem to understand the equality of pairs. Outputting this instead:</p>
  
  <pre><code>pair&lt;1,2&gt;: [1]
  pair&lt;1,2&gt;: [1]
  pair&lt;1,2&gt;: [1]
  </code></pre>
  
  <p>Not sure what I'm missing there. As you can see I've tried several things like adding a custom sorter (which I don't believe I need and using a Grouping comparator, custom partitioner), but I think simply overriding hashcode/equals should account for that? (not sure). All the examples I found online seem to follow this and they all seem to work.</p>