普开数据大数据应用实例：基于Hadoop的大规模数据排序算法

Map/Reduce框架运转在 键值对上，也就是说， 框架把作业的输入看为是一组 键值对，同样也产出一组 键值对做为作业的输出，这两组键值对的类型可能不同。 框架需要对key和value的类(class)进行序列化操作， 因此，这些类需要实现 Writable接口。 另外，为了方便框架执行排序操作，key类必须实现 WritableComparable接口。 一个Map/Reduce 作业的输入和输出类型如下所示：

(input) -> map ->-> combine ->-> reduce ->(output) (5) Map/Reduce - 用户界面

这部分文档为用户将会面临的Map/Reduce框架中的各个环节提供了适当的细节。这应该会帮助用户更细粒度地去实现、配置和调优作业。然而，需要注意每个类/接口的javadoc文档提供最全面的文档。 我们会先看看Mapper和Reducer接口。应用程序通常会通过提供map和reduce方法来实现它们。 然后，我们会讨论其他的核心接口，其中包括： JobConf，JobClient，Partitioner， OutputCollector，Reporter， InputFormat，OutputFormat等等。

最后，我们将通过讨论框架中一些有用的功能点(例如：DistributedCache， IsolationRunner等等)来收尾。

三、 大规模数据排序 1. 简介

使用hadoop进行大量的数据排序排序最直观的方法是把文件所有内容给map之后，map不做任何处理，直接输出给一个reduce，利用hadoop的自己的shuffle机制，对所有数据进行排序，而后由reduce直接输出。

然而这样的方法跟单机毫无差别，完全无法用到多机分布式计算的便利。因此这种方法是不行的。 利用hadoop分而治之的计算模型，可以参照快速排序的思想。在这里我们先简单回忆一下快速排序。快速排序基本步骤就是需要现在所有数据中选取一个作为支点。然后将大于这个支点的放在一边，小于这个支点的放在另一边。

设想如果我们有N个支点(这里可以称为标尺)，就可以把所有的数据分成N+1个part，将这N+1个part丢给reduce，由hadoop自动排序，最后输出N+1个内部有序的文件，再把这N+1个文件首尾相连合并成一个文件，收工。

由此我们可以归纳出这样一个用hadoop对大量数据排序的步骤： ①对待排序数据进行抽样; ②对抽样数据进行排序，产生标尺;

③Map对输入的每条数据计算其处于哪两个标尺之间;将数据发给对应区间ID的reduce ④ Reduce将获得数据直接输出。 这里使用对一组url进行排序来作为例子：

图表 3 url排序

如何将数据发给一个指定ID的reduce?hadoop提供了多种分区算法。这些算法根据map输出的数据的key来确定此数据应该发给哪个reduce(reduce的排序也依赖key)。因此，如果需要将数据发给某个reduce，只要在输出数据的同时，提供一个 key(在上面这个例子中就是reduce的ID+url)，数据就该去哪儿去哪儿了。 2. Nutch

Nutch是一个由Java实现的，刚刚诞生开放源代码(open-source)的web搜索引擎。

Nutch主要分为爬虫crawler和查询searcher两部分。Crawler主要用于从网络上抓取网页并为这些网页建立索引。Searcher主要利用这些索引检索用户的查找关键词来产生查找结果。两者之间的接口是索引，所以除去索引部分，两者之间的耦合度很低。

Crawler的重点在两个方面，Crawler的工作流程和涉及的数据文件的格式和含义。

Crawler的工作原理：首先Crawler根据WebDB生成一个待抓取网页的URL集合叫做Fetchlist，接着下载线程Fetcher根据Fetchlist将网页抓取回来，如果下载线程有很多个，那么就生成很多个Fetchlist，也就是一个Fetcher对应一个Fetchlist。然后Crawler用抓取回来的网页更新WebDB，根据更新后的WebDB生成新的Fetchlist，里面是未抓取的或者新发现的URLs，然后下一轮抓取循环重新开始。 四、 算法分析

1. Sort算法分析 (1) 排序实例

排序实例仅仅用 map/reduce框架来把输入目录排序放到输出目录。输入和输出必须是顺序文件，键和值是BytesWritable. mapper是预先定义的IdentityMapper，reducer 是预先定义的 IdentityReducer， 两个都是把输入直接的输出。要运行这个例 子：bin/hadoop jar hadoop-*-examples.jar sort [-m <#maps>] [-r <#reduces>]

(2) 运行排序基准测试

为了使得排序例子作为一个 基准测试，用 RandomWriter产 生10GB/node 的数据。然后用排序实例来进行排序。这个提供了一个可扩展性依赖于集群的大小的排序基准。默认情况下，排序实例用1.0*capacity作为 reduces的数量，依赖于你的集群的大小你可能会在1.75*capacity的情况下得到更好的结果。 (3) 代码分析 在eclipse中设置参数：

/home/hadoop/rand/part-00000 /home/hadoop/rand-sort

其中/home/hadoop/rand/part-00000 表示输入路径，/home/hadoop/rand-sort表示输出路径。 数据来源

我们这里输入参数中的“/home/hadoop/rand/part-00000”是通过hadoop实例 RandomWriter 这个实例得到的。为了节省时间，hadoop实例 RandomWriter 中得到了两个文件，我们这里指使用了一个文件part-00000。如果要对两个文件都进行排序操作，那么输入路径只需要是目录即可。 Sort算法源代码 a) 源码位置

/local/zkl/hadoop/hadoop-0.20.1/hadoop-0.20.1/src/examples/org/apache/hadoop/examples/Sort.java b) 下面程序是一段关于Sort算法的源代码： 26. * To run: bin/hadoop jar build/hadoop-examples.jar sort 27. * [-m maps] [-r reduces] 28. * [-inFormat input format class] 29. * [-outFormat output format class] 30. * [-outKey output key class]

31. * [-outValue output value class] 32. * [-totalOrder pcntnum samplesmax splits] 33. * in-dirout-dir 34. */

35. public class Sort extends Configured implements Tool { 36. private RunningJob jobResult = null; 37.

166. //input attr:/home/hadoop/rand/part-00000 /home/hadoop/rand-sort 167.

168. public static void main(String[] args) throws Exception { 169. int res = ToolRunner.run(new Configuration(), new Sort(), args); 170. System.exit(res); 171. } 172. 173. /**

174. * Get the last job that was run using this instance. 175. * @return the results of the last job that was run 176. */

177. public RunningJob getResult() { 178. return jobResult; 179. } 180. }

2. Secondsort算法分析 (1) 工作原理