MapReduce的序列化
序列化（Serialization）是指把结构化对象转化为字节流。
反序列化（Deserialization）是序列化的逆过程。把字节流转为结构化对象。
当要在进程间传递对象或持久化对象的时候，就需要序列化对象成字节流，反之当要将接收到或从磁盘读取的字节流转换为对象，就要进行反序列化。Java 的序列化（Serializable）是一个重量级序列化框架，一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系…），不便于在网络中高效传输；所以，hadoop 自己开发了一套序列化机制（ Writable），精简，高效。不用像 java 对象类一样传输多层的父子关系，需要哪个属性就传输哪个属性值，大大的减少网络传输的开销。
Writable是Hadoop的序列化格式，hadoop定义了这样一个Writable接口。一个类要支持可序列化只需实现这个接口即可。
public class BeanDemo implements Writable {
private long id;
private String desc;
//一定要有无参构造,不然反序列化的时候会报错
public BeanDemo() {
}
public BeanDemo(long id, String desc) {
this.id = id;
this.desc = desc;
}
/**
* 序列化方法
*
* @param out
* @throws IOException
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeLong(id);
out.writeUTF(desc);
}
/**
* 反序列化方法
*
* @param in
* @throws IOException
*/
@Override
public void readFields(DataInput in) throws IOException {
this.id = in.readLong();
this.desc = in.readUTF();
}
}
MapReduce的排序
如果我们需要用某个我们自定义的JavaBean中的某个字段进行结果的排序，那么就需要把这个JavaBean放到key中传输，因为在MapReduce的过程中一定会对key进行排序，而且我们可以自己定义排序的方式，一旦我们需要把JavaBean放到key中传输的话，那么这个JavaBean需要实现Comparable接口的compareTo方法：
public class BeanDemo implements Writable, Comparable<BeanDemo> {
private long id;
private String desc;
//一定要有无参构造,不然反序列化的时候会报错
public BeanDemo() {
}
public BeanDemo(long id, String desc) {
this.id = id;
this.desc = desc;
}
/**
* 序列化方法
*
* @param out
* @throws IOException
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeLong(id);
out.writeUTF(desc);
}
/**
* 反序列化方法
*
* @param in
* @throws IOException
*/
@Override
public void readFields(DataInput in) throws IOException {
this.id = in.readLong();
this.desc = in.readUTF();
}
/**
* @param o
* @return
*/
@Override
public int compareTo(BeanDemo o) {
//按照id倒序排
//虽然this.id 比o.id 大 依然返回-1 认为小 由于排序规则谁大谁在后 所以就形成了倒序
return this.id > o.id ? -1 : 1;
}
}
这样得出的结果就以id倒序排序了。
MapReduce的分区
如果有一种需求，需要将结果根据规则分别写到不同的文件中去，那么我们就需要多个reduce来执行，既然需要多个reduce那么就需要多个分区，让每个reduce拉取属于自己分区的数据进行操作和输出
Mapreduce 中会将 map 输出的 kv 对，按照相同 key 分组，然后分发给不同的 reducetask。
默认的分发规则为：根据 key 的 hashcode%reducetask 数来分发；
所以：如果要按照我们自己的需求进行分组，则需要改写数据分发（分组）组件 Partitioner，自定义一个 CustomPartitioner 继承抽象类：Partitioner，然后在job 对象中，设置自定义 partitioner： job.setPartitionerClass(CustomPartitioner.class)。
案例：以 上一篇的简单案例 为基础，在此基础上实现：a开头的写到一个文件中，b开头的写到一个文件中，其他的写到另外一个文件中
首先编写Partitioner类：
public class WordPartitioner extends Partitioner<Text, IntWritable> {
public static HashMap<String, Integer> map = new HashMap<String, Integer>();
static {
map.put("a", 0);
map.put("b", 1);
}
@Override
public int getPartition(Text key, IntWritable intWritable, int numPartitions) {
//获取每个词的第一个字母 在 map中对应的数字
Integer code = map.get(key.toString().substring(0, 1));
if (code != null) {
return code;
}
return 2;
}
}
修改执行类，修改reduce个数和添加自定义分区组件：
public class WordCountRunner {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
//指定mr采用本地模式运行 本地测试用
conf.set("mapreduce.framework.name", "local");
//使用job构建本次mr程序
Job job = Job.getInstance(conf);
//指定本次mr程序运行的主类
job.setJarByClass(WordCountRunner.class);
//指定本次mr程序的mapper reducer
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);
//指定本次mr程序map阶段的输出类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//指定本次mr程序reduce阶段的输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//设置使用几个Reduce执行 要和下面的WordPartitioner内的分区数相同
job.setNumReduceTasks(3);
//设置自定义分区组件
job.setPartitionerClass(WordPartitioner.class);
//指定本次mr程序处理的数据目录 输出结果的目录
// FileInputFormat.setInputPaths(job, new Path("/wordcount/input"));
// FileOutputFormat.setOutputPath(job, new Path("/wordcount/output"));
//本地测试用
FileInputFormat.setInputPaths(job, new Path("D:\\wordcount\\input"));
FileOutputFormat.setOutputPath(job, new Path("D:\\wordcount\\output"));//输出的文件夹不能提前创建 否则会报错
//提交本次mr的job
//job.submit();
//提交任务 并且追踪打印job的执行情况
boolean b = job.waitForCompletion(true);
System.exit(b ? 0 : -1);
}
}