2.3 HDFS的数据存储

前面主要介绍了HDFS系统的运行机制和原理，本节将介绍HDFS系统中的文件数据是如何存储和管理的。

2.3.1 数据完整性

I/O操作过程中，难免会出现数据丢失或脏数据，数据传输的量越大，出错的概率越高。校验错误最常用的办法，就是传输前计算一个校验和，传输后计算一个校验和，两个校验和如果不相同，就说明数据存在错误。为了保证数据的完整性，一般采用下列数据校验技术：①奇偶校验技术；②MD5、SHA1等校验技术；③CRC-32循环冗余校验技术；④ECC内存纠错校验技术。其中，比较常用的错误校验码是CRC-32。

HDFS将一个文件分割成一个或多个数据块，这些数据块被编号后，由名字节点保存，通常需要记录的信息包括文件的名称、文件被分成多少块、每块有多少个副本、每个数据块存放在哪个数据节点上、其副本存放于哪些节点上，这些信息被称为元数据。

HDFS为了保证数据的完整性，采用校验和（checksum）检测数据是否损坏。当数据第一次引入系统时计算校验和，并且在一个不可靠的通道中传输的时候，再次检验校验和。但是，这种技术并不能修复数据（注意：校验和也可能损坏，但是，由于校验和小得多，所以可能性非常小）。数据校验和采用的是CRC-32，任何大小的数据输入都可以通过计算，得出一个32位的整数校验和。

DataNode在接收到数据后存储该数据及其校验和，或者将数据和校验和复制到其他的DataNode上。当客户端写数据时，会将数据及其DataNode发送到DataNode组成的管线，最后一个DataNode负责验证校验和，如果有损坏，则抛出ChecksumException，这个异常属于IOException的子类。客户端读取数据的时候，也会检验校验和，会与DataNode上的校验和进行比较。每个DataNode上面都会有一个用于记录校验和的日志。客户端验证完之后，会告诉DataNode，然后更新这个日志。

不仅客户端在读写数据的时候验证校验和，每个DataNode也会在后台运行一个DataBlockScanner，从而定期检查存储在该DataNode上面的数据块。

如果客户端发现有block坏掉，按照以下步骤进行恢复。

（1）客户端在抛出ChecksumException之前，会把坏的block和block所在的DataNode报告给NameNode。

（2）NameNode把这个block标记为已损坏，这样，NameNode就不会把客户端指向这个block，也不会复制这个block到其他的DataNode。

（3）NameNode会把一个好的block复制到另外一个DataNode。

（4）NameNode把坏的block删除。

HDFS会存储每个数据块的副本，可以通过数据副本来修复损坏的数据块。客户端在读取数据块时，如果检测到错误，首先向NameNode报告已损坏的数据块及其正在尝试读取操作的这个DataNode。NameNode会将这个数据块标记为已损坏，对这个数据块的请求会被NameNode安排到另一个副本上。之后，它安排这个数据块的另一个副本复制到另一个DataNode上，如此，数据块的副本因子又回到期望水平。此后，已损坏的数据块副本会被删除。

Hadoop的LocalFileSystem执行客户端的校验和验证。当写入一个名为filename的文件时，文件系统客户端会明确地在包含每个文件块校验和的同一个目录内建立一个名为filename.crc的隐藏文件。

2.3.2 数据压缩

Hadoop作为一个较通用的海量数据处理平台，每次运算都会需要处理大量的数据。使用文件和数据压缩技术有明显的优点：①节省数据占用的磁盘空间；②加快数据在磁盘和网络中的传输速度，从而提高系统的处理速度。我们来了解一下Hadoop中的文件压缩。

Hadoop支持多种压缩格式。我们可以把数据文件压缩后再存入HDFS，以节省存储空间。在表2-2中，列出了几种压缩格式。

