Hadoop中Writable类之四

1.定制Writable类型

Hadoop中有一套Writable实现，例如：IntWritable、Text等，但是，有时候可能并不能满足自己的需求，这个时候，就需要自己定制Writable类型。

定制分以下几步：

• 需要实现WritableComparable接口，因为Writable常常作为健值对出现，而在MapReduce中，中间有个排序很重要，因此，Hadoop中就让Writable实现了WritableComparable
• 需要实现WritableComparable的write()、readFields()、compareTo()方法
• 需要重写java.lang.Object中的hashCode()、equals()、toString()方法。

由于hashCode()方法对reduce分区很重要，所以，需要重写java.lang.Object的hashCode()方法

如果要结合使用TextOutputFormat和定制的Writable，则许重写java.lang.Object的toString()方法

 

Example:

CreateWritable.java

package cn.roboson.writable;

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

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
/**
 * 1.定制一个Writable类型,里面包含两个Text
 * 2.和IntWritable等原有的Writable类似，它需要实现WritableComparable
 * 3.由于hashCode()方法对reduce分区很重要，所以，需要重写java.lang.Object的hashCode()方法
 * 4.如果要结合使用TextOutputFormat和定制的Writable，则许重写java.lang.Object的toString()方法
 * @author roboson
 *
 */

public class CreateWritable implements WritableComparable<CreateWritable>{
    
    private Text first;
    private Text second;
    
    //构造方法，这个是必须要的
    public CreateWritable(){
        
    }
    
    public CreateWritable(String first,String second){
        set(new Text(first), new Text(second));
    }
    
    public CreateWritable(Text first,Text second){
        set(first,second);
    }
    
    
    
    public Text getFirst() {
        return first;
    }

    public void setFirst(Text first) {
        this.first = first;
    }

    public Text getSecond() {
        return second;
    }

    public void setSecond(Text second) {
        this.second = second;
    }

    public void set(Text first,Text second){
        this.first=first;
        this.second=second;
    }
    

    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        first.readFields(in);
        second.readFields(in);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        // TODO Auto-generated method stub
        first.write(out);
        second.write(out);
    }

    @Override
    public int compareTo(CreateWritable other) {
        // TODO Auto-generated method stub
        int cmp=first.compareTo(other.getFirst());
        if(cmp!=0){
            return cmp;
        }
        return second.compareTo(other.getSecond());
    }
    
    @Override
    public int hashCode() {
        // TODO Auto-generated method stub
        return first.hashCode()*163 + second.hashCode();
    }
    
    @Override
    public String toString() {
        // TODO Auto-generated method stub
        return first+"	"+second;
    }
    
    @Override
    public boolean equals(Object obj) {
        // TODO Auto-generated method stub
        if(obj instanceof CreateWritable){
            CreateWritable create = (CreateWritable) obj;
            return first.equals(create.getFirst()) && second.equals(create.getSecond());
        }
        return false;
    }

}

 

Writable06.java

package cn.roboson.writable;

import org.apache.hadoop.io.Text;

public class Writable06 {
    
    public static void main(String[] args) {
        CreateWritable createWritable01 = new CreateWritable("Hadoop", "roboson");
        CreateWritable createWritable02 = new CreateWritable(new Text("Hadoop"), new Text("roboson"));
        
        //重写了equals()方法，相比叫first 和 second两者都相同的时候，返回true
        System.out.println(createWritable01.equals(createWritable02));
        
        //重写了compareTo()方法，先比较first，再比较second，返回0：相等 -1：first<second 1:first>second
        System.out.println(createWritable01.compareTo(createWritable02));
        
        //重写了toString()方法，返回 first +"	"+second
        System.out.println(createWritable01.toString());
    }
}

 

运行结果：

 

2.为速度实现一个RawComparator

关于Comparator方面的知识，可以参考我的博文《Hadoop中WritableComparable 和 comparator》，Comparator中，有一个方法是compare(byte[] b1,int s1,int l1,byte[] b2,int s2, int l2)，该方法直接比较的是序列化，不必先将序列化数据流转化为对象，再进行比较，所以，与上面的compareTo()方法，相比，其更高效！其实，我们查看HadoopAPI，就可以发现，基本类型的Writable类都实现有了Comparator作为其内部类：

好，再看看IntWritable.Comparator这个类，如下图所示，发现它是一个静态类，好好观察它的结构：

 

通过上面的，我们可以发现，要为一个Writable实现一个Comparator，安装Hadoop的格式来，需要注意以下几点：

• 将其在内部实现，作为内部类
• 是一个静态的内部类
• 注册，以便可以通过WritableComparator直接创建

 至于注册方面的知识，可以查看我的博客《Hadoop中Comparator原理》

Example:给上面自定义的CreateWritable类实现一个Comparator,也就是CreateWritable.Comparator类

CreateWritable.java

