Hadoop（19）MapReduce 文件处理：基于压缩的高效存储（二）

tythg

5.2 基于压缩的高效存储（续）
（仅包括技术27）



技术27 在MapReduce，Hive和Pig中使用可分块的LZOP

如果一个文本文件即使经过压缩后仍然比HDFS的块的大小要大，就需要考虑选择一个支持分块的压缩编码器，以防一个单一的map任务来处理整个超大的文件。

LZOP可以满足分块的要求，但是使用起来很复杂。原因在于LZOP不是直接支持分块。LZOP是基于块的格式，但是并不支持块的随机访问。


问题

需要选择一个压缩编码器使MapReduce可以调用多个任务并行处理一个单一的压缩文件。


方案

在MapReduce中，对LZOP压缩的输入文件进行分块需要使用针对LZOP的输入格式类，如LzoInputFormat。在Pig和Hive中使用LZOP也是如此。



讨论

压缩编码器中只有LZOP和bzip2支持分块。Bzip2压缩太慢，以至于不太可靠。LZOP在压缩率和速度之间取得了相对可靠的平衡。



在集群配置LZOP

不幸的是，由于授权原因，Hadoop并没有自带LZOP。要在集群中将一切准备工作做好将非常费劲。附录A.10中有配置LZOP的具体步骤。（译注：附录A.10的翻译见下一篇翻译文章。http://www.iyunv.com/datacloud/p/3617586.html）



在HDFS中读写LZOP文件

在此前的技术中（技术25， 26）介绍了如何读取并写入压缩文件。读写LZOP需要在代码中指定LZOP编码器。实现代码如下所示：



Methods to read and write LZOP files in HDFS




1 public static Path compress(Path src, Configuration config)
2     throws IOException {
3     
4     Path destFile = new Path(src.toString() + new LzopCodec().getDefaultExtension());
5     LzopCodec codec = new LzopCodec();
6     codec.setConf(config);
7     FileSystem hdfs = FileSystem.get(config);
8     InputStream is = null;
9
10     OutputStream os = null;
11     
12     try {
13         is = hdfs.open(src);
14         os = codec.createOutputStream(hdfs.create(destFile));
15         IOUtils.copyBytes(is, os, config);
16     } finally {
17         IOUtils.closeStream(os);
18         IOUtils.closeStream(is);
19     }
20     
21     return destFile;
22 }
23
24 public static void decompress(Path src, Path dest, Configuration config)
25     throws IOException {
26     
27     LzopCodec codec = new LzopCodec();
28     codec.setConf(config);
29     FileSystem hdfs = FileSystem.get(config);
30     InputStream is = null;
31     OutputStream os = null;
32     
33     try {
34         is = codec.createInputStream(hdfs.open(src));
35         os = hdfs.create(dest);
36         IOUtils.copyBytes(is, os, config);
37     } finally {
38         IOUtils.closeStream(os);
39         IOUtils.closeStream(is);
40     }
41 }



然后写入并读取一个LZOP文件，确保LZOP工具可以操作生成的文件。脚本如下所示：

$ hadoop fs -put $HADOOP_HOME/conf/core-site.xml core-site.xml

$ bin/run.sh com.manning.hip.ch5.LzopFileReadWrite core-site.xml


上述代码将在HDFS中生成一个core-site.xml.lzo文件。现在需要确定可以通过lzop程序来处理这个LZOP文件。操作步骤如下：

在主机上安装lzop程序。（RedHat和Centos上可以从http://pkgs.repoforge.org/lzop/lzop-1.03-1.el5.rf.x86_64.rpm 安装。）
将LZOP文件从HDFS上拷贝到本地磁盘。
使用lzop程序解压缩这个LZOP文件。
将解压缩后的文件和原始文件进行对比。
操作脚本如下：




$ hadoop fs -get core-site.xml.lzo /tmp/core-site.xml.lzo

$ lzop -l /tmp/core-site.xml.lzo

method compressed uncompr. ratio uncompressed_name
LZO1X-1 454 954 47.6% core-site.xml

$ cd /tmp

$ lzop -d core-site.xml.lzo

$ ls -ltr

-rw-r--r-- 1 aholmes aholmes 954 Sep 11 09:05 core-site.xml
-rw-r--r-- 1 aholmes aholmes 504 Sep 11 09:05 core-site.xml.lzo

$ diff core-site.xml $HADOOP_HOME/conf/core-site.xml
$



通过diff程序的比较，说明使用LZOP编码器压缩的文件可以被lzop程序解压缩。然后就需要为LZOP文件建立索引，使它可以被分块。



为LZOP文件创建索引

LZOP支持分块，但是不支持随机访问。这是因为LZOP没有存储每个块的地址信息（地址偏移量）。那么现在需要做的就是创建一个包含LZOP压缩文件中每个块的地址信息（地址偏移量）的索引。创建方法如图5.5所示，遍历一次LZOP的压缩文件，将每个块的地址偏移量保存在索引文件中。索引文件是一个包含了一系列连续的64位的数字。这些数字包含了LZOP压缩文件中每个块的地址偏移量。