所有的压缩算法都存在空间与时间的权衡：更快的压缩速率和解压速率是以牺牲压缩率为代价的。Deflate算法是同时使用了LZ77与哈夫曼编码的一个无损数据压缩算法，源代码可以在zlib库中找到。Gzip算法是以Deflate算法为基础扩展出来的一种算法。Gzip在时间和空间上比较适中，Bzip2算法压缩比Gzip更有效，但速度更慢。Bzip2的解压速度比它的压缩速度要快，但与其他压缩格式相比，又是最慢的，但压缩效果明显是最好的。

使用压缩，有两个比较麻烦的地方：第一，有些压缩格式不能被分块、并行地处理，比如Gzip；第二，另外的一些压缩格式虽然支持分块处理，但解压的过程非常缓慢，使作业瓶颈转移到了CPU上，例如Bzip2。LZO是一种既能够被分块并且并行处理速度也非常快的压缩算法。在Hadoop中，使用LZO压缩算法可以减小数据的大小并缩短数据的磁盘读写时间，在HDFS中存储压缩数据，可以使集群能保存更多的数据，延长集群的使用寿命。不仅如此，由于MapReduce作业通常瓶颈都在I/O上，存储压缩数据就意味着更少的I/O操作，作业运行更加高效。例如，将压缩文件直接作为入口参数交给MapReduce处理，MapReduce会自动根据压缩文件的扩展名来自动选择合适的解压器处理数据。处理流程如图2-12所示。

LZO的压缩文件是由许多小的blocks组成（约256KB），使得Hadoop的作业可以根据block的划分来分块工作（split job）。不仅如此，LZO在设计时就考虑到了效率问题，它的解压速度是Gzip的两倍，这就让它能够节省很多的磁盘读写，它的压缩比不如Gzip，大约压缩出来的文件比Gzip压缩的大一半，但是，这仍然比没有经过压缩的文件要节省20%～50%的存储空间，这样，就可以在效率上大大地提高作业执行的速度。

在考虑如何压缩由MapReduce程序将要处理的数据时，压缩格式是否支持分割是很重要的。比如，存储在HDFS中的未压缩的文件大小为1GB, HDFS的块大小为64MB，所以该文件将被存储为16块，将此文件用作输入的MapReduce作业，会创建1个输入分片（split，也称为“分块”。对应block，我们统一称为“块”），每个分片都被作为一个独立map任务的输入，单独进行处理。

现在假设该文件是一个Gzip格式的压缩文件，压缩后的大小为1GB。与前面一样，HDFS将此文件存储为16块。然而，针对每一块创建一个分块是没有用的，因为不可能从Gzip数据流中的任意点开始读取，map任务也不可能独立于其他分块只读取一个分块中的数据。Gzip格式使用Deflate算法来存储压缩过的数据，Deflate将数据作为一系列压缩过的块进行存储。但是，每块的开始没有指定用户在数据流中任意点定位到下一个块的起始位置，而是其自身与数据流同步。因此，Gzip不支持分割（块）机制。

在这种情况下，MapReduce不分割Gzip格式的文件，因为它知道输入是Gzip压缩格式的（通过文件扩展名得知），而Gzip压缩机制不支持分割机制。这样是以牺牲本地化为代价的：一个map任务将处理16个HDFS块。大都不是map的本地数据。与此同时，因为map任务少，所以作业分割的粒度不够细，从而导致运行时间变长。

在我们假设的例子中，如果是一个LZO格式的文件，我们会遇到同样的问题，因为基本压缩格式不为reader提供方法使其与流同步。但是，Bzip2格式的压缩文件确实提供了块与块之间的同步标记（一个48位的PI近似值），因此它支持分割机制。

对于文件的收集，这些问题会稍有不同。Zip是存档格式，因此，它可以将多个文件合并为一个Zip文件。每个文件单独压缩，所有文档的存储位置存储在Zip文件的尾部。这个属性表明Zip文件支持文件边界处分割，每个分片中包括Zip压缩文件中的一个或多个文件。

2.3.3 序列化

