1.Hadoop序列化和自定义实现序列化
序列化就是把内存中的对象，转换成字节序列，或这是其余传入协定，而后进行网络传输或者长久化到磁盘中。
反序列化就是一个相同的过程，将收到的字节序列或者磁盘中长久化的数据，转换成内存中的对象。
hadoop有一套本人的序列化机制，像后面提到的BooleanWritable、ByteWritable、IntWritable、FloatWritable、LongWritable、DoubleWritable、Text、MapWritable、ArrayWritable，这些都是hadoop实现好的序列化类型，咱们能够间接拿来应用。
但有些时候，这些根本的序列化类型不能满足咱们的需要，咱们就须要本人去实现一个序列化类型。
该怎么去实现？先看看hadoop中实现形式，就看IntWritable
public class IntWritable implements WritableComparable<IntWritable> 它实现WritableComparable这接口，那咱们自定义的类也能够实现这个接口，那么先看看这个接口有哪些要实现的办法
public interface WritableComparable<T> extends Writable, Comparable<T> {
}好难堪，这个接口啥都没有啊，那持续看它继承的接口，有两个。
第一个Writable接口
public interface Writable {
// 序列化办法
void write(DataOutput var1) throws IOException;
// 反序列化办法
void readFields(DataInput var1) throws IOException;
}两个要实现的办法，一个序列化办法和一个反序列化办法
而后咱们第二个接口
public interface Comparable<T> {
public int compareTo(T o);
}为什么要实现这个接口？
在MapReduce的shuffle过程要求key必须可能排序，所以如果自定义序列化类须要放在key中传输，那么就要实现这个接口中的办法，当然如果不须要，就能够略过了。
示例，实现一个能够序列化的StuInfoBean
public class StuInfoBean implements WritableComparable<StuInfoBean> {
private Integer stuid;
private String name;
private Integer age;
private Boolean sex;
public StuInfoBean(){}
public Integer getStuid() {
return stuid;
}
public void setStuid(Integer stuid) {
this.stuid = stuid;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
public Boolean getSex() {
return sex;
}
public void setSex(Boolean sex) {
this.sex = sex;
}
@Override
public String toString() {
return "StuInfoBean{" +
"stuid=" + stuid +
", name='" + name + '\'' +
", age=" + age +
", sex=" + sex +
'}';
}
/**
* 如果须要就实现，不须要就略过
*/
@Override
public int compareTo(StuInfoBean o) {
return this.stuid.compareTo(o.getStuid());
}
@Override
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeInt(stuid);
dataOutput.writeUTF(name);
dataOutput.writeInt(age);
dataOutput.writeBoolean(sex);
}
/**
* 这里要留神，反序列的程序要和序列化保持一致
*/
@Override
public void readFields(DataInput dataInput) throws IOException {
stuid = dataInput.readInt();
name = dataInput.readUTF();
age = dataInput.readInt();
sex = dataInput.readBoolean();
}
}2.排序
MapTask收集咱们的map()办法输入的kv对，放到内存缓冲区中，这是一个环形数据结构，其实就是一个字节数组，叫kvbuffer。
外面寄存着咱们的数据及索引数据kvmeta。
kvbuffer的大小尽管能够设置，但终归是无限的，当写满的时候，内存中的数据就会刷到磁盘上，这个过程就是溢写，溢写产生的文件可能会不止一个，多个溢出文件会被合并成大的溢出文件。
在这个过程中，都要调用Partitioner进行分区和针对key进行排序sort。
分区之前曾经说了，这里就说下排序sort。
MapTask和ReduceTask均会对数据依照key进行排序，任何应用程序中的数据不论业务逻辑是否须要，在hadoop中均会被排序。
默认是依照字典排序，实现形式是快排。
自定义类型排序的实现，就是后面compareTo办法。
@Override
public int compareTo(StuInfoBean o) {
return this.stuid.compareTo(o.getStuid());
}3.规约（Combiner）
这个也是合并，每一个 map 都可能会产生大量的本地输入，Combiner 的作用就是对 map 端的输入先做一次合并，以缩小在map 和 reduce节点之间的数据传输量，以进步网络IO性能，是MapReduce 的一种优化伎俩之一。
但有个前提就是不能扭转业务最终逻辑，Combiner的输入kv应该跟Reducer的输出kv类型要对应起来。
Combiner是MR程序中Mapper和Reducer之外的一种组件，它的父类就是Reduce。
举个例子
class1_aaa 50
class2_bbb 100
class3_ccc 80
class1_ddd 10
class2_eee 100
class3_fff 70
class1_hhh 150
class2_lll 100
class3_www 80需要求出每个属性值的和，那么咱们能够自定义MyCombiner，先部分求和，最初汇总到Reduce求和。
mapper代码
public class ClMap extends Mapper<LongWritable, Text,Text, IntWritable> {
// 输入的k和v
Text outk = new Text();
IntWritable outv = new IntWritable();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] contents = line.split(" ");
String outkey = contents[0].split("_")[0];
outk.set(outkey);
outv.set(Integer.parseInt(contents[1]));
context.write(outk,outv);
}
}MyCombiner代码
public class MyCombiner extends Reducer<Text, IntWritable,Text,IntWritable> {
private IntWritable v= new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable value : values) {
sum+=value.get();
}
v.set(sum);
context.write(key,v);
}
}Reduce代码
public class ClReduce extends Reducer<Text, IntWritable,Text,IntWritable> {
// private Text outk = new Text();
private IntWritable outv = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
for (IntWritable value : values) {
outv.set(value.get());
}
context.write(key,outv);
}
}Driver代码
public class ClDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(ClDriver.class);
job.setMapperClass(ClMap.class);
job.setReducerClass(ClReduce.class);
job.setMapOutputValueClass(IntWritable.class);
// 设置应用MyCombiner
job.setCombinerClass(MyCombiner.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
boolean b = job.waitForCompletion(true);
System.exit(b ? 0:1);
}
}
发表回复