这篇文章展示了使用Avro MapReduce API的Avro MapReduce示例程序。
作为示例，使用了字数统计MapReduce程序，其中输出将是Avro数据文件。

所需的jar包

avro-mapred-1.8.2.jar

Avro字数统计MapReduce示例

由于输出是Avro文件，因此必须定义Avro架构，因此架构中将有两个字段" word"和" count"。

在代码中，我们可以看到键和值对使用AvroKey和AvroValue。同样对于输出，使用AvroKeyOutputFormat类。
要定义地图输出，并将MaReduce作业的输出AvroJob类用于作业配置。

Avro MapReduce

import java.io.IOException;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class AvroWordCount extends Configured implements Tool{
	/// Schema
  private	static final Schema AVRO_SCHEMA = new Schema.Parser().parse(
    "{\n" +
    "	\"type\":	\"record\",\n" +				
    "	\"name\":	\"WordCount\",\n" +
    "	\"doc\":	\"word count\",\n" +
    "	\"fields\":\n" + 
    "	[\n" + 
    "			{\"name\": \"word\",	\"type\":	\"string\"},\n"+ 
    "			{\"name\":	\"count\", \"type\":	\"int\"}\n"+
    "	]\n"+
    "}\n");
	
	// Map function    
  public static class AvroWordMapper extends Mapper<LongWritable, Text, AvroKey<Text>, 
      AvroValue<GenericRecord>>{
    private Text word = new Text();
    private GenericRecord record = new GenericData.Record(AVRO_SCHEMA);
    public void map(LongWritable key, Text value, Context context) 
        throws IOException, InterruptedException {
      // Splitting the line on spaces
      String[] stringArr = value.toString().split("\s+");
      for (String str : stringArr) {
        word.set(str);
        // creating Avro record
        record.put("word", str);
        record.put("count", 1);
        context.write(new AvroKey<Text>(word), new AvroValue<GenericRecord>(record));
      }
    }
  }
	
	// Reduce function
  public static class AvroWordReducer extends Reducer<AvroKey<Text>, AvroValue<GenericRecord>,
      AvroKey<GenericRecord>, NullWritable>{       
    public void reduce(AvroKey<Text> key, Iterable<AvroValue<GenericRecord>> values, Context context) 
        throws IOException, InterruptedException {
      int sum = 0;
      for (AvroValue<GenericRecord> value : values) {
        GenericRecord    record = value.datum();
        sum += (Integer)record.get("count");
      }
      GenericRecord record = new    GenericData.Record(AVRO_SCHEMA);
      record.put("word", key.datum());
      record.put("count", sum);
      context.write(new AvroKey<GenericRecord>(record), NullWritable.get());
    }
  }

  public static void main(String[] args) throws Exception{
    int exitFlag = ToolRunner.run(new AvroWordCount(), args);
    System.exit(exitFlag);
  }
	

  @Override
  public int run(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "AvroWC");
    job.setJarByClass(getClass());
    job.setMapperClass(AvroWordMapper.class);    
    job.setReducerClass(AvroWordReducer.class);
    
    AvroJob.setMapOutputKeySchema(job, Schema.create(Schema.Type.STRING));
    AvroJob.setMapOutputValueSchema(job, AVRO_SCHEMA);
    AvroJob.setOutputKeySchema(job,	AVRO_SCHEMA);

    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(AvroKeyOutputFormat.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    return job.waitForCompletion(true) ? 0 : 1;
  }
}

创建jar之后，可以使用以下命令运行此Avro MapReduce程序。

hadoop jar /home/theitroad/theitroadhadoop.jar org.theitroad.AvroWordCount /user/input/count /user/out/result

该程序在只有两行的简单文本文件上执行。

This is a test file.
This is a Hadoop MapReduce program file.

可以使用avrotools.jar检查输出文件。

hadoop jar /path/to/avro-tools-1.8.2.jar tojson /user/out/result/part-r-00000.avro

{"word":"Hadoop","count":1}
{"word":"MapReduce","count":1}
{"word":"This","count":2}
{"word":"a","count":2}
{"word":"file.","count":2}
{"word":"is","count":2}
{"word":"program","count":1}
{"word":"test","count":1}