【Avro三】Hadoop MapReduce读写Avro文件

xinxuaw231

　　Avro是Doug Cutting(此人绝对是神一般的存在）牵头开发的。 开发之初就是围绕着完善Hadoop生态系统的数据处理而开展的（使用Avro作为Hadoop MapReduce需要处理数据序列化和反序列化的场景）,因此Hadoop MapReduce集成Avro也就是自然而然的事情。
　　这个例子是一个简单的Hadoop MapReduce读取Avro格式的源文件进行计数统计，然后将计算结果作为Avro格式的数据写到目标文件中，主要目的是体会下Hadoop MapReduce操作Avro的基本流程和Avro提供的API

1. Maven依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>learn</groupId>
<artifactId>learn.avro</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<!--avro core-->
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>1.7.7</version>
</dependency>
<!--avro rpc support-->
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro-ipc</artifactId>
<version>1.7.7</version>
</dependency>
<!--avro utilities for Hadoop MapReduce to process avro files -->
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro-mapred</artifactId>
<version>1.7.7</version>
</dependency>
<!--Avro and Hadoop Map Reduce-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-core</artifactId>
<version>1.2.1</version>
</dependency>

</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>1.7.7</version>
<executions>
<execution>
<phase>generate-sources</phase>
<goals>
<goal>schema</goal>
<goal>protocol</goal>
<goal>idl-protocol</goal>
</goals>
<configuration>
<sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
<outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.7</source>
<target>1.7</target>
</configuration>
</plugin>
</plugins>
</build>
</project>

 

2. MapReduce代码：

package examples.avro.mapreduce;
import examples.avro.simple.User;
import org.apache.avro.Schema;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyInputFormat;
import org.apache.avro.mapreduce.AvroKeyValueOutputFormat;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.IOException;
public class MapReduceColorCount extends Configured implements Tool {
///Mapper定义：
///输入Key类型是AvroKey<User>，输入Value类型是NullWritable
///输出Key类型是Text，输出Value类型是IntWritable
public static class ColorCountMapper extends
Mapper<AvroKey<User>, NullWritable, Text, IntWritable> {
@Override
public void map(AvroKey<User> key, NullWritable value, Context context)
throws IOException, InterruptedException {
CharSequence color = key.datum().getFavoriteColor();
if (color == null) {
color = "none";
}
context.write(new Text(color.toString()), new IntWritable(1));
}
}
///Reducer定义：
///输入Key类型是Text，输入Value类型是IntWritable(跟Key的输出Key/Value类型一致)
///输出Key类型是AvroKey<CharSequence>，输出Value类型是AvroValue<Integer>
public static class ColorCountReducer extends
Reducer<Text, IntWritable, AvroKey<CharSequence>, AvroValue<Integer>> {
@Override
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable value : values) {
sum += value.get();
}
context.write(new AvroKey<CharSequence>(key.toString()), new AvroValue<Integer>(sum));
}
}
public int run(String[] args) throws Exception {
if (args.length != 2) {
System.err.println("Usage: MapReduceColorCount <input path> <output path>");
return -1;
}
Job job = new Job(getConf());
job.setJarByClass(MapReduceColorCount.class);
job.setJobName("Color Count");
///指定输入路径，输入文件是Avro格式
FileInputFormat.setInputPaths(job, new Path(args[0]));
///指定输出路径，输出文件格式是Key/Value组成的Avro文件，见AvroKeyValueOutputFormat
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//AvroKeyInputFormat: A MapReduce InputFormat that can handle Avro container files.
job.setInputFormatClass(AvroKeyInputFormat.class);
job.setMapperClass(ColorCountMapper.class);
AvroJob.setInputKeySchema(job, User.getClassSchema());
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//AvroKeyValueOutputFormat: FileOutputFormat for writing Avro container files of key/value pairs
job.setOutputFormatClass(AvroKeyValueOutputFormat.class);
job.setReducerClass(ColorCountReducer.class);
AvroJob.setOutputKeySchema(job, Schema.create(Schema.Type.STRING));
AvroJob.setOutputValueSchema(job, Schema.create(Schema.Type.INT));
return (job.waitForCompletion(true) ? 0 : 1);
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new MapReduceColorCount(), args);
System.exit(res);
}
}

3. 主要类注释
　　3.1 AvroKey

/** The wrapper of keys for jobs configured with {@link AvroJob} . */
　　3.2 AvroValue

/** The wrapper of values for jobs configured with {@link AvroJob} . */
　　3.3 AvroJob

/** Setters to configure jobs for Avro data. */
　　3.4 AvroKeyInputFormat

/**
* A MapReduce InputFormat that can handle Avro container files.
*
* <p>Keys are AvroKey wrapper objects that contain the Avro data.  Since Avro
* container files store only records (not key/value pairs), the value from
* this InputFormat is a NullWritable.</p>
*/
　　3.5 AvroKeyValueOutputFormat

/**
* FileOutputFormat for writing Avro container files of key/value pairs.
*
* <p>Since Avro container files can only contain records (not key/value pairs), this
* output format puts the key and value into an Avro generic record with two fields, named
* 'key' and 'value'.</p>
*
* <p>The keys and values given to this output format may be Avro objects wrapped in
* <code>AvroKey</code> or <code>AvroValue</code> objects.  The basic Writable types are
* also supported (e.g., IntWritable, Text); they will be converted to their corresponding
* Avro types.</p>
*
* @param <K> The type of key. If an Avro type, it must be wrapped in an <code>AvroKey</code>.
* @param <V> The type of value. If an Avro type, it must be wrapped in an <code>AvroValue</code>.
*/
　　3.6

  /**
* Sets the job input key schema.
*
* @param job The job to configure.
* @param schema The input key schema.
*/
public static void setInputKeySchema(Job job, Schema schema) {
job.getConfiguration().set(CONF_INPUT_KEY_SCHEMA, schema.toString());
}
/**
* Sets the job input value schema.
*
* @param job The job to configure.
* @param schema The input value schema.
*/
public static void setInputValueSchema(Job job, Schema schema) {
job.getConfiguration().set(CONF_INPUT_VALUE_SCHEMA, schema.toString());
}
　　3.7

/**
* Sets the map output key schema.
*
* @param job The job to configure.
* @param schema The map output key schema.
*/
public static void setMapOutputKeySchema(Job job, Schema schema) {
job.setMapOutputKeyClass(AvroKey.class);
job.setGroupingComparatorClass(AvroKeyComparator.class);
job.setSortComparatorClass(AvroKeyComparator.class);
AvroSerialization.setKeyWriterSchema(job.getConfiguration(), schema);
AvroSerialization.setKeyReaderSchema(job.getConfiguration(), schema);
AvroSerialization.addToConfiguration(job.getConfiguration());
}
/**
* Sets the map output value schema.
*
* @param job The job to configure.
* @param schema The map output value schema.
*/
public static void setMapOutputValueSchema(Job job, Schema schema) {
job.setMapOutputValueClass(AvroValue.class);
AvroSerialization.setValueWriterSchema(job.getConfiguration(), schema);
AvroSerialization.setValueReaderSchema(job.getConfiguration(), schema);
AvroSerialization.addToConfiguration(job.getConfiguration());
}