Hadoop——序列化

快来打我* 2022-07-16 15:11 225阅读 0赞

序列化是指将结构化对象转化为字节流以便在网络上传输或写到磁盘进行永久存储的过程。反序列化是指将字节流传回结构化对象的逆过程。  
序列化在分布式数据处理的两大领域经常出现：进程间通信和永久存储。  
在Hadoop中，系统中多个节点上进程间的通信是通过“远程过程调用”(RPC)实现的。RPC协议将消息序列化成二进制流后发送到远程节点，远程节点接着将二进制流反序列化为原始消息。  
RPC序列化格式特点：

1、 紧凑：紧凑格式能充分利用网络带宽
    2、快速：进程间通信形成了分布式系统的顾家，所以需要尽量减少序列化和反序列化的性能开销，这是最基本的。
    3、可扩展：为了满足新的需求，协议不断变化，所以控制客户端和服务器的过程中，需要直接引进相应的协议。
    4、支持互操作：对于某些系统来说，协议能支持不同语言的客户端与服务器交互。所以需要设计需要一种特定的格式来满足这一需求。

一、Writable接口  
Writable接口定义两个方法：一个将其状态写到DataOutput二进制流，另一个从DataInput二进制流读取数据。

import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public interface Writable { 
                 void write(DataOutput output) throws IOException;
                 void readLine(DataInput input) throws IOException;
    }

WritableComparable接口和comparator  
IntWritable实现原始的WritableComparable接口，该接口继承自Writable和java.lang.Comparable接口：

import org.apache.hadoop.io.Writable;
    public interface WritableCompatable<T> extends Writable, Comparable<T> { 
    }

Hadoop提供了一个优化接口是继承自Java Comparator的RawComparator接口：

import java.util.Comparator;
    public interface RawComparator<T> extends Comparator<T>{ 
                /** * 该方法可以从每个字节数组b1和b2读取给定起始位置(s1和s2)以及长度(l1和l2)的一个整数进而直接进行比较 */
                 public int compare(byte[] b1,int s1,int l1,byte[] b2, int s2, int l2);
    }

该接口允许其实现直接比较数据流中的记录，无须先把数据流反序列化为对象。  
WritableComparator是对继承自WritableComparable类的RawComparator类的一个通用实现。他提供两个主要功能：

1、他提供了对原始compare()方法的一个默认实现，该方法能够反序列化将在流中进行比较的对象，并调用对象的compare()方法。
    2、它充当的是RawComparator实例的工厂。

获得IntWritable的comparator：

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.WritableComparator;
    
    public class WritableTest {
                RawComparator<IntWritable> comparator=WritableComparator.get(IntWritable.class) ;
    }

二、Writable类  
1、java基本类型的Writable封装器  
Writable类对Java基本类型提供封装，char除外（可以存储在IntWritable中）。所有的封装都包含get()和set()方法用于读取或存储封装的值。  
![这里写图片描述][SouthEast]

变长格式一般更省空间。变长编码的优点在于可以在VintWritable和VlongWritable转换，他们的编码是一致。  
2、Text类型  
Text是针对UTF-8序列的Writable类。一般可以认为等价于java.lang.String的Writable。不支持对字节数超过32767的字符串进行编码，它使用的是Java的UTF-8修订版。  
Text类型使用整型来存储字符串编码所需要的字节数，因此该最大值是2G。  
（1）、索引：charAt()方法返回的是一个表示Unicode编码位置的int类型值。  
find()方法类似java.lang.String中的indexOf()方法，返回字节的偏移量。  
（2）、迭代：将Text对像转换为java.nio.ByteBuffer对象，然后利用缓冲区对Text对象反复调用bytesToCodePoint()静态方法。该方法能够获取下一代码的位置并返回int值，最后更新缓冲区的位置。当bytesToCodePoiint()返回-1时，则检测到字符串的末尾。

import java.nio.ByteBuffer;
    import org.apache.hadoop.io.Text;
    
    public class TextIterator { 
                 public static void main(String[] args) {
                             // TODO Auto-generated method stub
                            Text text= new Text( "\u0041\u00DF\u6771\uD801\uDC00" );
                            ByteBuffer buffer=ByteBuffer. wrap(text.getBytes(), 0, text.getLength());
                             int cp;
                             while(( buffer.hasRemaining()&&( cp=Text. bytesToCodePoint(buffer))!=-1)){
                                        System. err.println(Integer.toHexString( cp));
                            }
                }
    }

（3）、 可变性：Text是可变，通过调用其中一个set()方法重用Text实例。  
（4）、对String重新排序：Text对象通过调用toString()方法转换为String对象。

