Writable接口与序列化机制

序列化概念

• 序列化（Serialization）是指把结构化对象转化为字节流。


• 反序列化（Deserialization）是序列化的逆过程。即把字节流转回结构化对象。


• Java序列化（java.io.Serializable）

Hadoop序列化的特点

• 序列化格式特点：
  
  
  
  
  • 紧凑：高效使用存储空间。
  
  
  • 快速：读写数据的额外开销小
  
  
  • 可扩展：可透明地读取老格式的数据
  
  
  • 互操作：支持多语言的交互
  
  
  
  


• Hadoop的序列化格式：Writable

Hadoop序列化的作用

• 序列化在分布式环境的两大作用：进程间通信，永久存储。


• Hadoop节点间通信。


• 

Writable接口

Writable接口, 是根据 DataInput 和 DataOutput 实现的简单、有效的序列化对象.MR的任意Key和Value必须实现Writable接口.

常用的Writable实现类

Text一般认为它等价于java.lang.String的Writable。针对UTF-8序列。

例:

Text test = new Text("test");
IntWritable One= new IntWritable(1);

自定义Writable类

• Writable
  
  
  
  
  • write 是把每个对象序列化到输出流
  
  
  • readFields是把输入流字节反序列化
  
  
  • 
  
  
  
  


• 实现WritableComparable.


• Java值对象的比较：一般需要重写toString(),hashCode(),equals()方法
  

   

  
  

MapReduce输入的处理类

• 1、FileInputFormat:
  
  
  
  
  • FileInputFormat是所有以文件作为数据源的InputFormat实现的基类，FileInputFormat保存作为job输入的所有文件，并实现了对输入文件计算splits的方法。至于获得记录的方法是有不同的子类——TextInputFormat进行实现的。
  
  
  
  


• 2、InputFormat：


• 
  

   

  
  
  
  
  • InputFormat 负责处理MR的输入部分.有三个作用:
    
    
    
    
    • 验证作业的输入是否规范.
    
    
    • 把输入文件切分成InputSplit.
    
    
    • 提供RecordReader 的实现类，把InputSplit读到Mapper中进行处理.
    
    
    
    
  
  
  
  


• 3、InputSplit:
  
  
  
  
  • 在执行mapreduce之前，原始数据被分割成若干split，每个split作为一个map任务的输入，在map执行过程中split会被分解成一个个记录（key-value对），map会依次处理每一个记录。
  
  
  • FileInputFormat只划分比HDFS block大的文件，所以FileInputFormat划分的结果是这个文件或者是这个文件中的一部分.
  
  
  • 如果一个文件的大小比block小，将不会被划分，这也是Hadoop处理大文件的效率要比处理很多小文件的效率高的原因。
  
  
  • 当Hadoop处理很多小文件（文件大小小于hdfs block大小）的时候，由于FileInputFormat不会对小文件进行划分，所以每一个小文件都会被当做一个split并分配一个map任务，导致效率底下。
  
  
  • 例如：一个1G的文件，会被划分成16个64MB的split，并分配16个map任务处理，而10000个100kb的文件会被10000个map任务处理。 
  
  
  
  

• 4、TextInputFormat:
  
  
  
  
  • TextInputformat是默认的处理类，处理普通文本文件。
  
  
  • 文件中每一行作为一个记录，他将每一行在文件中的起始偏移量作为key，每一行的内容作为value。
  
  
  • 默认以\n或回车键作为一行记录。
  
  
  • TextInputFormat继承了FileInputFormat。
  
  
  
  

InputFormat类的层次结构

其他输入类

• 1、CombineFileInputFormat
  
  
  
  
  • 相对于大量的小文件来说，hadoop更合适处理少量的大文件。
  
  
  • CombineFileInputFormat可以缓解这个问题，它是针对小文件而设计的。
  
  
  
  


• 2、KeyValueTextInputFormat
  
  
  
  
  • 当输入数据的每一行是两列，并用tab分离的形式的时候，KeyValueTextInputformat处理这种格式的文件非常适合。
  
  
  
  


• 3、NLineInputformat
  
  
  
  
  • NLineInputformat可以控制在每个split中数据的行数。
  
  
  
  


• 4、SequenceFileInputformat
  
  
  
  
  • 当输入文件格式是sequencefile的时候，要使用SequenceFileInputformat作为输入。
  
  
  
  

自定义输入格式

• 1、继承FileInputFormat基类。


• 2、重写里面的getSplits(JobContext context)方法。


• 3、重写createRecordReader(InputSplit split,TaskAttemptContext context)方法。

Hadoop的输出

• 1、TextOutputformat
  
  
  
  
  • 默认的输出格式，key和value中间值用tab隔开的。
  
  
  
  


• 2、SequenceFileOutputformat
  
  
  
  
  • 将key和value以sequencefile格式输出。
  
  
  
  


• 3、SequenceFileAsOutputFormat
  
  
  
  
  • 将key和value以原始二进制的格式输出。
  
  
  
  


