title: How to read multiple types of Avro data in single MapReduce
body: <p><strong>I have two different types of Avro data which have some common fields. I want to read those common fields in the mapper. I want to read this by spawning a single job in cluster.</strong></p>
  
  <p><em>Below is the sample avro schema</em></p>
  
  <p>Schema 1:</p>
  
  <blockquote>
    <p>{"type":"record","name":"Test","namespace":"com.abc.schema.SchemaOne","doc":"Avro storing with schema using MR.","fields":[{"name":"EE","type":"string","default":null},
    {"name":"AA","type":["null","long"],"default":null},
    {"name":"BB","type":["null","string"],"default":null},
    {"name":"CC","type":["null","string"],"default":null}]}</p>
  </blockquote>
  
  <p>Schema 2 :          </p>
  
  <blockquote>
    <p>{"type":"record","name":"Test","namespace":"com.abc.schema.SchemaTwo","doc":"Avro
    storing with schema using
    MR.","fields":[{"name":"EE","type":"string","default":null},
    {"name":"AA","type":["null","long"],"default":null},
    {"name":"CC","type":["null","string"],"default":null},
    {"name":"DD","type":["null","string"],"default":null}]}</p>
  </blockquote>
  
  <p>Driver Class:</p>
  
  <pre><code>package com.mango.schema.aggrDaily;
  
  import java.util.Date;
  import org.apache.avro.Schema;
  import org.apache.avro.mapred.AvroJob;
  import org.apache.avro.mapred.Pair;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.FileOutputFormat;
  import org.apache.hadoop.mapred.JobClient;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.RunningJob;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  
  public class AvroDriver extends Configured implements Tool {
  
      @Override
      public int run(String[] args) throws Exception {
  
          JobConf conf = new JobConf(super.getConf(), getClass());
          conf.setJobName("DF");
  
          args[0] = "hdfs://localhost:9999/home/hadoop/work/alok/aggrDaily/data/avro512MB/part-m-00000.avro";
          args[1] = "/home/hadoop/work/alok/tmp"; // temp location
          args[2] = "hdfs://localhost:9999/home/hadoop/work/alok/tmp/10";
  
          FileInputFormat.addInputPaths(conf, args[0]);
          FileOutputFormat.setOutputPath(conf, new Path(args[2]));
  
          AvroJob.setInputReflect(conf);
          AvroJob.setMapperClass(conf, AvroMapper.class);
  
          AvroJob.setOutputSchema(
                  conf,
                  Pair.getPairSchema(Schema.create(Schema.Type.STRING),
                          Schema.create(Schema.Type.INT)));
  
          RunningJob job = JobClient.runJob(conf);
  
          return 0;
      }
  
      public static void main(String[] args) throws Exception {
          long startTime = new Date().getTime();
          System.out.println("Start Time :::::" + startTime);
          Configuration conf = new Configuration();
          int exitCode = ToolRunner.run(conf, new AvroDriver(), args);
          long endTime = new Date().getTime();
          System.out.println("End Time :::::" + endTime);
          System.out.println("Total Time Taken:::"
                  + new Double((endTime - startTime) * 0.001) + "Sec.");
          System.exit(exitCode);
      }
  }
  </code></pre>
  
  <p>Mapper class:</p>
  
  <pre><code>package com.mango.schema.aggrDaily;
  
  import java.io.IOException;
  import org.apache.avro.generic.GenericData;
  import org.apache.avro.mapred.AvroCollector;
  import org.apache.avro.mapred.AvroMapper;
  import org.apache.avro.mapred.Pair;
  import org.apache.hadoop.mapred.Reporter;
  
  public class AvroMapper extends
          AvroMapper&lt;GenericData, Pair&lt;CharSequence, Integer&gt;&gt; {
  
      @Override
      public void map(GenericData record,
          AvroCollector&lt;Pair&lt;CharSequence, Integer&gt;&gt; collector, Reporter reporter) throws IOException {
          System.out.println("record :: " + record);
      }
  
  }
  </code></pre>
  
  <p><em>I able to read Avro data with this code by setting the input schema.</em></p>
  
  <p><code>AvroJob.setInputSchema(conf, new AggrDaily().getSchema());</code></p>
  
  <p>As the Avro data has builtin schema into the data, I don't want to pass the specific schema to the job explicitly. I achieve this in Pig. But now I want to achieve the same in MapReduce also.</p>
  
  <p>Can anybody help me to achieve this through MR code or let me know where am I going wrong?</p>