前言

Hadoop I/O 操作

Hadoop 的 I/O 操作

数据完整性

在数据进入系统时计算校验和,并在数据通过一个不可靠通道时再次校验和,这样就能发现数据是否损坏.
常用的错误检测码是 CRC-32(循环冗余校验),任何大小的数据输入均计算得到一个32为的整数校验和.
datanode在收到客户端的数据或者其他datanode的数据时存储该数据和校验和.客户端将数据以及校验和发送到一系列datanode组成的管线,管线中的最后一个datanode负责验证校验和.如果有错,客户端会收到ChecksumException异常.
客户端从datanode中读取数据时,也会验证校验和.每个datanode均持久保持一个用于验证的校验和日志.当有客户端验证成功时,日志就会更新.datanode中也有DataBlockScanner定期验证这个datanode上的数据块.

LocalFileSystem

文件块的大小作为元数据存储在.crc文件中.
校验和的代价是相当低的.

ChecksumFileSystem

LocalFileSystem 通过 ChecksumFileSystem 来完成任务,有了这个类,向其他文件系统(无校验和系统)加入校验和就非常简单,因为ChecksumFileSystem 类继承自FileSystem类.

压缩

通过CompressionCodec对数据进行压缩和解压缩
通过compressionCodecFactory推断出CompressionCode
在读取一个压缩文件时,通常可以通过文件拓展名推断需要使用哪一个codec.
通过使用getCodec()的方法,CompressionCodeFactory提供一种可以将文件拓展名映射到一个CompressionCodec的方法.
为了性能,最好使用”原生”(native)类库来实现压缩和解压缩.
默认情况下,Hadoop 会根据自身运行的平台搜索原生代码库,如果找到相应的代码库就会自动加载
CodecPool

如果使用的是原生代码库并且需要在应用中执行大量的压缩和解压缩操作, 可以考虑使用CodecPool, 它支持反复使用压缩和解压缩,已分摊创建这些对象的开销.

压缩和输入分片

压缩格式支持切分(splitting)才可以被MapReduce处理.压缩格式必须支持数据读取和数据流同步.

###
要想压缩MapReduce作业的输出,应在作业配置过程中进行相应的设置.

序列化

序列化(serialization) 在分布式数据处理的两大领域之间经常出现:进程间通信和永久存储.
在Hadoop中, 系统中的多个节点上的进程间通信是通过”远程过程调用”(remote procedure call, RPC)实现的.RPC协议将消息序列化成二进制流后发送到远程节点,远程节点接着将二进制流反序列化成为原始消息.
通常, RPC序列化格式如下:

紧凑(充分利用带宽)
快速(最基本的)
可拓展(引进相应的协议)
支持互操作(特定的格式满足不同语言)
因为序列化的数据并不需要永久储存,它们存活时间不到一秒钟.因此,对数据永久存储而言,RPC序列化格式的4大理想属性非常重要.
储存格式比较紧凑(今儿高效使用存储空间)
快速(以读/写数据的额外开销比较小)
可拓展(以可以透明地读取老格式的数据)
互操作(可以使用不同的语言读/写永久存储的数据)

Hadoop 使用的是自己的序列化格式 writable, 它绝对紧凑,速度快,但是不太容易使用java以外的语言进行拓展.因为Writable是Hadoop的核心(大多数MapReduce程序都会为键和值使用它).

Writable 接口

Writable 接口定义了两个方法: 一个将其状态写到DataOutput二进制流,另一个从DataInput二进制流读取状态:

package org.apache.hadoop.io
import java.io.DataOutput;
import java.io.DataInput;
import java.io.IOException;
public interface Writable {
    void write(DataOutput out) throws IOException;
    void readFields(DataInput in) throws IOException;
}

使用IntWritable来封装java int 类型.

1 2	IntWritable writable = new IntWritable(); Writable.set(163);

WritableComparable 接口和 comparator
IntWritable 实现原始的WritableComparable 接口,该接口继自Writable和java.lang.Comparable接口.

1
2
3

package org.apache.hadoop.io;
public interface WritableComparable<T> extends Writable, Comparable<T> {
}

对MapReduce来说,类型比较重要,因为中间有个基于键的排序阶段.Hadoop 提供的一个优化接口是继承自Java Comparator 的 RawComparator接口.

WritableComparator是对继承自WritableComparable类的Rawcomparator类的一个通用实现.它提供两个功能.第一,它提供对原始compare()方法的一个默认实现,该方法能够反序列化将在流中进行比较的对象,并调用对象的compare()方法.第二,他充当的是RawComparator实例的工厂(已注册Writable的实现).例如,为了获得IntWritable的comparator,我们直接如下调用:

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

这个comparator可以用于比较两个IntWritable对象或者两个序列化对象.

Writable类

Hadoop 自带的org.apache.hadoop.io包中有广泛的Writable类可选.

java基本类型的Writable封装器

Writable类对java基本类型提供封装,char除外(可以存储在IntWritable中).所有的封装包含get()和set()两个方法用于读取或储存封装的值.

对整数进行编码时,定长格式非常适合对整个值域空间中分布非常均匀的数值进行编码,如精心设计的哈希函数.大多数数值变量的分布都不均匀,而且变长格式一般更加节省空间.变长编码的另一个优点是可以在VIntWritable 和 VLongWritable转换,因为他们的编码实际上是一致的.所以选择变长格式之后,便有增长的空间,不必一开始就用8字节的long表示.

Text类型

Text 是针对UTF-8序列的Writable类.Text替代了UTF8类,但并不是一个很好的替代,一者因为不支持对字节数超过32767的字符进行编码,二者因为它使用的是Java的UTF-8修订版.

Text类使用整型(变长编码)来储存字符串所需的字节数,因此最大值为2GB.
由于着重使用标准的UTF-8编码,因此Text类和java String 类之间存在一定的差别.
对于Text类的索引是根据编码后字节序列中的位置是实现的(String 是根据char序列),并非字符串中的Unicode字符,也不是java char中的编码单元.
charAt()返回的是表示编码位置的整数.getBytes()方法返回的字节数组可能比getLength()函数返回的长度更长.

可变性与String相比,Text的另一个区别在于它是可变的(与所有Hadoop的Writable接口实现类似, NullWritable除外, 它是单例对象).可以通过调用其中一个set()方法来重用Text实例.

BytesWritable

BytesWritable 是对二进制数据数组的封装.它的序列化格式为一个指定所含数据字节的整数域(4字节),后跟数据内容本身.可通过getLenth()方法确定所存数据大小.

NullWritable

NullWritable是Writable的特殊类型,它的序列化长度为0.它并不从数据流中读取数据,也不写入数据.它充当占位符.如果不需要使用键或值的序列化地址,就可以将键或值声明为NullWritable, 结果是高效地存储常量空值.如果希望存储一系列数值,与键/值对相对,NullWritable也可以用作在SequenceFile中的键.它是一个不可变的单例类型,同过调用NullWritable.get()方法可以获取这个实例.

ObjectWritable 和GenericWritable

ObjectWritable 是对Java基本类型(String, enum, Writable, null 或这些类型组成的数组)的一个通用封装.它在hadoop RPC中用于对方法的参数和返回类型进行封装和解封装.

Writable集合类

org.apache.hadoop.io软件包中一共有6个Writable集合类:ArrayWritable, ArrayPrimitiveWritable, TwoDArrayWritable, MapWritable, SortedMapWritable 和 EnumMapWritable

实现定制的Writable集合

为速度实现一个RawComparator
定制的comparator