Hadoop-2.4.1学习之RawComparator及其实现，hadoop2.4.1


      Hadoop支持对序列化的二进制流直接进行比较，相对于将序列化的二进制流反序列化对象再进行比较，显然前者具有更高的效率。而之所以需要对二进制流进行比较是因为Hadoop多个节点上的进程间通信是通过远程过程调用（RemoteProcedure Call Protocol，RPC）实现的，而RPC协议会将消息序列化为二进制流后再发送到远程节点，远程节点接收到二进制流后再反序列化为原始消息，如果对序列化的二进制流直接比较则会提高效率。本篇文章将学习Hadoop中基于字节的比较器RawComparator及其实现。

      RawComparator接口及其实现类的关系如下图所示。RawComparator接口继承自java.util.Comparator<T>，因此除了提供了基于字节比较的方法外，该接口还提供对象比较的方法。实际上，在RawComparator的很多实现类中，在基于字节比较的方法中，最终都调用了对象比较的方法。下面就对该接口及其实现做个简要的介绍。


      RawComparator：对对象的字节表示进行比较的接口，该接口的唯一方法是int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2)，其中b1为第一个对象所在的字节数组，s1为该对象在b1中的起始位置，l1为对象在b1中的长度，b2为第一个对象所在的字节数组，s2为该对象在b2中的起始位置，l2为对象在b2中的长度。

      DeserializerComparator<T>：抽象类，该类实现了RawComparator中的compare(byte[]b1, int s1, int l1, byte[] b2, int s2, int l2)，但未实现compare(T o1, T o2)。在该类中使用Deserializer将要比较的对象反序列化，然后使用标准的java.util .Comparator接口中的compare方法对反序列化后的对象进行比较。JavaSerializationComparator直接继承自DeserializerComparator，并且使用JavaSerialization.JavaSerializationDeserializer<T>对要比较的对象进行反序列化。这两个类在Hadoop很少使用，这是因为Java本身自带的序列化机制不如Hadoop实现的序列化机制高效。

      WritableComparator：该类直接实现RawComparator接口，用于比较实现了WritableComparables接口的对象。基本的实现是按照自然顺序进行比较，若想按照其它顺序比较，比如逆序，可以继承该类并覆盖compare(WritableComparable,WritableComparable)来实现定制的比较器。这是因为在针对字节的compare方法中，最终调用了compare(WritableComparable,WritableComparable)方法，因此该方法决定了按照什么方式进行比较。可以通过覆盖compare(byte[],int,int,byte[],int,int)优化比较操作，该类提供了许多静态方法以帮助该方法的优化实现。

      KeyFieldBasedComparator：该类继承自WritableComparator且实现了Configurable接口。该类提供了部分Unix/GNU排序的特性，这些特性有：-n（数字排序），-r（对比较结果逆序）等。

      接下来通过DeserializerComparator和WritableComparator的源代码来学习一下这两个RawComparator实现类是如何实现基于字节的比较方法的。首先是DeserializerComparator类，该类的实现代码如下：

@Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
try {
  //InputBuffer buffer,重置buffer读取的数据
      buffer.reset(b1, s1, l1);
  //T key1,T key2.Deserializer<T> deserializer
  //从底层输入流反序列化下一个对象。如果该方法的参数t不为null
  //那么该反序列化器可能设置它的内部状态为从输入流读取的下一个对象
  //如果t为null，新的反序列化器对象将被创建
      key1 = deserializer.deserialize(key1);
      buffer.reset(b2, s2, l2);
      key2 = deserializer.deserialize(key2);
    } catch (IOException e) {
      throw new RuntimeException(e);
}
//对反序列化后的对象进行比较
    return compare(key1, key2);
}

      在上面的代码中，使用Deserializer将byte[] b1中的字节反序列化为对象，然后再对对象进行比较。在该类的唯一子类JavaSerializationComparator中将Deserializer指定为JavaSerialization.JavaSerializationDeserializer<T>，后者实现了Deserializer接口，源代码如下：

