022_Hadoop中的數據類型（Writable、WritableComparable、Comparator、RawComparator…）

1、 在hadoop中所有的key/value都必須實現Writable接口，有兩個方法，分別用於讀（反序列化）和寫（序列化）操作。

參考代碼：

package org.dragon.hadoop.mapreduce.app;

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

import org.apache.hadoop.io.Writable;

/**
 * 
 * @author ZhuXY
 * @time 2016-3-10 下午3:49:55
 * 
 */
public class DataWritable implements Writable {

    // telsphone

    // upload
    private int upPackNum;
    private int upPayLoad;

    // download
    private int downPackNum;
    private int downPayLoad;

    public DataWritable() {

    }

    public void set(int upPackNum, int upPayLoad, int downPackNum,
            int downPayload) {
        this.upPackNum = upPackNum;
        this.upPayLoad = upPayLoad;
        this.downPackNum = downPackNum;
        this.downPayLoad = downPayload;

    }

    public int getUpPackNum() {
        return upPackNum;
    }

    public int getUpPayLoas() {
        return upPayLoad;
    }

    public int getDownPackNum() {
        return downPackNum;
    }

    public int getDownPayload() {
        return downPayLoad;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(upPackNum);
        out.writeInt(upPayLoad);
        out.writeInt(downPackNum);        
        out.writeInt(downPayLoad);
    }

    /**
     * 讀出的順序要和寫入的順序相同
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        this.upPackNum = in.readInt();
        this.upPayLoad = in.readInt();
        this.downPackNum = in.readInt();
        this.downPayLoad = in.readInt();
    }

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

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + downPackNum;
        result = prime * result + downPayLoad;
        result = prime * result + upPackNum;
        result = prime * result + upPayLoad;
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        DataWritable other = (DataWritable) obj;
        if (downPackNum != other.downPackNum)
            return false;
        if (downPayLoad != other.downPayLoad)
            return false;
        if (upPackNum != other.upPackNum)
            return false;
        if (upPayLoad != other.upPayLoad)
            return false;
        return true;
    }

}

 

2、所有的key必須實現Comparable接口，在MapReduce過程中需要對Key/Value對進行反復的排序。默認情況下依據Key進行排序的，要實現comparaTo()方法。所以通過Key既要實現Writable接口又要實現Comparable接口，Hadoop中提供了一個公共的接口，叫做WritableComparable接口：

3、由於需要序列化反序列化和進行比較，對java對象需要重寫一下幾個方法：

①    equals();

②    hashCode();

③    toString()方法

如IntWritable類型的實現：

package org.apache.hadoop.io;

import java.io.*;

/** A WritableComparable for ints. */
public class IntWritable implements WritableComparable {
  private int value;

  public IntWritable() {}

  public IntWritable(int value) { set(value); }

  /** Set the value of this IntWritable. */
  public void set(int value) { this.value = value; }

  /** Return the value of this IntWritable. */
  public int get() { return value; }

  public void readFields(DataInput in) throws IOException {
    value = in.readInt();
  }

  public void write(DataOutput out) throws IOException {
    out.writeInt(value);
  }

  /** Returns true iff <code>o</code> is a IntWritable with the same value. */
  public boolean equals(Object o) {
    if (!(o instanceof IntWritable))
      return false;
    IntWritable other = (IntWritable)o;
    return this.value == other.value;
  }

  public int hashCode() {
    return value;
  }

  /** Compares two IntWritables. */
  public int compareTo(Object o) {
    int thisValue = this.value;
    int thatValue = ((IntWritable)o).value;
    return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
  }

  public String toString() {
    return Integer.toString(value);
  }

4、數據類型，必須有一個無參的構造方法，為了方便反射創建對象。

在自定義數據類型中，建議使用java原生數據類型，最好不要使用hadoop對原生類型封裝好的數據類型，即如下樣例代碼：

推薦使用：

不建議使用：

5、問題：

　　當數據寫入磁盤時，如果要進行排序的話，需要首先從磁盤中讀取數據進行反序列化成對象，然后在內存中對反序列化的對象進行比較。

　　對字節（未經過反序列化字節）進行直接比較，不需要進行反序列化以后再比較呢？如果要實現上述功能，Hadoop數據類型需要實現一個接口RawComparator。

　　在Hadoop中有一個針對Writable數據類型，進行實現的一個通用實現類WritableComparator類。所有的數據類型，只需要繼承通用類，再去需要具體功能復寫相應的compara()方法。一下以IntWritable為例，查看一下：