有两种方法可以创建索引文件，正如下面的两个代码片段。如果只是要为一个LZOP文件创建一个索引文件，以下就是一个可以完成这个目标的简单库：

shell$ bin/run.sh \         com.hadoop.copmression.lzo.DistributedLzoIndexer \         core.site.xml.lzo \         /path/to/lzop
如果需要批量处理LZOP文件，生成索引，那么就需要更简便的方法。下面这段代码调用一个MapReduce作业来创建索引文件。它通过遍历文件夹中的LZOP文件来支持文件夹作为输入源。也可以以文件作为输入源。

shell$ bin/run.sh \          com.hadoop.copmression.lzo.DistributedLzoIndexer \          core.site.xml.lzo \          /path/to/lzop
前述两段代码都会在LZOP文件的同一个目录下生成一个索引文件。索引文件的文件名是在原LZOP文件名后面加上.index。以上代码会生成文件名为core-site.xml.lzo.index的索引文件。接下来介绍如何在JAVA代码中调用LzoIndexer。以下代码可以同步处理LZOP文件，同步生成索引文件：


1 LzoIndexer lzoIndexer = new LzoIndexer(new Configuration());2 3 for (String arg: args) {4     try {5         lzoIndexer.index(new Path(arg));6     } catch (IOException e) {7         LOG.error("Error indexing " + arg, e);8     }9 ...

通过DistributedLzoIndexer，MapReduce作业将会为每个lzo文件调用一个map任务。不需要reduce任务。Map任务通过自定义的LzoSplitInputFormat和LzoIndexOutputFormat可以直接生成索引文件。如果需要自定义调用MapReduce的JAVA代码，可以参考DistributedLzoIndexer的源代码。
通过前面的代码，为LZOP文件生成了相应的索引文件。接下来介绍如何在MapReduce中使用它们。

MAPREDUCE和LZOP
有了LZOP文件和相应的索引文件之后，就可以在MapReduce中处理LZOP文件了。然而，Hadoop中并没有哪个输入格式直接支持LZOP和它的索引文件。这里就需要为LZOP自定义输入格式类。
以下代码展示了如何在MapReduce作业中处理LZOP。代码可以用来处理文本文件压缩后的LZOP文件。

1 job.setInputFormatClass(LzoTextInputFormat.class);2 job.setOutputFormatClass(TextOutputFormat.class);3 job.getConfiguration().setBoolean("mapred.output.compress", true);4 job.getConfiguration().setClass("mapred.output.compression.codec",5 LzopCodec.class, CompressionCodec.class);
另外，压缩map的中间输出也可以提高MapReduce作业的运行速度。代码如下：

1 conf.setBoolean("mapred.compress.map.output", true);2 conf.setClass("mapred.map.output.compression.codec", LzopCodec.class, CompressionCodec.class);
通过配置集群的hdfs-site.xml文件可以总是压缩map的输出：


1 <property>2     <name>mapred.compress.map.output</name>3     <value>true</value>4 </property>5 <property>6     <name>mapred.map.output.compression.codec</name>7     <value>com.hadoop.compression.lzo.LzopCodec</value>8 </property>

需要注意的是，LZOP文件中分块的数量是文件占据的LZOP的块的数量，不是文件占据的HDFS块的数量。
接下来介绍如何在Pig和Hive中处理LZOP。

PIG和HIVE
Elephant Bird是Twitter维护的一个项目，包含处理LZOP的工具。它提供了很多有用的MapReduce和Pig类来处理LZOP。Elephant Bird提供了LzoPigStorage来处理Pig中文本文件的LZOP压缩数据。
Hive可以通过com.hadoop.mapred.DeprecatedLzoTextInputFormat这个输入格式来处理文本文件的LZOP压缩文件。这个输入格式类可以在Todd Lipcon和Kevin Weil的LZO项目中找到。

小结
在Hadoop中处理可分块的压缩数据很有技巧性。如果恰好可以直接用SequenceFiles或Avro来处理数据，那么是再简单不过了。如果一定要压缩并分块，那只能用LZOP了。
正如前面提到的，Elephant Bird项目提供了一些有用的LZOP输入格式来处理LZOP的压缩文件，比如说XML和纯文本的压缩文件。如果你需要处理不被Todd Lipcon的LZO项目和Elephant Bird支持的LZOP的压缩文件格式，就需要自定义输入格式。这对大部分人来说相当困难。期望Hadoop可以早日提供对有特别的分块逻辑的压缩文件的支持，减少开发者的负担。
压缩是所有的生产环境所需要的，因为资源永远稀缺。压缩可以加快执行，减少存储空间。在前面一个章节介绍了如何评价并选择合适的编码器。然后介绍了如何在HDFS，MapReduce，Pig和Hive中使用压缩。最后介绍了如何处理LZOP压缩。