Hadoop中WritableComparable 和 comparator

1.WritableComparable

查看HadoopAPI,如圖所示：

WritableComparable繼承自Writable和java.lang.Comparable接口，是一個Writable也是一個Comparable，也就是說，既可以序列化，也可以比較！

再看看它的實現類，發現BooleanWritable, BytesWritable, ByteWritable, DoubleWritable, FloatWritable, IntWritable, LongWritable, MD5Hash, NullWritable, Record, RecordTypeInfo, Text, VIntWritable, VLongWritable都實現了WritableComparable類！

WritableComparable的實現類之間相互來比較，在Map/Reduce中，任何用作鍵來使用的類都應該實現WritableComparable接口！

 

Example:

package cn.roboson.writable;

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

import org.apache.hadoop.io.WritableComparable;

/**
 * 1.自定義一個類，繼承WritableComparable
 * 2.發現有三個未實現的方法，兩個是Writable接口的（序列化），一個是Comparable接口的（用來比較）
 * 3.自定義比較，這里以counter來作為比較
 * @author roboson
 *
 */
public class MyWritableComparable implements WritableComparable<MyWritableComparable>{
    
    private int counter;
    private long timestamp;
    public MyWritableComparable() {
        // TODO Auto-generated constructor stub
    }

    public MyWritableComparable(int counter,long timestamp) {
        // TODO Auto-generated constructor stub
        this.counter = counter;
        this.timestamp = timestamp;
    }
    
    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        
        //將輸入流中的字節流數據轉化為結構化數據
        counter = in.readInt();
        timestamp = in.readLong();
    }

    @Override
    public void write(DataOutput out) throws IOException {
        // TODO Auto-generated method stub
        
        //講結構化數據寫入輸出流
        out.writeInt(counter);
        out.writeLong(timestamp);
    }

    @Override
    public int compareTo(MyWritableComparable other) {
        // TODO Auto-generated method stub
        int thisValue = this.counter;
        int otherValue = other.counter;
        return (thisValue < otherValue ? -1 : (thisValue == otherValue ? 0 : 1));
    }

    public int getCounter() {
        return counter;
    }

    public void setCounter(int counter) {
        this.counter = counter;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }
    

    public static void main(String[] args) {
        MyWritableComparable comparable = new MyWritableComparable(3,4);
        MyWritableComparable otherComparable = new MyWritableComparable(4, 5);
        int value = comparable.compareTo(otherComparable);
        if(value==-1){
            System.out.println("comparable<otherComparable");
        }else if(value==0){
            System.out.println("comparable=otherComparable");
        }else{
            System.out.println("comparable>otherComparable");
        }
    }
}

 

運行結果：

 

2.RawComparator

對於MapReduce來說，因為中間有個基於鍵的排序階段，所以類型的比較是非常重要的。Hadoop中提供了原生的比較接口RawComparator，該接口繼承子Java Comparator接口。RawComparator接口允許其實現直接比較數據流中的記錄，無需先把數據流飯序列化為對象，這樣便避免了新建對象的額外開銷。

package org.apache.hadoop.io;

import java.util.Comparator;

public interface RawComparator<T> extends Comparator<T>{
    
    //自己的方法
    public int compare(byte[] b1, int s1, int l1, byte[] b2,int s2, int l2);
    
    //繼承自Comparator的方法
    @Override
    public int compare(T o1, T o2);
    
    @Override
    public boolean equals(Object obj);
}

 

查看HadoopAPI：

該類並非被多數的衍生類所實現，其具體的子類為WritableComparator，多數情況下是作為實現Writable接口的類的內置類，提供序列化字節的比較。如下圖說所示：BooleanWritable, BytesWritable, ByteWritable, org.apache.hadoop.io.serializer.DeserializerComparator, DoubleWritable, FloatWritable, IntWritable, JavaSerializationComparator, LongWritable, LongWritable, MD5Hash, NullWritable， RecordComparator, Text, UTF8,都實現了RawComparator，作為其內部類。

而WritableComparator則是其的具體子類。

3.WritableComparator

在《Hadoop權威指南》中，說到這兒，很模糊，只說WritableComparator是對繼承自WritableComparable類的RawCompartor類的一個通用實現。讓人看着很迷惑，這句話什么意思呢？