package cn.roboson.writable;

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

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;
/**
 * 1.给定制的CreateWritable类实现一个Comparator，CreateWritable.Comparator
 * @author roboson
 *
 */

public class CreateWritable implements WritableComparable<CreateWritable>{
    
    private Text first;
    private Text second;
    
    //构造方法，这个是必须要的
    public CreateWritable(){
        
    }
    
    public CreateWritable(String first,String second){
        set(new Text(first), new Text(second));
    }
    
    public CreateWritable(Text first,Text second){
        set(first,second);
    }
    
    
    
    public Text getFirst() {
        return first;
    }

    public void setFirst(Text first) {
        this.first = first;
    }

    public Text getSecond() {
        return second;
    }

    public void setSecond(Text second) {
        this.second = second;
    }

    public void set(Text first,Text second){
        this.first=first;
        this.second=second;
    }
    

    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        first.readFields(in);
        second.readFields(in);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        // TODO Auto-generated method stub
        System.out.println(first);
        first.write(out);
        second.write(out);
    }

    @Override
    public int compareTo(CreateWritable other) {
        // TODO Auto-generated method stub
        int cmp=first.compareTo(other.getFirst());
        if(cmp!=0){
            return cmp;
        }
        return second.compareTo(other.getSecond());
    }
    
    @Override
    public int hashCode() {
        // TODO Auto-generated method stub
        return first.hashCode()*163 + second.hashCode();
    }
    
    @Override
    public String toString() {
        // TODO Auto-generated method stub
        return first+"	"+second;
    }
    
    @Override
    public boolean equals(Object obj) {
        // TODO Auto-generated method stub
        if(obj instanceof CreateWritable){
            CreateWritable create = (CreateWritable) obj;
            return first.equals(create.getFirst()) && second.equals(create.getSecond());
        }
        return false;
    }
    
    
    //创建静态内部类：CreateWritable.Comparator
    public static class Comparator extends WritableComparator{

        public Comparator() {
            super(CreateWritable.class);
            // TODO Auto-generated constructor stub
        }
        private static final Text.Comparator TEXT_COMPARATOR = new Text.Comparator();
        
        @Override
        public int compare(byte[] b1, int s1, int l1, byte[] b2,
                int s2, int l2) {
            // TODO Auto-generated method stub
            int firstL1 = 0,firstL2 = 0;
            try {
                firstL1 = WritableUtils.decodeVIntSize(b1[s1])+readVInt(b1, s1);
                firstL2 = WritableUtils.decodeVIntSize(b2[s2])+readVInt(b2, s2);
                int cmp = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
                if(cmp !=0){
                    return cmp;
                }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return TEXT_COMPARATOR.compare(b1, s1+firstL1, l1-firstL1, b2, s2+firstL2, l2-firstL2);
        }
        
    }
    static{
        WritableComparator.define(CreateWritable.class,new Comparator());
    }
}

 

Writable07.java

package cn.roboson.writable;

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

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

public class Writable07 {
    
    public static void main(String[] args) throws IOException {
        
        CreateWritable create01 = new CreateWritable(new Text("Hadoop"), new Text("1"));
        CreateWritable create02 = new CreateWritable(new Text("Hadoop"), new Text("2"));
        byte[] b1 = serialize(create01);
        byte[] b2 = serialize(create02);
        RawComparator<CreateWritable> comparator = WritableComparator.get(CreateWritable.class);
        int cmp =comparator.compare(b1, 0, b1.length, b2, 0, b2.length);
        if(cmp ==0){
            System.out.println("create01 == create02");
        }else if(cmp ==-1){
            System.out.println("create01<create02");
        }else{
            System.out.println("create01>create02");
        }
    }
    
    public static byte[] serialize(Writable writable) throws IOException{
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        DataOutputStream dataOut = new DataOutputStream(out);
        writable.write(dataOut);
        return out.toByteArray();
        
    }
}

 

运行结果：

分析：

在上面的CreateWritable的内部类Comparator中，需要实现一个方法compare()方法，那么compare该如何实现，才能够对序列化数据流进行比较。

CreateWritable是由两个Text对象组成的（Text first, Text second），而Text对象的二进制表示，是一个长度可变的整数，包含字符串之UTF-8表示的字节数以及UTF-8字节本身。诀窍在于读取该对象的起始长度，由此得知第一个Text对象的字节表示有多长；然后将该长度传递给Text对象的RawComparator方法，最后通过计算第一个字符串和第二个字符串恰当的偏移量，这样便可以实现对象的比较。

也就是说Text对象序列化后，也是由两部分组成，一部分是记录本身有多少个字节，另一部分就是它自己的字节数！

记录它本身有多少个字节所占用的字节长度：WritableUtils.decodeVintSize（）方法返回一个整数，代表它的字节长度。

它自己本身的字节数：WritableComparator的readVInt()方法返回一个整数，代表它本身的字节长度。

这样就方便了，相比叫first的序列化数据流，根据结果，判断，看是否需要再比较second的序列化数据流。