private ObjectInputStream ois;
@Override
public void open(InputStream in) throws IOException {
   //创建ObjectInputStream对象，用于从输入流中反序列化对象
ois = new ObjectInputStream(in) {
        @Override
 protected void readStreamHeader() {
          // no header
        }
   };
}
@Override
@SuppressWarnings("unchecked")
public T deserialize(T object) throws IOException {
   try {
      //使用Java的反序列化机制从输入流中反序列化对象
      return (T) ois.readObject();
   } catch (ClassNotFoundException e) {
     throw new IOException(e.toString());
   }
}

      由上面的代码可知DeserializerComparator及其子类在底层实际调用ObjectInputStream反序列化对象，这就要求Deserializer反序列化的对象是由ObjectOutputStream序列化的。正如前面曾提到了Hadoop提供了比Java本身自带的序列化机制更高效的序列化机制，因此不推荐使用DeserializerComparator及其子类，而在hadoop的源代码中也未曾有使用它们的地方。

      在看过了基于Java的反序列化机制的比较器后，接下来学习一下hadoop中广泛使用的WritableComparator及其子类。WritableComparator中基于字节的比较方法代码如下：

//Hadoop中做为key的类型都必须实现WritableComparable接口
//该接口继承自Writable和Comparable
private final WritableComparable key1;
private final WritableComparable key2;
//从内存缓存中读取数据的DataInput，该类继承自DataInputStream
private final DataInputBuffer buffer;
@Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
    try {
      buffer.reset(b1, s1, l1);                   // parse key1
     //从buffer中反序列化key1对象的字段,调用了buffer的
      key1.readFields(buffer);
      buffer.reset(b2, s2, l2);                   // parse key2
      key2.readFields(buffer);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
    return compare(key1, key2);                   // compare them
}

      上面的compare方法中，本质上调用了DataInputStream中readXXX方法，比如对于key1为LongWritable来说，调用的就是readLong，对于IntWritable来说就是readInt。然而在众多WritableComparable的实现内部都定义了WritableComparator的子类，比如在IntWritable中，WritableComparator的子类中的定义如下。IntWritable中的WritableComparator子类覆盖了父类的compare方法（即上段代码）：

/** A Comparator optimized for IntWritable. */ 
  public static class Comparator extends WritableComparator {
    public Comparator() {
      super(IntWritable.class);
    }
    @Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
  // WritableComparator中的静态方法readInt
      int thisValue = readInt(b1, s1);
      int thatValue = readInt(b2, s2);
      return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
    }
  }

      readInt方法的代码如下：

/** Parse an integer from a byte array. */
  public static int readInt(byte[] bytes, int start) {
    // bytes[start  ]*224 + bytes[start+1]*216+ bytes[start+2]*28+ bytes[start+3]
    return (((bytes[start  ] & 0xff) << 24) + ((bytes[start+1] & 0xff) << 16) +
            ((bytes[start+2] & 0xff) <<  8) + ((bytes[start+3] & 0xff)));
  }

      而Java中DataInputStream的readInt方法如下，具体到DataInputBuffer中，in为ByteArrayInputStream的子类DataInputBuffer.Buffer。

public final int readInt() throws IOException {
        int ch1 = in.read();
        int ch2 = in.read();
        int ch3 = in.read();
        int ch4 = in.read();
        if ((ch1 | ch2 | ch3 | ch4) < 0)
            throw new EOFException();
        return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
}

      对比上面两段代码不难发现，IntWritable中的readInt要比Java原始的readInt更加高效，因为Java中的readInt需要操作底层数据流，而hadoop中的readInt只需要操作内存中的字节数组。除了上述的readInt外WritableComparator中还有readLong、readFloat、readDouble等优化过的方法，可以通过这些方法实现自定义的WritableComparator。

相关内容