@@ -0,0 +1,2822 @@
+package org.myorg;%0A%0Aimport java.io.IOException;%0Aimport java.util.*;%0A%0Aimport org.apache.hadoop.fs.Path;%0Aimport org.apache.hadoop.conf.*;%0Aimport org.apache.hadoop.io.*;%0Aimport org.apache.hadoop.mapred.*;%0Aimport org.apache.hadoop.util.*;%0A%0Apublic class FullIndexTwoWord %7B%0A%0A  public static class Map extends MapReduceBase implements Mapper%3CLongWritable, Text, Text, Text%3E %7B%0A    private final static IntWritable one = new IntWritable(1);%0A    //Create variables to send as (key, value) pair%0A    private Text word = new Text();%0A    private Text fileVal = new Text();%0A%0A    public void map(LongWritable key, Text value, OutputCollector%3CText, Text%3E output, Reporter reporter) throws IOException %7B%0A      FileSplit fileSplit = (FileSplit) reporter.getInputSplit();%0A      String fileName = fileSplit.getPath().getName();%0A      //The key to be sent is appended with the line offset%0A      fileVal.set(fileName+%22@%22+key+%22, %22);%0A%0A      String line = value.toString();%0A      //Split string with whitespace%0A      String%5B%5D words = line.split(%22%5C%5Cs+%22);%0A      //form pairs of two word phrases%0A      for (int i = 0; i %3C (words.length-1); i++) %7B%0A%09if ( !(words%5Bi%5D.equals(%22%22)) && !(words%5Bi+1%5D.equals(%22%22)) ) %7B%0A%09%09word.set(words%5Bi%5D+%22 %22+words%5Bi+1%5D+%22 : %22);%0A%09%09output.collect(word, fileVal);%09%0A%09%7D%0A      %7D%0A    %7D%0A  %7D%0A%0A  public static class Reduce extends MapReduceBase implements Reducer%3CText, Text, Text, Text%3E %7B%0A    public void reduce(Text key, Iterator%3CText%3E values, OutputCollector%3CText, Text%3E output, Reporter reporter) throws IOException %7B%0A      int sum = 0;%0A      String tmpOut = %22%22;%0A      String tmpTest = %22%22;%0A      List%3CString%3E LList = new LinkedList%3CString%3E();%0A      //Create a list of the incoming values %0A      while (values.hasNext()) %7B%0A%09tmpOut = values.next().toString();%0A%09if ( !(LList.contains(tmpOut)) && (tmpOut != %22%22)) %7B%0A%09%09LList.add(tmpOut);%0A%09%7D%0A      %7D%0A      //Sort the list%0A      Collections.sort(LList);%0A      tmpOut = %22%22;%0A      ListIterator itr = LList.listIterator(0);%0A      //Append all the values together to create a new string which is sent out as value%0A      while (itr.hasNext()) %7B%0A%09tmpTest = itr.next().toString();%0A%09if (tmpTest != %22%22) %09%0A%09%09tmpOut += tmpTest; %0A      %7D%0A      output.collect(key, new Text(%22%22+tmpOut));%0A    %7D%0A  %7D%0A%0A  public static void main(String%5B%5D args) throws Exception %7B%0A    JobConf conf = new JobConf(FullIndexTwoWord.class);%0A    conf.setJobName(%22wordcount%22);%0A%0A    conf.setOutputKeyClass(Text.class);%0A    conf.setOutputValueClass(Text.class);%0A%0A    conf.setMapperClass(Map.class);%0A    conf.setCombinerClass(Reduce.class);%0A    conf.setReducerClass(Reduce.class);%0A%0A    conf.setInputFormat(TextInputFormat.class);%0A    conf.setOutputFormat(TextOutputFormat.class);%0A%0A    FileInputFormat.setInputPaths(conf, new Path(args%5B0%5D));%0A    FileOutputFormat.setOutputPath(conf, new Path(args%5B1%5D));%0A%0A    JobClient.runJob(conf);%0A  %7D%0A%7D%0A

Create FullIndexTwoWord.java