title: Hadoop JAVA MR job
body: <p>Hi I am new to Hadoop MR. I tried to write a simple MR job to count the shortest path for a node to its destination node.Basically the logic is like : </p>
  
  <p>If the input text file has the following paths given: 
  ABCD ABD ACD BED BD BACD </p>
  
  <p>The output should be as : ABD  BD </p>
  
  <p>which just gives the shortest path between nodes A and D and  the shortest path between B and D. </p>
  
  <p>The output I am getting is :  [ABCD ABD ACD BED BD  BACD] </p>
  
  <p>I have wrote the following MR to do the same. But its not giving the desired answer. I am running the MR in stand alone mode. </p>
  
  <p>Please let me know what is wrong with the code and the solution for it. Thanks a lot for your time. </p>
  
  <pre><code>public class Shpath {
  
  
      public static class Map extends MapReduceBase implements Mapper&lt;LongWritable, Text, Text, Text&gt; {
  
          public void map(LongWritable key, Text value, OutputCollector&lt;Text, Text&gt; output, Reporter reporter) throws IOException {
              String[] line = value.toString().split("\t");
              List&lt;String&gt; l = new ArrayList&lt;String&gt;();
  
              for(String lin :line){
                  l.add(lin);
              }
  
              List &lt;String&gt;startEnd = new ArrayList&lt;String&gt;();
              for(String s : l){
                  String g = s.substring(0,1)+s.substring((s.length())-1);
                  if(!startEnd.contains(g))
                  {
                      startEnd.add(g);
                  }
              }
  
              List &lt;String&gt; uniqueStringList = new ArrayList&lt;String&gt;();
              java.util.Map finalMap = new HashMap();
              for(String s1 : startEnd){
  
                  for(String s : l) {
                      if(s.startsWith(s1.substring(0,1)) &amp;&amp; (s.endsWith(s1.substring((s1.length())-1)))){
                          uniqueStringList.add(s);
                      }
                   }
                   String smallestKey = null;
                   int minSize = Integer.MAX_VALUE;
                   String smallest = null;
                   for(String s2 : uniqueStringList){
  
                       if(s2.length() &lt; minSize) {
                           minSize = s2.length();
                           smallest  = s2;
                           smallestKey  = s1;
                       }    
                       finalMap.put(s1,smallest);
  
                   }
                   uniqueStringList.clear();
              }output.collect(new Text(),new Text(finalMap.values().toString()));
          }
      }
  
      public static class Reduce extends MapReduceBase implements Reducer&lt;Text, Text, Text, Text&gt; {
          public void reduce(Text key, Iterator&lt;Text&gt; value, OutputCollector&lt;Text, Text&gt; output, Reporter reporter) throws IOException {
  
             while (value.hasNext()){
                 output.collect(new Text(key),new Text(value.next()));
             }
          }
      }
  
      public static void main(String[] args) throws Exception {
          JobConf conf = new JobConf(Shpath.class);
          conf.setJobName("shpath");
  
          conf.setOutputKeyClass(Text.class);
          conf.setOutputValueClass(Text.class);
  
          conf.setMapperClass(Map.class);
          conf.setCombinerClass(Reduce.class);
          conf.setReducerClass(Reduce.class);
  
          conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
          conf.setOutputFormat(org.apache.hadoop.mapred.TextOutputFormat.class);
  
          org.apache.hadoop.mapred.FileInputFormat.setInputPaths(conf, new Path(args[0]));
          org.apache.hadoop.mapred.FileOutputFormat.setOutputPath(conf, new Path(args[1]));
  
          JobClient.runJob(conf);
      }
  }
  </code></pre>