首先、在第二個小標題RawComparator中，我門都知道WritableComparator實現了RawComparator這個接口，也就是說，WritableComparator是RawComparator的實現。

其次、是對繼承自WritableComparable類的RawComparator的一個通用實現。那么繼承自WritableComparable類的RawComparator都有哪些呢？也就是說那些類，繼承自WritableComparator，並且實現了RawComparator？在第二個小標題RawComparator中有也都說明清楚了，上面的紅色部分!同理，實現了WritableComparable類的在第一個小標題WritableComparable中也有說明，紅色部分字體！也就誰說WritableComparator是對BooleanWritable.Comparator, BytesWritable.Comparator, ByteWritable.Comparator, DoubleWritable.Comparator, FloatWritable.Comparator, IntWritable.Comparator, LongWritable.Comparator, MD5Hash.Comparator, NullWritable.Comparator, RecordComparator, Text.Comparator, UTF8.Comparator這些類的一個通用實現！這句話就引出了WritableComparator的兩個功能：第一，它提供了對原始compare()方法的一個默認實現。該方法能夠飯序列化將流中進行比較的對象，並調用對象的compara()方法。第二，它充當的是RawComparator實例的工廠（已注冊Writable的實現）。例如，為了獲得IntWratable的comparator，我們直接如下調用：

RawComparator<IntWritable> comparator = WritableComparator.get(IntWratable.class);

再來看看WritableComparator這個類是如何定義的，如下圖所示：

WritableComparator類類似於一個注冊表，里面記錄了所有Comparator類的集合。Comparators成員用一張Hash表記錄Key=Class，value=WritableComprator的注冊信息.這就是它能夠充當RawComparator實例工廠的原因!因為它本省的實現中有意個HashMap集合，HashMap<Class,WritableComparator>根據對應的Class，就能返回一個響應的WritableComparator!

 

Example:

package cn.roboson.writable;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparator;

/**
 * 1.通過WritableComparator獲得IntWritable類的RawComparator實例
 * 2.通過兩種方式來比較
 * @author roboson
 *
 */

public class ComparableFinish {
    
    public static void main(String[] args) throws IOException {
        
        //創建兩個IntWritable來比較
        IntWritable writable1 = new IntWritable(163);
        IntWritable writable2 = new IntWritable(165);
        
        //獲得IntWritable的RawComparator實例
        RawComparator<IntWritable> intRawComparator = WritableComparator.get(IntWritable.class);
        
        //直接比較對象
        int value1 =intRawComparator.compare(writable1, writable2);
        
        if(value1==-1){
            System.out.println("writable1<writable2");
        }else if(value1==0){
            System.out.println("writable1=writable2");
        }else{
            System.out.println("writable1>writable2");
        }
        
        //序列化兩個對象，獲得其字節流
        byte[] byte1 = serizlize(writable1);
        byte[] byte2 = serizlize(writable2);
        
        //直接通過字符流比較大小
        int value2 = intRawComparator.compare(byte1, 0, 4, byte2, 0, 4);
        if(value2==-1){
            System.out.println("writable1<writable2");
        }else if(value2==0){
            System.out.println("writable1=writable2");
        }else{
            System.out.println("writable1>writable2");
        }
    }
    
    public static byte[] serizlize(Writable writable) throws IOException{
        
        //創建一個輸出字節流對象
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        DataOutputStream dataout = new DataOutputStream(out);
        
        //將結構化數據的對象writable寫入到輸出字節流。
        writable.write(dataout);
        return out.toByteArray();
    }
    
    public static byte[] deserizlize(Writable writable,byte[] bytes) throws IOException{
        
        //創建一個輸入字節流對象，將字節數組中的數據，寫入到輸入流中
        ByteArrayInputStream in = new ByteArrayInputStream(bytes);
        DataInputStream datain = new DataInputStream(in);
        
        //將輸入流中的字節流數據反序列化
        writable.readFields(datain);
        return bytes;
        
    }
}

 

運行結果：

關於序列化方面的知識，可以參考我的博客《Hadoop序列化》地址如下：

http://www.cnblogs.com/robert-blue/p/4157768.html

參考博文：

http://blog.csdn.net/keda8997110/article/details/8518255

http://www.360doc.com/content/12/0827/09/9318309_232551844.shtml