3、ByteWritable  
ByteWritable是对二进制数据数组的封装。它的序列化格式为一个指定所含数据字节数的整数域，后跟数据内容本身。  
例如：长度为2的字节数组包含数值3和5，序列化形式为一个4字节的整数(00000002)和该数组中的两个字节(03和05)；  
ByteWritable是可变的，其值可以通过set()方法进行修改。可以通过getLength()方法确定ByteWritable的大小。  
4、NullWritable  
NullWritable是Writable的特殊类型，它的序列化长度为0.它并不从流中读取数据，也不写入数据，他充当占位符。在MapReduce中如果不需要使用键或值的序列化地址，就可以将键和值声明为NullWritable，结果是搞笑的存储常量空值。NullWritable也可以用作在SequenceFile中的值。它是一个不可变的单实例类型，通过调用NullWritable.get(方法可以获取这个实例。  
5、ObjectWritable和GenericWritable  
ObjectWritable是对Java基本类型(String,enum,Writable,null或这些类型组成的数组)的一个通用封装。他在Hadoop RPC中用于对方法的参数和返回类型进行封装和解封装。  
GenericWritable使用静态类型的数组，并使用对序列化后的类型的引用加入位置索引来提高性能。故在的子类中需要指定什么类型。  
6、Writable集合类  
org.apache.hadoop.io软件包中一共有6个Writable集合类：ArrayWritable、ArrayPrimitiveWritable、TwoDArrayWritable、MapWritable、SortedMapWritable和EnumMapWritable。  
其中：  
（1）ArrayWritable和TwoDArrayWritable是对Writbale的数组和两维数组（数组的数组）的实现。ArrayWritable或TwoDArrayWritable中所有的元素必须是同一类（在构造函数中指定）的实例。

ArrayWritable writable=new ArrayWritable(Text.class);

如果Writable按照类型来定义，例如SequenceFile的键或值，或更多时候作为MapReduce的输入，则需要继承ArrayWritable(或相应的TwoDArrayWritable Writable类)并设置静态类型。  
ArrayWritable和TwoDArrayWritable都有get()、set()和toArray()方法。toArray()方法用于新建该数组(或二维数组)的一个浅拷贝。

（1） ArrayPrimitiveWritable是对Java基本数组类型的一个封装。调用set()方法时，可以识别响应组件类型，因此无需通过继承该类来设置类型。  
（2） MapWritable和SortMapWritable分别实现了java.util.Map

import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.Text;
    
    /** * 存储一对Text对象的Writable * @author Administrator * */
    public class TextPair implements WritableComparable<TextPair> {
    
                 private Text first ;
                 private Text second ;
    
                 public TextPair(){
                            set( new Text(), new Text());
                }
    
                 public TextPair(String first ,String second ){
                            set( new Text( first), new Text( second));
                }
    
                 public TextPair(Text first ,Text second ){
                            set( first, second);
                }
    
                 public void set(Text first ,Text second ){
                             this. first= first;
                             this. second= second;
                }
    
                 public Text getFirst() {
                             return first ;
                }
    
                 public Text getSecond() {
                             return second ;
                }
    
                 @Override
                 public void readFields(DataInput input) throws IOException {
                             // TODO Auto-generated method stub
                             first.readFields( input);
                             second.readFields( input);
                }
    
                 @Override
                 public void write(DataOutput out) throws IOException {
                             // TODO Auto-generated method stub
                             first.write( out);
                             first.write( out);
                }
    
                 @Override
                 public int hashCode() {
                             return first .hashCode()*163+second.hashCode();
                };
    
    
    
                 @Override
                 public boolean equals(Object obj) {
                             if ( obj instanceof TextPair) {
                                        TextPair tp=(TextPair) obj;
                                         return first .equals(tp .first )&&second .equals(tp .second );
                            }
                             return false ;
                }
    
                 @Override
                 public String toString() {
                             return "TextPair [first=" + first + ", second=" + second + "]" ;
                }
                 @Override
                 public int compareTo(TextPair tp) {
                             // TODO Auto-generated method stub
                             int cmp= first.compareTo( tp. first);
                             if( cmp!=0){
                                         return cmp ;
                            }
                             return second .compareTo(tp .second );
                }
    
    }

这个定制Writable实现的第一部分非常直观：包括两个Text实例变量(first和second)和相关的构造函数，以及setter和getter（即设置函数和提取函数）。所有Writable实现都必须有一个默认的构造函数以便MapReduce框架可以对他们进行实例化，然后再调用readFields()函数查看各个字段的值。Writable实例是可变的并且通常可以重用，所以应该尽量避免在Writable或readFields()方法中分配对象。  
TextPair类的write()方法依次每个Text对象序列化到输出流中。  
HashPartitioner（MapReduce中的默认分区类）通常用hashCode()方法来选择reduce分区。  
TextPair是WritableComparable的一个实现，它提供了compareTo()方法，该方法可以强制数据排序：先按照第一个字符排序，如果第一个字符相同则按照第二个字符排序。TextArrayWritable只继承Writable。  
TextPair中的两个Text对象连接而成。所以可以通过Text的RawComparator方法进行比较。

[SouthEast]: /images/20220709/eacbedc10f2d4f13861aff49038f4639.png