序列化是指将结构化对象转换成字节流，以便于进行网络传输，或写入持久存储的过程。与之相对的反序列化，就是将字节流转化为一系列结构化对象的过程。

（1）序列化有以下特征。

●紧凑：可以充分利用稀缺的带宽资源。

●快速：通信时大量使用序列化机制，因此，需要减少序列化和反序列化的开销。

●可扩展：随着通信协议的升级而可升级。

●互操作：支持不同开发语言的通信。

（2）序列化的主要作用如下：

●作为一种持久化格式。

●作为一种通信的数据格式，支持不同开发语言的通信。

●作为一种数据拷贝机制。

Hadoop的序列化机制与Java的序列化机制不同，它实现了自己的序列化机制，将对象序列化到流中，值得一提的是，Java的序列化机制是不断地创建对象，但在Hadoop的序列化机制中，用户可以复用对象，减少了Java对象的分配和回收，提高了应用效率。

在分布式系统中，进程将对象序列化为字节流，通过网络传输到另一进程，另一进程接收到字节流，通过反序列化，转回到结构化对象，以实现进程间通信。在Hadoop中，Mapper、Combiner、Reducer等阶段之间的通信都需要使用序列化与反序列化技术。举例来说，Mapper产生的中间结果<key: value1, value2...>需要写入到本地硬盘，这是序列化过程（将结构化对象转化为字节流，并写入硬盘），而Reducer阶段，读取Mapper的中间结果的过程则是一个反序列化过程（读取硬盘上存储的字节流文件，并转回为结构化对象）。需要注意的是，能够在网络上传输的只能是字节流，Mapper的中间结果在不同主机间洗牌时，对象将经历序列化和反序列化两个过程。

序列化是Hadoop核心的一部分，在Hadoop中，位于org.apache.hadoop.io包中的Writable接口是Hadoop序列化格式的实现，Writable接口提供两个方法：

        public interface Writable {
            void write(DataOutput out) throws IOException;
            void readFields(DataInput in) throws IOException;
        }

不过，没有提供比较功能，需要进行比较的话，要实现WritableComparable接口：

        public interface WritableComparable<T> extends Writable, Comparable<T>
        { }

Hadoop的Writable接口是基于DataInput和DataOutput实现的序列化协议，紧凑（高效使用存储空间）、快速（读写数据、序列化与反序列化的开销小）。

Hadoop中的键（key）和值（value）必须是实现了Writable接口的对象（键还必须实现WritableComparable，以便进行排序）。

Hadoop自身提供了多种具体的Writable类，包含了常见的Java基本类型（boolean、byte、short、int、float、long和double等）和集合类型（BytesWritable、ArrayWritable和MapWritable等），如图2-13所示。

Text:Text是UTF-8的Writable，可以理解为与java.lang.String相类似的Writable。

Text类替代了UTF-8类。Text是可变的，其值可以通过调用set（）方法来改变。最大可以存储2GB的大小。

NullWritable:NullWritable是一种特殊的Writable类型，它的序列化长度为零，可以用作占位符。

BytesWritable:BytesWritable是一个二进制数据数组封装，序列化格式是一个int字段。BytesWritable是可变的，其值可以通过调用set（）方法来改变。

ObjectWritable:ObjectWritable适用于字段使用多种类型时。

ArrayWritable和TwoDArrayWritable是针对数组和二维数组的。

MapWritable和SortedMapWritable是针对Map和SortMap的。

虽然Hadoop内建了多种Writable类供用户选择，Hadoop对Java基本类型的包装Writable类实现的RawComparable接口，使得这些对象不需要反序列化过程，便可以在字节流层面进行排序，从而大大缩短了比较的时间开销。但是，当我们需要更加复杂的对象时，Hadoop的内建Writable类就不能满足我们的需求了（需要注意的是Hadoop提供的Writable集合类型并没有实现RawComparable接口，因此也不满足我们的需要），这时，我们就需要定制自己的Writable类，特别在将其作为键（key）的时候更应该如此，以求实现更高效的存储和快速的比较。