Hadoop数据类型

末蓝、 2022-09-23 09:30 122阅读 0赞

> **Writable类**
> 
> Hadoop将许多Writable类归入org.apache.hadoop.io包。形成如下图所示的类层次结构。![20160717201959639][]
> 
> Writable的Java基本类封装
> 
> ![20160717202036843][]  
> 
> 
> ## 除char类型以外，所有的原生类型都有对应的Writable类，并且通过get和set方法（或者new的方式）可以获取和设置它们 ##
> 
> ## 的值。 ##
> 
>   
> 
> 
> **自定义Writable：**
> 
> Hadoop自带一系列有用的Writable实现，可以满足绝大多数用途。但有时，我们需要编写自己的自定义实现。通过自定
> 
> 义Writable，我们能够完全控制二进制表示和排序顺序。Writable是MapReduce数据路径的核心，所以调整二进制表示对其性
> 
> 能有显著影响。现有的Hadoop Writable应用已得到很好的优化，但为了对付更复杂的结构，最好创建一个新的Writable类型，
> 
> 而不是使用已有的类型。
> 
>   
> 
> 
> 编写一个表示一对字符串的实现，名为TextPair：
> 
>     import java.io.*;
>     import org.apache.hadoop.io.*;
>     public class TextPair implements WritableComparable<textpair> {
>     	private Text first;
>     	private Text second;
>     	public TextPair() {
>     		set(newText(),newText());
>     	}
>     	public TextPair(String first, String second) {
>     		set(newText(first),newText(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 write(DataOutput out)throws IOException {
>     		first.write(out);
>     		second.write(out);
>     	}
>     	@Override
>     	public void readFields(DataInput in)throwsIOException {
>     		first.readFields(in);
>     		second.readFields(in);
>     	}
>     	@Override
>     	public int hashCode() {
>     		return first.hashCode() *163+ second.hashCode();
>     	}
>     	@Override
>     	public boolean equals(Object o) {
>     		if(o instanceof TextPair) {
>     			TextPair tp = (TextPair) o;
>     			return first.equals(tp.first) && second.equals(tp.second);
>     		}
>     		return false;
>     	}
>     	@Override
>     	public String toString() {
>     		return first +"\t"+ second;
>     	}
>     	@Override
>     	public int compareTo(TextPair tp) {
>     		int cmp = first.compareTo(tp.first);
>     		if(cmp !=0) {
>     			return cmp;
>     		}
>     		return second.compareTo(tp.second);
>     	}
>     }
> 
>   
> 
> 
> 此实现的第一部分直观易懂：有两个Text实例变量(first和second)和相关的构造函数、get方法和set方法。所有的Writable
> 
> 实现都必须有一个默认的构造函数，以便MapReduce框架能够对它们进行实例化，进而调用readFields()方法来填充它们的字
> 
> 段。 Writable实例是易变的、经常重用的，所以我们应该尽量避免在write()或readFields()方法中分配对象。
> 
>   
> 
> 
> 通过委托给每个Text对象本身，TextPair的write()方法依次序列化输出流中的每一个Text对象。同样，也通过委托给Text对
> 
> 象本身，readFields()反序列化输人流中的字节。DataOutput和DataInput接口有丰富的整套方法用于序列化和反序列化Java基
> 
> 本类型，所以在一般情况下，我们能够完全控制Writable对象的数据传输格式。
> 
>   
> 
> 
> 正如为Java写的任意值对象一样，我们会重写java.lang.Object的hashCode()方法，equals()方法和toString()方法。
> 
> HashPartitioner使用hashCode()方法来选择reduce分区，所以应该确保写一个好的哈希函数来确保reduce函数的分区在大小上
> 
> 是相当的。
> 
>   
> 
> 
> TextPair是WritableComparable的实现，所以它提供了compareTo()方法的实现，加入我们希望的顺序：它通过一个一个
> 
> String逐个排序。请注意，TextPair不同于前面的TextArrayWritable类(除了它可以存储Text对象数之外)，因为
> 
> TextArrayWritable只是一个Writable，而不是WritableComparable。
> 
>   
> 
> 
> **实现一个快速的RawComparator**
> 
>   
> 
> 
> 上例中所示代码能够有效工作，但还可以进一步优化。正如前面所述，在MapReduce中，TextPair被用作键时，它必须被
> 
> 反序列化为要调用的compareTo()方法的对象。是否可以通过查看其序列化表示的方式来比较两个TextPair对象。
> 
>   
> 
> 
> 事实证明，我们可以这样做，因为TextPair由两个Text对象连接而成，二进制Text对象表示是一个可变长度的整型，包含
> 
> UTF-8表示的字符串中的字节数，后跟UTF-8字节本身。关键在于读取开始的长度。从而得知第一个Text对象的字节表示有多
> 
> 长，然后可以委托Text对象的RawComparator，然后利用第一或者第二个字符串的偏移量来调用它。下面例子给出了具体方法
> 
> (注意，该代码嵌套在TextPair类中)。
> 
>   
> 
> 
> 用于比较TextPair字节表示的RawComparator：
> 
>     public static class Comparator extends WritableComparator {
>     	private static final Text.Comparator TEXT_COMPARATOR =new Text.Comparator();
>     	public Comparator() {
>     		super(TextPair.class);
>     	}
>     	@Override
>     	public int  compare(byte[] b1,int s1,int l1,
>     	                    byte[] b2,int s2,int l2) {
>     		try {
>     			int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1);
>     			int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2);
>     			int cmp = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
>     			if(cmp != 0) {
>     				return cmp;
>     			}
>     			return TEXT_COMPARATOR.compare(b1, s1 + firstL1, l1 - firstL1,
>     			                               b2, s2 + firstL2, l2 - firstL2);
>     		} catch(IOException e) {
>     			throw new IllegalArgumentException(e);
>     		}
>     	}
>     }
>     static {
>     	WritableComparator.define(TextPair.class,newComparator());
>     }
> 
>   
> 
> 
> 事实上，我们一般都是继承WritableComparator，而不是直接实现RawComparator，因为它提供了一些便利的方法和默
> 
> 认实现。这段代码的精妙之处在于计算firstL1和firstL2,每个字节流中第一个Text字段的长度。每个都由可变长度的整型(由
> 
> WritableUtils的decodeVIntSize()返回)和它的编码值(由readVInt()返问)组成。
> 
>   
> 
> 
> 静态代码块注册原始的comparator以便MapReduce每次看到TextPair类，就知道使用原始comparator作为其默认
> 
> comparator。  
> 
> 
>   
> 
> 
> **自定义comparator**
> 
>   
> 
> 
> 从TextPair可知，编写原始的cornparator比较费力，因为必须处理字节级别的细节。如果需要编写自己的实现，
> 
> org.apache.hadoop.io包中Writable的某些前瞻性实现值得研究研究。WritableUtils的有效方法也比较非常方便。
> 
>   
> 
> 
> 如果可能，还应把自定义comparator写为RawComparators。这些comparator实现的排序顺序不同于默认comparator定义
> 
> 的自然排序顺序。下面的例子显示了TextPair的comparator，称为First Comparator。只考虑了一对Text对象中的第一个字符
> 
> 串。请注意，我们重写了compare()方法使其使用对象进行比较，所以两个compare()方法的语义是相同的。
> 
>   
> 
> 
> 自定义的RawComparator,用于比较TextPair字节表示中的第一字段：
> 
>     public static class FirstComparator extends WritableComparator {
>     	private static final Text.Comparator TEXT_COMPARATOR =newText.Comparator();
>     	public FirstComparator() {
>     		super(TextPair.class);
>     	}
>     	@Override
>     	public int compare(byte[] b1,ints1,intl1,
>     	                   byte[] b2,ints2,intl2) {
>     		try {
>     			int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1);
>     			int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2);
>     			return TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
>     		} catch(IOException e) {
>     			throw new IllegalArgumentException(e);
>     		}
>     	}
>     	@Override
>     	public int compare(WritableComparable a, WritableComparable b) {
>     		if(a instanceof TextPair && b instanceof TextPair) {
>     			return((TextPair) a).first.compareTo(((TextPair) b).first);
>     		}
>     		return super.compare(a, b);
>     	}
>     }
> 
>   
> 更多Writable参见 [http://hadoop.apache.org/docs/current/api/org/apache/hadoop/io/Writable.html][http_hadoop.apache.org_docs_current_api_org_apache_hadoop_io_Writable.html]
> 
>   
>

[20160717201959639]: /images/20220720/6e94c8cafe6845efaceefbfb6b17d391.png
[20160717202036843]: /images/20220720/bad09bc5289b40aa8e5f923c1b1411e2.png
[http_hadoop.apache.org_docs_current_api_org_apache_hadoop_io_Writable.html]: http://hadoop.apache.org/docs/current/api/org/apache/hadoop/io/Writable.html