title: Hadoop facebook mutual friends using mapreduce
body: <p>I was trying a mapreduce program in hadoop (Java version) , to find the mutual friends list from a json file . The json file content has the following pattern :</p>
  
  <pre><code>{"name":"abc","id":123} [{"name":"xyz","id":124},{"name":"def","id":125},{"name":"cxf","id":155}]
  {"name":"cxf","id":155} [{"name":"xyz","id":124},{"name":"abc","id":123},{"name":"yyy","id":129}]
  </code></pre>
  
  <p>Pattern to be interpreted as follows :</p>
  
  <p>friend json tab separated by array of related friends json's</p>
  
  <p>Hence abc has xyz , def and cxf as friends
  cxf has xyz abc and yyy as friends .</p>
  
  <p>Given the above the mutual friends between abc and cxf is xyz .</p>
  
  <p>tried to implement the same using mapreduce by creating custom writables , with the mapper emitting following key values , key being pair of friends and value being related friends of the first friend in the key (ie , pair of friends)</p>
  
  <pre><code>K-&gt;V
  (abc,xyz) -&gt; [xyz,def,cxf]
  (abc,def) -&gt; [xyz,def,cxf]
  (abc,cxf) -&gt; [xyz,def,cxf]
  (cxf,xyz) -&gt; [xyz,abc,yyy]
  (cxf,abc) -&gt; [xyz,abc,yyy]
  (cxf,yyy) -&gt; [xyz,abc,yyy]
  </code></pre>
  
  <p>The key here is actually a Custom writable , created a class which extends WritableComparable and i have overridden the compareTo method so that both these pairs (a,b) and (b,a) are same . But the problem i am facing is that the compareTo method is not invoked for all combinations of pairs and hence the reducer logic is failing. </p>
  
  <p>Based on the above example , there are 6 K, V pairs emitted by the mapper . But compareTo is invoked only 5 times key1.compareTo(key2) , key2.compareTo(key3), key3.compareTo(key4),key4.compareTo(key5),,key5.compareTo(key6) .</p>
  
  <p>Any idea why this is happening ?</p>
  
  <p>Below is the code as per the logic suggested by f11ler</p>
  
  <p><strong>Driver class :</strong></p>
  
  <pre><code>package com.facebook.updated;
  
  import java.io.IOException;
  
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  import org.apache.log4j.Logger;
  
  public class FacebookMain extends Configured implements Tool
  
  {
  
      Logger logger = Logger.getLogger(FacebookMain.class);
  
      public static void main(String[] args) throws Exception {
          System.exit(ToolRunner.run(new FacebookMain(), args));
  
      }
  
      @Override
      public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
          logger.info("Running======&gt;");
          Job job = Job.getInstance();
  
          job.setJarByClass(FacebookMain.class);
          job.setJobName("FBApp");
  
          job.setMapOutputKeyClass(Friend.class);
          job.setMapOutputValueClass(Friend.class);
  
          job.setOutputKeyClass(FriendPair.class);
          job.setOutputValueClass(Friend.class);
  
          job.setMapperClass(FacebookMapper.class);
          job.setReducerClass(FacebookReducer.class);
  
          job.setInputFormatClass(org.apache.hadoop.mapreduce.lib.input.TextInputFormat.class);
          job.setOutputFormatClass(SequenceFileOutputFormat.class);
  
          FileInputFormat.setInputPaths(job, new Path(args[0]));
          FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
          boolean val = job.waitForCompletion(true);
  
          return val ? 0 : 1;
  
      }
  
  }
  </code></pre>
  
  <p><strong>The customWritables (used to represent a friend and friendpair)</strong></p>
  
  <pre><code>package com.facebook.updated;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  
  import lombok.Getter;
  import lombok.Setter;
  
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.log4j.Logger;
  
  @Getter
  @Setter
  public class Friend implements WritableComparable&lt;Friend&gt; {
  
      Logger logger = Logger.getLogger(Friend.class);
  
      private IntWritable id;
      private Text name;
  
      public Friend() {
          this.id = new IntWritable();
          this.name = new Text();
      }
  
      @Override
      public int compareTo(Friend arg0) {
          int val = getId().compareTo(arg0.getId());
          logger.info("compareTo Friend ======&gt; " + arg0 + " and " + this + " compare is " + val);
          return val;
      }
  
      @Override
      public void readFields(DataInput in) throws IOException {
          id.readFields(in);
          name.readFields(in);
      }
  
      @Override
      public void write(DataOutput out) throws IOException {
          id.write(out);
          name.write(out);
      }
  
      @Override
      public boolean equals(Object obj) {
          Friend f2 = (Friend) obj;
          boolean val = this.getId().equals(f2.getId());
          //logger.info("equals Friend ======&gt; " + obj + " and " + this);
          return val;
      }
  
      @Override
      public String toString() {
          return id + ":" + name + " ";
      }
  }
  
  package com.facebook.updated;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  
  import lombok.Getter;
  import lombok.Setter;
  
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.log4j.Logger;
  
  @Getter
  @Setter
  public class FriendPair implements WritableComparable&lt;FriendPair&gt; {
  
      Logger logger = Logger.getLogger(FriendPair.class);
  
      private Friend first;
      private Friend second;
  
      public FriendPair() {
          this.first = new Friend();
          this.second = new Friend();
      }
  
      public FriendPair(Friend f1, Friend f2) {
          this.first = f1;
          this.second = f2;
      }
  
      @Override
      public int compareTo(FriendPair o) {
  
          logger.info("compareTo FriendPair ======&gt; " + o + " and " + this);
          FriendPair pair2 = o;
          int cmp = -1;
  
          if (getFirst().compareTo(pair2.getFirst()) == 0 || getFirst().compareTo(pair2.getSecond()) == 0) {
              cmp = 0;
          }
          if (cmp != 0) {
              // logger.info("compareTo FriendPair ======&gt; " + o + " and " + this
              // + " comparison is " + cmp);
              return cmp;
          }
          cmp = -1;
          if (getSecond().compareTo(pair2.getFirst()) == 0 || getSecond().compareTo(pair2.getSecond()) == 0) {
              cmp = 0;
          }
  
          // logger.info("compareTo FriendPair ======&gt; " + o + " and " + this +
          // " comparison is " + cmp);
  
          // logger.info("getFirst() " + getFirst());
          // logger.info("pair2.getFirst() " + pair2.getFirst());
          // logger.info("getFirst().compareTo(pair2.getFirst()) " +
          // getFirst().compareTo(pair2.getFirst()));
          // logger.info("getFirst().compareTo(pair2.getSecond()) " +
          // getFirst().compareTo(pair2.getSecond()));
          // logger.info("getSecond().compareTo(pair2.getFirst()) " +
          // getSecond().compareTo(pair2.getFirst()));
          // logger.info("getSecond().compareTo(pair2.getSecond()) " +
          // getSecond().compareTo(pair2.getSecond()));
          // logger.info("pair2.getSecond() " + pair2.getSecond());
          // logger.info("getSecond() " + getSecond());
          // logger.info("pair2.getFirst() " + pair2.getFirst());
          // logger.info("pair2.getSecond() " + pair2.getSecond());
  
          return cmp;
      }
  
      @Override
      public boolean equals(Object obj) {
  
          FriendPair pair1 = this;
          FriendPair pair2 = (FriendPair) obj;
  
          boolean eq = false;
  
          logger.info("equals FriendPair ======&gt; " + obj + " and " + this);
  
          if (pair1.getFirst().equals(pair2.getFirst()) || pair1.getFirst().equals(pair2.getSecond()))
              eq = true;
  
          if (!eq) {
              // logger.info("equals FriendPair ======&gt; " + obj + " and " + this +
              // " equality is " + eq);
              return false;
          }
          if (pair1.getSecond().equals(pair2.getFirst()) || pair1.getSecond().equals(pair2.getSecond()))
              eq = true;
  
          // logger.info("equals FriendPair ======&gt; " + obj + " and " + this +
          // " equality is " + eq);
  
          return eq;
      }
  
      @Override
      public void readFields(DataInput in) throws IOException {
          first.readFields(in);
          second.readFields(in);
      }
  
      @Override
      public void write(DataOutput out) throws IOException {
          first.write(out);
          second.write(out);
      }
  
      @Override
      public String toString() {
          return "[" + first + ";" + second + "]";
      }
  
      @Override
      public int hashCode() {
          logger.info("hashCode FriendPair ======&gt; " + this);
          return first.getId().hashCode() + second.getId().hashCode();
      }
  }
  </code></pre>
  
  <p><strong>Mapper and Reducer</strong></p>
  
  <pre><code>package com.facebook.updated;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.StringTokenizer;
  
  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.log4j.Logger;
  
  import com.mongodb.BasicDBList;
  import com.mongodb.BasicDBObject;
  import com.mongodb.util.JSON;
  
  public class FacebookMapper extends Mapper&lt;LongWritable, Text, Friend, Friend&gt; {
  
      Logger log = Logger.getLogger(FacebookMapper.class);
  
      @Override
      protected void map(LongWritable key, Text value, Mapper&lt;LongWritable, Text, Friend, Friend&gt;.Context context)
              throws IOException, InterruptedException {
  
          String line = value.toString();
          StringTokenizer st = new StringTokenizer(line, "\t");
          String person = st.nextToken();
          String friends = st.nextToken();
  
          BasicDBObject personObj = (BasicDBObject) JSON.parse(person);
          BasicDBList friendsList = (BasicDBList) JSON.parse(friends);
  
          List&lt;Friend&gt; frndJavaList = new ArrayList&lt;&gt;();
  
          for (Object frndObj : friendsList) {
              frndJavaList.add(getFriend((BasicDBObject) frndObj));
          }
  
          Friend frnd = getFriend(personObj);
          Friend[] array = frndJavaList.toArray(new Friend[frndJavaList.size()]);
          for (Friend f : array) {
              log.info("Map output is " + f + " and " + frnd);
              context.write(f, frnd);
          }
      }
  
      private static Friend getFriend(BasicDBObject personObj) {
          Friend frnd = new Friend();
          frnd.setId(new IntWritable(personObj.getInt("id")));
          frnd.setName(new Text(personObj.getString("name")));
          frnd.setHomeTown(new Text(personObj.getString("homeTown")));
          return frnd;
      }
  }
  
  package com.facebook.updated;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.log4j.Logger;
  
  public class FacebookReducer extends Reducer&lt;Friend, Friend, FriendPair, Friend&gt; {
  
      Logger log = Logger.getLogger(FacebookReducer.class);
  
      @Override
      protected void reduce(Friend friend, Iterable&lt;Friend&gt; vals,
              Reducer&lt;Friend, Friend, FriendPair, Friend&gt;.Context context) throws IOException, InterruptedException {
          List&lt;Friend&gt; friends = new ArrayList&lt;&gt;();
          for (Friend frnd : vals) {
              friends.add(frnd);
          }
          log.info("Reducer output is " + friend + " and values are " + friends);
          if (friends.size() == 2) {
              FriendPair key = new FriendPair(friends.get(0), friends.get(1));
              context.write(key, friend);
          } else {
              //log.info("Size of friends is not 2 key is " + friend + " and values are " + friends);
          }
  
      }
  }
  </code></pre>
  
  <p><strong>Input json file containing 2 lines</strong></p>
  
  <pre><code>{"name":"abc","id":123} [{"name":"xyz","id":124},{"name":"def","id":125},{"name":"cxf","id":155}]
  {"name":"cxf","id":155} [{"name":"xyz","id":124},{"name":"abc","id":123},{"name":"yyy","id":129}]
  </code></pre>
  
  <p><strong>Output of reducer</strong>
  (abc,abc)->xyz</p>