title: DataJoins in Hadoop MapReduce
body: <p>I am trying to implement one use case as given in Book Hadoop In Action, but I am not being to compile the code. I am new to Java so, not being able to understand the exact reasons behind the errors.</p>
  
  <p>Interesting thing is, another piece of coding using same classes and methods are compiled successfully.</p>
  
  <pre><code>hadoop@hadoopnode1:~/hadoop-0.20.2/playground/src$ javac -classpath /home/hadoop/hadoop-0.20.2/hadoop-0.20.2-core.jar:/home/hadoop/hadoop-0.20.2/lib/commons-cli-1.2.jar:/home/hadoop/hadoop-0.20.2/contrib/datajoin/hadoop-0.20.2-datajoin.jar -d ../classes DataJoin2.java 
  DataJoin2.java:49: cannot find symbol
  symbol  : constructor TaggedWritable(org.apache.hadoop.io.Text)
  location: class DataJoin2.TaggedWritable
              TaggedWritable retv = new TaggedWritable((Text) value);
                                    ^
  DataJoin2.java:69: cannot find symbol
  symbol  : constructor TaggedWritable(org.apache.hadoop.io.Text)
  location: class DataJoin2.TaggedWritable
              TaggedWritable retv = new TaggedWritable(new Text(joinedStr));
                                    ^
  DataJoin2.java:113: setMapperClass(java.lang.Class&lt;? extends org.apache.hadoop.mapreduce.Mapper&gt;) in org.apache.hadoop.mapreduce.Job cannot be applied to (java.lang.Class&lt;DataJoin2.MapClass&gt;)
          job.setMapperClass(MapClass.class);
             ^
  DataJoin2.java:114: setReducerClass(java.lang.Class&lt;? extends org.apache.hadoop.mapreduce.Reducer&gt;) in org.apache.hadoop.mapreduce.Job cannot be applied to (java.lang.Class&lt;DataJoin2.Reduce&gt;)
          job.setReducerClass(Reduce.class);
             ^
  4 errors
  </code></pre>
  
  <p>----------------code----------------------</p>
  
  <pre><code>import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapred.KeyValueTextInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  import org.apache.hadoop.util.GenericOptionsParser;
  
  // DataJoin Classes
  import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;
  import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;
  import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;
  
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  
  
  public class DataJoin2
  {
      public static class MapClass extends DataJoinMapperBase
      {
          protected Text generateInputTag(String inputFile)
          {
              String datasource = inputFile.split("-")[0];
              return new Text(datasource);            
          }
  
          protected Text generateGroupKey(TaggedMapOutput aRecord)
          {
              String line = ((Text) aRecord.getData()).toString();
              String[] tokens = line.split(",");
              String groupKey = tokens[0];
              return new Text(groupKey);
          }
  
          protected TaggedMapOutput generateTaggedMapOutput(Object value)
          {
              TaggedWritable retv = new TaggedWritable((Text) value);
              retv.setTag(this.inputTag);
              return retv;
          }
      } // End of class MapClass
  
      public static class Reduce extends DataJoinReducerBase
      {
          protected TaggedMapOutput combine(Object[] tags, Object[] values)
          {
              if (tags.length &lt; 2) return null;
              String joinedStr = "";
              for (int i=0;i&lt;values.length;i++)
              {
                  if (i&gt;0) joinedStr += ",";
                  TaggedWritable tw = (TaggedWritable) values[i];
                  String line = ((Text) tw.getData()).toString();
                  String[] tokens = line.split(",",2);
                  joinedStr += tokens[1];
              }
              TaggedWritable retv = new TaggedWritable(new Text(joinedStr));
              retv.setTag((Text) tags[0]);
              return retv;
          }
      } // End of class Reduce
  
      public static class TaggedWritable extends TaggedMapOutput 
      {
          private Writable data;
  
          public TaggedWritable()
          {
              this.tag = new Text("");
              this.data = data;
          }
  
          public Writable getData()
          {
              return data;
          }
  
          public void write(DataOutput out) throws IOException
          {
              this.tag.write(out);
              this.data.write(out);
          }
  
          public void readFields(DataInput in) throws IOException
          {
              this.tag.readFields(in);
              this.data.readFields(in);
          }       
      } // End of class TaggedWritable
  
      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: DataJoin2 &lt;in&gt; &lt;out&gt;");
            System.exit(2);
          }
          Job job = new Job(conf, "DataJoin");
          job.setJarByClass(DataJoin2.class);     
          job.setMapperClass(MapClass.class);
          job.setReducerClass(Reduce.class);
          job.setInputFormatClass(TextInputFormat.class);
  
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(TaggedWritable.class);
  
          FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
          FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
          System.exit(job.waitForCompletion(true) ? 0 : 1);               
      }
  }
  </code></pre>