    對於自定義的Comparator類需要以下幾步：

1） 推薦Comparator類定義在數據類型內部，靜態內部類，實現WritableComparator類。

　　2） 重寫默認無參構造方法，方法內必須調用父類有參構造方法，如下截圖：

3） 重載父類的compare()方法，依據具體功能覆寫。

4） 向WritableComparator類中注冊自定義的Comparator類，代碼如下：

 

5、自定義數據類型

　　樣例代碼：

package org.dragon.hadoop.mr.io;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;

/**
 * 自定義數據類型對Writable的實現。
 * 快捷鍵get、set選擇器alt+shift+s
 * @author ZhuXY
 * @time 2016-3-9 下午10:40:02
 * 
 */
/**
 * 1、Hadoop之——數據類型
    1）    在hadoop中所有的key/value都必須實現Writable接口，有兩個方法，分別用於讀（反序列化）和寫（序列化）操作。
    2）    所有的key必須實現Comparable接口，在MapReduce過程中需要對Key/Value對進行反復的排序。默認情況下依據Key進行排序的，要實現comparaTo()方法。所以通過Key既要實現Writable接口又要實現Comparable接口，Hadoop中提供了一個公共的接口，叫做WritableComparable接口
     3）    由於需要序列化反序列化和進行比較，對java對象需要重寫一下幾個方法：
        ①    equals();
        ②    hashCode();
        ③    toString()方法
    4）    數據類型，必須有一個無參的構造方法，為了方便反射創建對象。
    5）    在自定義數據類型中，建議使用java原生數據類型，最好不要使用hadoop對原生類型封裝好的數據類型，即
     
 */

/**
 * 問題：
    當數據寫入磁盤時，如果要進行排序的話，需要首先從磁盤中讀取數據進行反序列化成對象，然后在內存中對反序列化的對象進行比較。
 * 對字節（未經過反序列化字節）進行直接比較，不需要進行反序列化以后再比較呢？如果要實現上述功能，Hadoop數據類型需要實現一個接口RawComparator。
            在Hadoop中有一個針對Writable數據類型，進行實現的一個通用實現類WritableComparator類。所有的數據類型，只需要繼承通用類，再去需要具體功能復寫相應的compara()方法。
     對於自定義的Comparator類需要以下幾步：
    1）    推薦Comparator類定義在數據類型內部，靜態內部類，實現WritableComparator類。
    2）    重寫默認無參構造方法，方法內必須調用父類有參構造方法，如下截圖：
     
    3）    重載父類的compare()方法，依據具體功能覆寫。
    4）    向WritableComparator類中注冊自定義的Comparator類，代碼如下：

 */

/**
 * WritableCOmparator是RawComparator對WritableComparable類的一個通用實現。它提供兩個主要的功能。
 * 首先他提供了一個默認的對原始compare()函數的調用，對從數據流對要比較的對象進行反序列化，然后調用對象
 * 的compare方法。
 * 其次，他充當的是RawComparator實例的工廠方法（Writable方法已經注冊）。
 * @author ZhuXY
 *
 */
public class PairWritable implements WritableComparable<PairWritable> {

    private String name;// Text
    private Integer age;// IntWritale

    public PairWritable() {
    }

    public PairWritable(String name, Integer age) {
        this.set(name, age);
    }

    public void set(String name, Integer age) {
        this.name = name;
        this.age = age;
    }

    public String getName() {
        return name;
    }

    public Integer getAge() {
        return age;
    }

    /**
     * write方法是在寫入數據時調用，進行序列化
     */
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(name);
        out.writeInt(age);
    }

