public class ReadAvroInput {

public static class ReadAvroInputMap extends Mapper
    
     
      , NullWritable, Text, IntWritable> {    private Text oKey = new Text();    private final IntWritable ONE = new IntWritable(1);    private UserActionLog keyData;    @Override    protected void map(AvroKey
      
        key, NullWritable value,            Mapper
       
        
         , NullWritable, Text, IntWritable>.Context context)            throws IOException, InterruptedException {        keyData = key.datum();        oKey.set(keyData.getProvience().toString());        context.write(oKey, ONE);    }}public static class ReadAvroInputReducer extends Reducer
         
           { private int sum; private IntWritable oValue = new IntWritable(); @Override protected void reduce(Text key, Iterable
          
            values, Reducer
           
            .Context context) throws IOException, InterruptedException { sum=0; for (IntWritable value : values) { sum += value.get(); } oValue.set(sum); context.write(key, oValue); }}public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration configuration =new Configuration(); Job job =Job.getInstance(configuration); job.setJarByClass(ReadAvroInput.class); job.setJobName("重读avro文件进行mr计算"); job.setMapperClass(ReadAvroInputMap.class); job.setCombinerClass(ReadAvroInputReducer.class); job.setReducerClass(ReadAvroInputReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setInputFormatClass(AvroKeyInputFormat.class); AvroJob.setInputKeySchema(job, UserActionLog.getClassSchema()); FileInputFormat.addInputPath(job, new Path("/ReducerJoin/part-r-00000.avro")); Path outputPath =new Path("/ReadAvroInput"); outputPath.getFileSystem(configuration).delete(outputPath, true); FileOutputFormat.setOutputPath(job, outputPath); System.exit(job.waitForCompletion(true)?0:1);}
           
          
         
        
       
      
     
    

}

UserActionLog是通过mvn 指令通过schema框架生成的