• 4、MapFileOutputFormat
  
  
  
  
  • 将key和value写入MapFile中。由于MapFile中的key是有序的，所以写入的时候必须保证记录是按key值顺序写入的。
  
  
  
  


• 5、MultipleOutputFormat
  
  
  
  
  • 默认情况下一个reducer会产生一个输出，但是有些时候我们想一个reducer产生多个输出，MultipleOutputFormat和MultipleOutputs可以实现这个功能。
  
  
  
  

案例实现：

数据

      136315798506613726230503248124681200
      1363157995052138265441012640200
      1363157991076139264356561321512200
      1363154400022139262511062400200
      13631579930441821157596115272106200
      13631579950748413841341161432200
      1363157993055135604396581116954200
      13631579950331592013325731562936200
      1363157983019137191994192400200
      1363157984041136605779916960690200
      13631579730981501368585836593538200
      1363157986029159890021191938180200
      1363157992093135604396589184938200
      136315798604113480253104180180200
      13631579840401360284656519382910200

      13726230503248124681sum

DataBean类

    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;

    import org.apache.hadoop.io.Writable;

    public class DataBean implements Writable{

      //电话号码
      private String phone;
      //上行流量
      private Long upPayLoad;
      //下行流量
      private Long downPayLoad;
      //总流量
      private Long totalPayLoad;

      public DataBean(){}

      public DataBean(String phone,Long upPayLoad, Long downPayLoad) {
          super();
          this.phOne=phone;
          this.upPayLoad = upPayLoad;
          this.downPayLoad = downPayLoad;
          this.totalPayLoad=upPayLoad+downPayLoad;
      }

      /**
        * 序列化
        * 注意：序列化和反序列化的顺序和类型必须一致
        */
      @Override
      public void write(DataOutput out) throws IOException {
          // TODO Auto-generated method stub
          out.writeUTF(phone);
          out.writeLong(upPayLoad);
          out.writeLong(downPayLoad);
          out.writeLong(totalPayLoad);
      }

      /**
        * 反序列化
        */
      @Override
      public void readFields(DataInput in) throws IOException {
          // TODO Auto-generated method stub
          this.phOne=in.readUTF();
          this.upPayLoad=in.readLong();
          this.downPayLoad=in.readLong();
          this.totalPayLoad=in.readLong();
      }

      @Override
      public String toString() {
          return upPayLoad +"\t"+ downPayLoad +"\t"+ totalPayLoad;
      }

      public String getPhone() {
          return phone;
      }

      public void setPhone(String phone) {
          this.phOne= phone;
      }

      public Long getUpPayLoad() {
          return upPayLoad;
      }

      public void setUpPayLoad(Long upPayLoad) {
          this.upPayLoad = upPayLoad;
      }

      public Long getDownPayLoad() {
          return downPayLoad;
      }

      public void setDownPayLoad(Long downPayLoad) {
          this.downPayLoad = downPayLoad;
      }

      public Long getTotalPayLoad() {
          return totalPayLoad;
      }

      public void setTotalPayLoad(Long totalPayLoad) {
          this.totalPayLoad = totalPayLoad;
      }
    }

DataCount类

    import java.io.IOException;

    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    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;

    public class DataCount {

      public static void main(String[] args) throws IOException, ClassNotFoundException,
      InterruptedException {
          // TODO Auto-generated method stub
          Job job=Job.getInstance(new Configuration());

          job.setJarByClass(DataCount.class);

          job.setMapperClass(DataCountMapper.class);
          job.setMapOutputKeyClass(Text.class);
          job.setMapOutputValueClass(DataBean.class);
          FileInputFormat.setInputPaths(job, args[0]);

          job.setReducerClass(DataCountReducer.class);
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(DataBean.class);
          FileOutputFormat.setOutputPath(job, new Path(args[1]));

          job.waitForCompletion(true);
      }

      public static class DataCountMapper extends Mapper{

          @Override
          protected void map(LongWritable key, Text value,
          Mapper.Context context)
                throws IOException, InterruptedException {
            String hang=value.toString();
            String[] strings=hang.split("\t");
            String phOne=strings[1];
            long up=Long.parseLong(strings[2]);
            long down=Long.parseLong(strings[3]);
            DataBean dataBean=new DataBean(phone,up, down);

            context.write(new Text(phone), dataBean);
          }

      }

      public static class DataCountReducer extends Reducer{

          @Override
          protected void reduce(Text k2, Iterable v2,
          Reducer.Context context)
                throws IOException, InterruptedException {
            long upSum=0;
            long downSum=0;

            for(DataBean dataBean:v2){
                upSum += dataBean.getUpPayLoad();
                downSum += dataBean.getDownPayLoad();
            }

            DataBean dataBean=new DataBean(k2.toString(),upSum,downSum);

            context.write(new Text(k2), dataBean);
          }

      }
    }