    /**
     * readField()方法是在取出數據時調用的方法，反序列化方法
     * 以便生成對象
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        this.name = in.readUTF();
        this.age = in.readInt();
    }
    
    /**
     *   
        hashCode 的常規協定是：   
        1）在 Java 應用程序執行期間，在同一對象上多次調用 hashCode 方法時，必須一致地返回相同的整數，前提是對象上 equals 比較中所用的信息沒有被修改。從某一應用程序的一次執行到同一應用程序的另一次執行，該整數無需保持一致。   
        2）如果根據 equals(Object) 方法，兩個對象是相等的，那么在兩個對象中的每個對象上調用 hashCode 方法都必須生成相同的整數結果。   
        3）以下情況不 是必需的：如果根據 equals(java.lang.Object) 方法，兩個對象不相等，那么在兩個對象中的任一對象上調用 hashCode 方法必定會生成不同的整數結果。但是，程序員應該知道，為不相等的對象生成不同整數結果可以提高哈希表的性能。   
        4）實際上，由 Object 類定義的 hashCode 方法確實會針對不同的對象返回不同的整數。（這一般是通過將該對象的內部地址轉換成一個整數來實現的，但是 JavaTM 編程語言不需要這種實現技巧。）   
          
        5）當equals方法被重寫時，通常有必要重寫 hashCode 方法，以維護 hashCode 方法的常規協定，該協定聲明相等對象必須具有相等的哈希碼。  
     */
    @Override
    public int hashCode() {
        return name.hashCode() * 31 + age.hashCode();
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof PairWritable) {
            PairWritable pairWritable = (PairWritable) obj;

            return this.name.equals(pairWritable.getName())
                    && this.age.equals(pairWritable.getAge());
        }
        return false;
    }
    
    @Override
    public String toString() {
        // TODO Auto-generated method stub
        return this.name+"\t"+this.age;
    }
    
    @Override
    public int compareTo(PairWritable o) {
        int cmp=this.name.compareTo(o.getName());
         if (cmp!=0) {
            return cmp;
        }
         return this.age.compareTo(o.getAge());
    }
    
    public static class Comparator extends WritableComparator{
        
        public Comparator(){
            super(PairWritable.class);
        }
        
        /**
         * 第一個字節數組
         * byte[] b1, int s1, int l1, 
         * 字節數組起始位置長度
         * 
         * 第二個字節數組
         * byte[] b2, int s2, int l2
         * 字節數組的起始位置長度
         */
        /**
         * 
         * 核心：
         *         這個接口允許執行者比較從流中讀取的未被反序列化為對象的記錄，從而省去了創建對象的所有開銷。
         *         例如，IntWritables的comparator使用原始的compare()方法從每個字節數組的指定
         *         開始位置（S1和S2）和長度（L1和L2）讀取整數（b1和b2），然后直接進行比較。
         */
        @Override
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
            int n1=WritableUtils.decodeVIntSize(b1[s1]);
            int n2=WritableUtils.decodeVIntSize(b2[s2]);
            
            int cmp=WritableComparator.compareBytes(b1, s1+n1, l1-n1, b2, s2+n2, l2+n2);
            
            if (0!=cmp) {
                return cmp;
            }
            
            int thisValue=readInt(b1, l1-s1-n1);
            int thatValue=readInt(b2, l2-s2-n2);
            
            return (thisValue<thatValue ?-1:(thisValue==thatValue?0:1));
        }
        static {
            WritableComparator.define(PairWritable.class, new Comparator());
        }
    }
}

　　通常情況下，實現一個靜態方法read(DataInput)，用於構造數據類型的實例對象，方法內部調用readFields(DataInput)方法。

    Hadoop MapReduce Data Type中所有的Key，必須實現WritableComparable接口，官方文檔說明如下：

　　比較器RawComparator，官方文檔說明如下：

 

6、注意NullWritable類型

package org.apache.hadoop.io;

import java.io.*;

/** Singleton Writable with no data. */
public class NullWritable implements WritableComparable {

  private static final NullWritable THIS = new NullWritable();

  private NullWritable() {}                       // no public ctor

  /** Returns the single instance of this class. */
  public static NullWritable get() { return THIS; }
  
  public String toString() {
    return "(null)";
  }

  public int hashCode() { return 0; }
  public int compareTo(Object other) {
    if (!(other instanceof NullWritable)) {
      throw new ClassCastException("can't compare " + other.getClass().getName() 
                                   + " to NullWritable");
    }
    return 0;
  }
  public boolean equals(Object other) { return other instanceof NullWritable; }
  public void readFields(DataInput in) throws IOException {}
  public void write(DataOutput out) throws IOException {}

  /** A Comparator "optimized" for NullWritable. */
  public static class Comparator extends WritableComparator {
    public Comparator() {
      super(NullWritable.class);
    }

    /**
     * Compare the buffers in serialized form.
     */
    public int compare(byte[] b1, int s1, int l1,
                       byte[] b2, int s2, int l2) {
      assert 0 == l1;
      assert 0 == l2;
      return 0;
    }
  }

  static {                                        // register this comparator
    WritableComparator.define(NullWritable.class, new Comparator());
  }
}