方案1：可以估计每个文件安的大小为50G×64=320G，远远大于内存限制的4G。所以不可能将其完全加载到内存中处理。考虑采取分而治之的方法。

s 遍历文件a，对每个url求取，然后根据所取得的值将url分别存储到1000个小文件（记为）中。这样每个小文件的大约为300M。

s 遍历文件b，采取和a相同的方式将url分别存储到1000各小文件（记为）。这样处理后，所有可能相同的url都在对应的小文件（）中，不对应的小文件不可能有相同的url。然后我们只要求出1000对小文件中相同的url即可。

s 求每对小文件中相同的url时，可以把其中一个小文件的url存储到hash_set中。然后遍历另一个小文件的每个url，看其是否在刚才构建的hash_set中，如果是，那么就是共同的url，存到文件里面就可以了。

方案2：如果允许有一定的错误率，可以使用Bloom filter，4G内存大概可以表示340亿bit。将其中一个文件中的url使用Bloom filter映射为这340亿bit，然后挨个读取另外一个文件的url，检查是否与Bloom filter，如果是，那么该url应该是共同的url（注意会有一定的错误率）。

2. 有10个文件，每个文件1G，每个文件的每一行存放的都是用户的query，每个文件的query都可能重复。要求你按照query的频度排序。

方案1：

s 顺序读取10个文件，按照hash(query)%10的结果将query写入到另外10个文件（记为）中。这样新生成的文件每个的大小大约也1G（假设hash函数是随机的）。

s 找一台内存在2G左右的机器，依次对用hash_map(query, query_count)来统计每个query出现的次数。利用快速/堆/归并排序按照出现次数进行排序。将排序好的query和对应的query_cout输出到文件中。这样得到了10个排好序的文件（记为）。

s 对这10个文件进行归并排序（内排序与外排序相结合）。

方案2：

一般query的总量是有限的，只是重复的次数比较多而已，可能对于所有的query，一次性就可以加入到内存了。这样，我们就可以采用trie树/hash_map等直接来统计每个query出现的次数，然后按出现次数做快速/堆/归并排序就可以了。

方案3：

与方案1类似，但在做完hash，分成多个文件后，可以交给多个文件来处理，采用分布式的架构来处理（比如MapReduce），最后再进行合并。

3. 有一个1G大小的一个文件，里面每一行是一个词，词的大小不超过16字节，内存限制大小是1M。返回频数最高的100个词。

方案1：顺序读文件中，对于每个词x，取，然后按照该值存到5000个小文件（记为）中。这样每个文件大概是200k左右。如果其中的有的文件超过了1M大小，还可以按照类似的方法继续往下分，知道分解得到的小文件的大小都不超过1M。对每个小文件，统计每个文件中出现的词以及相应的频率（可以采用trie树/hash_map等），并取出出现频率最大的100个词（可以用含100个结点的最小堆），并把100词及相应的频率存入文件，这样又得到了5000个文件。下一步就是把这5000个文件进行归并（类似与归并排序）的过程了。

4. 海量日志数据，提取出某日访问百度次数最多的那个IP。

方案1：首先是这一天，并且是访问百度的日志中的IP取出来，逐个写入到一个大文件中。注意到IP是32位的，最多有个IP。同样可以采用映射的方法，比如模1000，把整个大文件映射为1000个小文件，再找出每个小文中出现频率最大的IP（可以采用hash_map进行频率统计，然后再找出频率最大的几个）及相应的频率。然后再在这1000个最大的IP中，找出那个频率最大的IP，即为所求。

5. 在2.5亿个整数中找出不重复的整数，内存不足以容纳这2.5亿个整数。

方案1：采用2-Bitmap（每个数分配2bit，00表示不存在，01表示出现一次，10表示多次，11无意义）进行，共需内存内存，还可以接受。然后扫描这2.5亿个整数，查看Bitmap中相对应位，如果是00变01，01变10，10保持不变。所描完事后，查看bitmap，把对应位是01的整数输出即可。

方案2：也可采用上题类似的方法，进行划分小文件的方法。然后在小文件中找出不重复的整数，并排序。然后再进行归并，注意去除重复的元素。

6. 海量数据分布在100台电脑中，想个办法高校统计出这批数据的TOP10。

方案1：

s 在每台电脑上求出TOP10，可以采用包含10个元素的堆完成（TOP10小，用最大堆，TOP10大，用最小堆）。比如求TOP10大，我们首先取前10个元素调整成最小堆，如果发现，然后扫描后面的数据，并与堆顶元素比较，如果比堆顶元素大，那么用该元素替换堆顶，然后再调整为最小堆。最后堆中的元素就是TOP10大。

s 求出每台电脑上的TOP10后，然后把这100台电脑上的TOP10组合起来，共1000个数据，再利用上面类似的方法求出TOP10就可以了。

7. 怎么在海量数据中找出重复次数最多的一个？

方案1：先做hash，然后求模映射为小文件，求出每个小文件中重复次数最多的一个，并记录重复次数。然后找出上一步求出的数据中重复次数最多的一个就是所求（具体参考前面的题）。

8. 上千万或上亿数据（有重复），统计其中出现次数最多的钱N个数据。

方案1：上千万或上亿的数据，现在的机器的内存应该能存下。所以考虑采用hash_map/搜索二叉树/红黑树等来进行统计次数。然后就是取出前N个出现次数最多的数据了，可以用第6题提到的堆机制完成。

9. 1000万字符串，其中有些是重复的，需要把重复的全部去掉，保留没有重复的字符串。请怎么设计和实现？

方案1：这题用trie树比较合适，hash_map也应该能行。

10. 一个文本文件，大约有一万行，每行一个词，要求统计出其中最频繁出现的前10个词，请给出思想，给出时间复杂度分析。

方案1：这题是考虑时间效率。用trie树统计每个词出现的次数，时间复杂度是O(n*le)（le表示单词的平准长度）。然后是找出出现最频繁的前10个词，可以用堆来实现，前面的题中已经讲到了，时间复杂度是O(n*lg10)。所以总的时间复杂度，是O(n*le)与O(n*lg10)中较大的哪一个。

11. 一个文本文件，找出前10个经常出现的词，但这次文件比较长，说是上亿行或十亿行，总之无法一次读入内存，问最优解。

方案1：首先根据用hash并求模，将文件分解为多个小文件，对于单个文件利用上题的方法求出每个文件件中10个最常出现的词。然后再进行归并处理，找出最终的10个最常出现的词。

12. 100w个数中找出最大的100个数。

方案1：在前面的题中，我们已经提到了，用一个含100个元素的最小堆完成。复杂度为O(100w*lg100)。

方案2：采用快速排序的思想，每次分割之后只考虑比轴大的一部分，知道比轴大的一部分在比100多的时候，采用传统排序算法排序，取前100个。复杂度为O(100w*100)。

方案3：采用局部淘汰法。选取前100个元素，并排序，记为序列L。然后一次扫描剩余的元素x，与排好序的100个元素中最小的元素比，如果比这个最小的要大，那么把这个最小的元素删除，并把x利用插入排序的思想，插入到序列L中。依次循环，知道扫描了所有的元素。复杂度为O(100w*100)。

13. 寻找热门查询：

搜索引擎会通过日志文件把用户每次检索使用的所有检索串都记录下来，每个查询串的长度为1-255字节。假设目前有一千万个记录，这些查询串的重复读比较高，虽然总数是1千万，但是如果去除重复和，不超过3百万个。一个查询串的重复度越高，说明查询它的用户越多，也就越热门。请你统计最热门的10个查询串，要求使用的内存不能超过1G。

(1) 请描述你解决这个问题的思路；

(2) 请给出主要的处理流程，算法，以及算法的复杂度。

方案1：采用trie树，关键字域存该查询串出现的次数，没有出现为0。最后用10个元素的最小推来对出现频率进行排序。

14. 一共有N个机器，每个机器上有N个数。每个机器最多存O(N)个数并对它们操作。如何找到个数中的中数？

方案1：先大体估计一下这些数的范围，比如这里假设这些数都是32位无符号整数（共有个）。我们把0到的整数划分为N个范围段，每个段包含个整数。比如，第一个段位0到，第二段为到，…，第N个段为到。然后，扫描每个机器上的N个数，把属于第一个区段的数放到第一个机器上，属于第二个区段的数放到第二个机器上，…，属于第N个区段的数放到第N个机器上。注意这个过程每个机器上存储的数应该是O(N)的。下面我们依次统计每个机器上数的个数，一次累加，直到找到第k个机器，在该机器上累加的数大于或等于，而在第k-1个机器上的累加数小于，并把这个数记为x。那么我们要找的中位数在第k个机器中，排在第位。然后我们对第k个机器的数排序，并找出第个数，即为所求的中位数。复杂度是的。

方案2：先对每台机器上的数进行排序。排好序后，我们采用归并排序的思想，将这N个机器上的数归并起来得到最终的排序。找到第个便是所求。复杂度是的。

15. 最大间隙问题

给定n个实数，求着n个实数在实轴上向量2个数之间的最大差值，要求线性的时间算法。

方案1：最先想到的方法就是先对这n个数据进行排序，然后一遍扫描即可确定相邻的最大间隙。但该方法不能满足线性时间的要求。故采取如下方法：

s 找到n个数据中最大和最小数据max和min。

s 用n-2个点等分区间[min, max]，即将[min, max]等分为n-1个区间（前闭后开区间），将这些区间看作桶，编号为，且桶的上界和桶i+1的下届相同，即每个桶的大小相同。每个桶的大小为：。实际上，这些桶的边界构成了一个等差数列（首项为min，公差为），且认为将min放入第一个桶，将max放入第n-1个桶。

s 将n个数放入n-1个桶中：将每个元素分配到某个桶（编号为index），其中，并求出分到每个桶的最大最小数据。

s 最大间隙：除最大最小数据max和min以外的n-2个数据放入n-1个桶中，由抽屉原理可知至少有一个桶是空的，又因为每个桶的大小相同，所以最大间隙不会在同一桶中出现，一定是某个桶的上界和气候某个桶的下界之间隙，且该量筒之间的桶（即便好在该连个便好之间的桶）一定是空桶。也就是说，最大间隙在桶i的上界和桶j的下界之间产生。一遍扫描即可完成。

16. 将多个集合合并成没有交集的集合：给定一个字符串的集合，格式如：。要求将其中交集不为空的集合合并，要求合并完成的集合之间无交集，例如上例应输出。

(1) 请描述你解决这个问题的思路；

(2) 给出主要的处理流程，算法，以及算法的复杂度；

(3) 请描述可能的改进。

方案1：采用并查集。首先所有的字符串都在单独的并查集中。然后依扫描每个集合，顺序合并将两个相邻元素合并。例如，对于，首先查看aaa和bbb是否在同一个并查集中，如果不在，那么把它们所在的并查集合并，然后再看bbb和ccc是否在同一个并查集中，如果不在，那么也把它们所在的并查集合并。接下来再扫描其他的集合，当所有的集合都扫描完了，并查集代表的集合便是所求。复杂度应该是O(NlgN)的。改进的话，首先可以记录每个节点的根结点，改进查询。合并的时候，可以把大的和小的进行合，这样也减少复杂度。

17. 最大子序列与最大子矩阵问题

数组的最大子序列问题：给定一个数组，其中元素有正，也有负，找出其中一个连续子序列，使和最大。

方案1：这个问题可以动态规划的思想解决。设表示以第i个元素结尾的最大子序列，那么显然。基于这一点可以很快用代码实现。

最大子矩阵问题：给定一个矩阵（二维数组），其中数据有大有小，请找一个子矩阵，使得子矩阵的和最大，并输出这个和。

方案1：可以采用与最大子序列类似的思想来解决。如果我们确定了选择第i列和第j列之间的元素，那么在这个范围内，其实就是一个最大子序列问题。如何确定第i列和第j列可以词用暴搜的方法进行。

    1. 安装增强功能包(Guest Additions)

    安装好ubuntu后，运行Ubuntu并登录。然后在VirtualBox的菜单里选择"设备(Devices)" -> "安装增强功能包(Install Guest Additions)"。

    你会发现在Ubuntu桌面上多出一个光盘图标，这张光盘默认被自动加载到了文件夹/media/cdom0。进入命令行终端，输入：

    cd /media/cdom0

    sudo ./VboxLinuxAdditions.run

    开始安装工具包。安装完毕后会提示要重启Ubuntu。

    2. 设置共享文件夹

    重启完成后点击"设备(Devices)" -> 共享文件夹(Shared Folders)菜单，添加一个共享文件夹，选项固定和临时是指该文件夹是否是持久的。共享名可以任取一个自己喜欢的，比如"gongxiang"，尽量使用英文名称。

    3. 挂载共享文件夹

    重新进入虚拟Ubuntu，在命令行终端下输入：

    sudo mkdir /mnt/shared

    sudo mount -t vboxsf gongxiang /mnt/shared

    其中"gongxiang"是之前创建的共享文件夹的名字。OK，现在Ubuntu和主机可以互传文件了。

    假如您不想每一次都手动挂载，可以在/etc/fstab中添加一项

    gongxiang /mnt/shared vboxsf rw,gid=100,uid=1000,auto 0 0

    这样就能够自动挂载了。

    4. 卸载的话使用下面的命令：

    sudo umount -f /mnt/shared

    注意：

    共享文件夹的名称千万不要和挂载点的名称相同。比如，上面的挂载点是/mnt/shared，如果共享文件夹的名字也是shared的话，在挂载的时候就会出现如下的错误信息(看http://www.virtualbox.org/ticket/2265)：

    /sbin/mount.vboxsf: mounting failed with the error: Protocol error

    原因分析可以看Tips on running Sun Virtualbox的Shared Folder on a Linux Guest节。

1、副本的存放，副本的存放是HDFS可靠性和性能的关键。HDFS采用一种称为rack-aware的策略来改进数据的可靠性、有效性和网络带宽的利用。这个策略实现的短期目标是验证在生产环境下的表现，观察它的行为，构建测试和研究的基础，以便实现更先进的策略。庞大的HDFS实例一般运行在多个机架的计算机形成的集群上，不同机架间的两台机器的通讯需要通过交换机，显然通常情况下，同一个机架内的两个节点间的带宽会比不同机架间的两台机器的带宽大。
    通过一个称为Rack Awareness的过程，Namenode决定了每个Datanode所属的rack id。一个简单但没有优化的策略就是将副本存放在单独的机架上。这样可以防止整个机架（非副本存放）失效的情况，并且允许读数据的时候可以从多个机架读取。这个简单策略设置可以将副本分布在集群中，有利于组件失败情况下的负载均衡。但是，这个简单策略加大了写的代价，因为一个写操作需要传输block到多个机架。
    在大多数情况下，replication因子是3，HDFS的存放策略是将一个副本存放在本地机架上的节点，一个副本放在同一机架上的另一个节点，最后一个副本放在不同机架上的一个节点。机架的错误远远比节点的错误少，这个策略不会影响到数据的可靠性和有效性。三分之一的副本在一个节点上，三分之二在一个机架上，其他保存在剩下的机架中，这一策略改进了写的性能。

2、副本的选择，为了降低整体的带宽消耗和读延时，HDFS会尽量让reader读最近的副本。如果在reader的同一个机架上有一个副本，那么就读该副本。如果一个HDFS集群跨越多个数据中心，那么reader也将首先尝试读本地数据中心的副本。

3、SafeMode
    Namenode启动后会进入一个称为SafeMode的特殊状态，处在这个状态的Namenode是不会进行数据块的复制的。Namenode从所有的 Datanode接收心跳包和Blockreport。Blockreport包括了某个Datanode所有的数据块列表。每个block都有指定的最小数目的副本。当Namenode检测确认某个Datanode的数据块副本的最小数目，那么该Datanode就会被认为是安全的；如果一定百分比（这个参数可配置）的数据块检测确认是安全的，那么Namenode将退出SafeMode状态，接下来它会确定还有哪些数据块的副本没有达到指定数目，并将这些block复制到其他Datanode。

五、文件系统元数据的持久化
    Namenode存储HDFS的元数据。对于任何对文件元数据产生修改的操作，Namenode都使用一个称为Editlog的事务日志记录下来。例如，在HDFS中创建一个文件，Namenode就会在Editlog中插入一条记录来表示；同样，修改文件的replication因子也将往 Editlog插入一条记录。Namenode在本地OS的文件系统中存储这个Editlog。整个文件系统的namespace，包括block到文件的映射、文件的属性，都存储在称为FsImage的文件中，这个文件也是放在Namenode所在系统的文件系统上。
    Namenode在内存中保存着整个文件系统namespace和文件Blockmap的映像。这个关键的元数据设计得很紧凑，因而一个带有4G内存的 Namenode足够支撑海量的文件和目录。当Namenode启动时，它从硬盘中读取Editlog和FsImage，将所有Editlog中的事务作用（apply)在内存中的FsImage ，并将这个新版本的FsImage从内存中flush到硬盘上,然后再truncate这个旧的Editlog，因为这个旧的Editlog的事务都已经作用在FsImage上了。这个过程称为checkpoint。在当前实现中，checkpoint只发生在Namenode启动时，在不久的将来我们将实现支持周期性的checkpoint。
    Datanode并不知道关于文件的任何东西，除了将文件中的数据保存在本地的文件系统上。它把每个HDFS数据块存储在本地文件系统上隔离的文件中。 Datanode并不在同一个目录创建所有的文件，相反，它用启发式地方法来确定每个目录的最佳文件数目，并且在适当的时候创建子目录。在同一个目录创建所有的文件不是最优的选择，因为本地文件系统可能无法高效地在单一目录中支持大量的文件。当一个Datanode启动时，它扫描本地文件系统，对这些本地文件产生相应的一个所有HDFS数据块的列表，然后发送报告到Namenode，这个报告就是Blockreport。

六、通讯协议
    所有的HDFS通讯协议都是构建在TCP/IP协议上。客户端通过一个可配置的端口连接到Namenode，通过ClientProtocol与 Namenode交互。而Datanode是使用DatanodeProtocol与Namenode交互。从ClientProtocol和 Datanodeprotocol抽象出一个远程调用(RPC），在设计上，Namenode不会主动发起RPC，而是是响应来自客户端和 Datanode 的RPC请求。

七、健壮性
    HDFS的主要目标就是实现在失败情况下的数据存储可靠性。常见的三种失败：Namenode failures, Datanode failures和网络分割（network partitions)。
1、硬盘数据错误、心跳检测和重新复制
    每个Datanode节点都向Namenode周期性地发送心跳包。网络切割可能导致一部分Datanode跟Namenode失去联系。 Namenode通过心跳包的缺失检测到这一情况，并将这些Datanode标记为dead，不会将新的IO请求发给它们。寄存在dead Datanode上的任何数据将不再有效。Datanode的死亡可能引起一些block的副本数目低于指定值，Namenode不断地跟踪需要复制的 block，在任何需要的情况下启动复制。在下列情况可能需要重新复制：某个Datanode节点失效，某个副本遭到损坏，Datanode上的硬盘错误，或者文件的replication因子增大。

2、集群均衡
   HDFS支持数据的均衡计划，如果某个Datanode节点上的空闲空间低于特定的临界点，那么就会启动一个计划自动地将数据从一个Datanode搬移到空闲的Datanode。当对某个文件的请求突然增加，那么也可能启动一个计划创建该文件新的副本，并分布到集群中以满足应用的要求。这些均衡计划目前还没有实现。

3、数据完整性
  从某个Datanode获取的数据块有可能是损坏的，这个损坏可能是由于Datanode的存储设备错误、网络错误或者软件bug造成的。HDFS客户端软件实现了HDFS文件内容的校验和。当某个客户端创建一个新的HDFS文件，会计算这个文件每个block的校验和，并作为一个单独的隐藏文件保存这些校验和在同一个HDFS namespace下。当客户端检索文件内容，它会确认从Datanode获取的数据跟相应的校验和文件中的校验和是否匹配，如果不匹配，客户端可以选择从其他Datanode获取该block的副本。

4、元数据磁盘错误
    FsImage和Editlog是HDFS的核心数据结构。这些文件如果损坏了，整个HDFS实例都将失效。因而，Namenode可以配置成支持维护多个FsImage和Editlog的拷贝。任何对FsImage或者Editlog的修改，都将同步到它们的副本上。这个同步操作可能会降低 Namenode每秒能支持处理的namespace事务。这个代价是可以接受的，因为HDFS是数据密集的，而非元数据密集。当Namenode重启的时候，它总是选取最近的一致的FsImage和Editlog使用。
   Namenode在HDFS是单点存在，如果Namenode所在的机器错误，手工的干预是必须的。目前，在另一台机器上重启因故障而停止服务的Namenode这个功能还没实现。

5、快照
   快照支持某个时间的数据拷贝，当HDFS数据损坏的时候，可以恢复到过去一个已知正确的时间点。HDFS目前还不支持快照功能。

八、数据组织
1、数据块
    兼容HDFS的应用都是处理大数据集合的。这些应用都是写数据一次，读却是一次到多次，并且读的速度要满足流式读。HDFS支持文件的write- once-read-many语义。一个典型的block大小是64MB，因而，文件总是按照64M切分成chunk，每个chunk存储于不同的 Datanode
2、步骤
    某个客户端创建文件的请求其实并没有立即发给Namenode，事实上，HDFS客户端会将文件数据缓存到本地的一个临时文件。应用的写被透明地重定向到这个临时文件。当这个临时文件累积的数据超过一个block的大小（默认64M)，客户端才会联系Namenode。Namenode将文件名插入文件系统的层次结构中，并且分配一个数据块给它，然后返回Datanode的标识符和目标数据块给客户端。客户端将本地临时文件flush到指定的 Datanode上。当文件关闭时，在临时文件中剩余的没有flush的数据也会传输到指定的Datanode，然后客户端告诉Namenode文件已经关闭。此时Namenode才将文件创建操作提交到持久存储。如果Namenode在文件关闭前挂了，该文件将丢失。
   上述方法是对通过对HDFS上运行的目标应用认真考虑的结果。如果不采用客户端缓存，由于网络速度和网络堵塞会对吞估量造成比较大的影响。

3、流水线复制
    当某个客户端向HDFS文件写数据的时候，一开始是写入本地临时文件，假设该文件的replication因子设置为3，那么客户端会从Namenode 获取一张Datanode列表来存放副本。然后客户端开始向第一个Datanode传输数据，第一个Datanode一小部分一小部分（4kb)地接收数据，将每个部分写入本地仓库，并且同时传输该部分到第二个Datanode节点。第二个Datanode也是这样，边收边传，一小部分一小部分地收，存储在本地仓库，同时传给第三个Datanode，第三个Datanode就仅仅是接收并存储了。这就是流水线式的复制。

九、可访问性
    HDFS给应用提供了多种访问方式，可以通过DFSShell通过命令行与HDFS数据进行交互，可以通过java API调用，也可以通过C语言的封装API访问，并且提供了浏览器访问的方式。正在开发通过WebDav协议访问的方式。具体使用参考文档。
十、空间的回收
1、文件的删除和恢复
    用户或者应用删除某个文件，这个文件并没有立刻从HDFS中删除。相反，HDFS将这个文件重命名，并转移到/trash目录。当文件还在/trash目录时，该文件可以被迅速地恢复。文件在/trash中保存的时间是可配置的，当超过这个时间，Namenode就会将该文件从namespace中删除。文件的删除，也将释放关联该文件的数据块。注意到，在文件被用户删除和HDFS空闲空间的增加之间会有一个等待时间延迟。
    当被删除的文件还保留在/trash目录中的时候，如果用户想恢复这个文件，可以检索浏览/trash目录并检索该文件。/trash目录仅仅保存被删除文件的最近一次拷贝。/trash目录与其他文件目录没有什么不同，除了一点：HDFS在该目录上应用了一个特殊的策略来自动删除文件，目前的默认策略是删除保留超过6小时的文件，这个策略以后会定义成可配置的接口。

2、Replication因子的减小
    当某个文件的replication因子减小，Namenode会选择要删除的过剩的副本。下次心跳检测就将该信息传递给Datanode， Datanode就会移除相应的block并释放空间，同样，在调用setReplication方法和集群中的空闲空间增加之间会有一个时间延迟。

参考资料：
HDFS Java API: http://hadoop.apache.org/core/docs/current/api/
HDFS source code: http://hadoop.apache.org/core/version_control.html

先决条件
请先确认Hadoop被正确安装、配置和正常运行中。更多信息见：

Hadoop快速入门对初次使用者。
Hadoop集群搭建对大规模分布式集群。
概述
Hadoop Map/Reduce是一个使用简易的软件框架，基于它写出来的应用程序能够运行在由上千个商用机器组成的大型集群上，并以一种可靠容错的方式并行处理上T级别的数据集。

一个Map/Reduce 作业（job） 通常会把输入的数据集切分为若干独立的数据块，由 map任务（task）以完全并行的方式处理它们。框架会对map的输出先进行排序， 然后把结果输入给reduce任务。通常作业的输入和输出都会被存储在文件系统中。 整个框架负责任务的调度和监控，以及重新执行已经失败的任务。

通常，Map/Reduce框架和分布式文件系统是运行在一组相同的节点上的，也就是说，计算节点和存储节点通常在一起。这种配置允许框架在那些已经存好数据的节点上高效地调度任务，这可以使整个集群的网络带宽被非常高效地利用。

Map/Reduce框架由一个单独的master JobTracker 和每个集群节点一个slave TaskTracker共同组成。master负责调度构成一个作业的所有任务，这些任务分布在不同的slave上，master监控它们的执行，重新执行已经失败的任务。而slave仅负责执行由master指派的任务。

应用程序至少应该指明输入/输出的位置（路径），并通过实现合适的接口或抽象类提供map和reduce函数。再加上其他作业的参数，就构成了作业配置（job configuration）。然后，Hadoop的 job client提交作业（jar包/可执行程序等）和配置信息给JobTracker，后者负责分发这些软件和配置信息给slave、调度任务并监控它们的执行，同时提供状态和诊断信息给job-client。

虽然Hadoop框架是用JavaTM实现的，但Map/Reduce应用程序则不一定要用 Java来写 。

Hadoop Streaming是一种运行作业的实用工具，它允许用户创建和运行任何可执行程序 （例如：Shell工具）来做为mapper和reducer。
Hadoop Pipes是一个与SWIG兼容的C++ API （没有基于JNITM技术），它也可用于实现Map/Reduce应用程序。
输入与输出
Map/Reduce框架运转在<key, value> 键值对上，也就是说， 框架把作业的输入看为是一组<key, value> 键值对，同样也产出一组 <key, value> 键值对做为作业的输出，这两组键值对的类型可能不同。

框架需要对key和value的类(classes)进行序列化操作， 因此，这些类需要实现 Writable接口。 另外，为了方便框架执行排序操作，key类必须实现 WritableComparable接口。

一个Map/Reduce 作业的输入和输出类型如下所示：

(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)

例子：WordCount v1.0
在深入细节之前，让我们先看一个Map/Reduce的应用示例，以便对它们的工作方式有一个初步的认识。

WordCount是一个简单的应用，它可以计算出指定数据集中每一个单词出现的次数。

这个应用适用于 单机模式， 伪分布式模式 或 完全分布式模式 三种Hadoop安装方式。

源代码
  WordCount.java
1. package org.myorg;
2.  
3. import java.io.IOException;
4. import java.util.*;
5.  
6. import org.apache.hadoop.fs.Path;
7. import org.apache.hadoop.conf.*;
8. import org.apache.hadoop.io.*;
9. import org.apache.hadoop.mapred.*;
10. import org.apache.hadoop.util.*;
11.  
12. public class WordCount {
13.  
14.    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
15.      private final static IntWritable one = new IntWritable(1);
16.      private Text word = new Text();
17.  
18.      public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
19.        String line = value.toString();
20.        StringTokenizer tokenizer = new StringTokenizer(line);
21.        while (tokenizer.hasMoreTokens()) {
22.          word.set(tokenizer.nextToken());
23.          output.collect(word, one);
24.        }
25.      }
26.    }
27.  
28.    public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
29.      public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
30.        int sum = 0;
31.        while (values.hasNext()) {
32.          sum += values.next().get();
33.        }
34.        output.collect(key, new IntWritable(sum));
35.      }
36.    }
37.  
38.    public static void main(String[] args) throws Exception {
39.      JobConf conf = new JobConf(WordCount.class);
40.      conf.setJobName("wordcount");
41.  
42.      conf.setOutputKeyClass(Text.class);
43.      conf.setOutputValueClass(IntWritable.class);
44.  
45.      conf.setMapperClass(Map.class);
46.      conf.setCombinerClass(Reduce.class);
47.      conf.setReducerClass(Reduce.class);
48.  
49.      conf.setInputFormat(TextInputFormat.class);
50.      conf.setOutputFormat(TextOutputFormat.class);
51.  
52.      FileInputFormat.setInputPaths(conf, new Path(args[0]));
53.      FileOutputFormat.setOutputPath(conf, new Path(args[1]));
54.  
55.      JobClient.runJob(conf);
57.    }
58. }
59.  

用法
假设环境变量HADOOP_HOME对应安装时的根目录，HADOOP_VERSION对应Hadoop的当前安装版本，编译WordCount.java来创建jar包，可如下操作：

$ mkdir wordcount_classes
$ javac -classpath ${HADOOP_HOME}/hadoop-${HADOOP_VERSION}-core.jar -d wordcount_classes WordCount.java
$ jar -cvf /usr/joe/wordcount.jar -C wordcount_classes/ .

假设：

/usr/joe/wordcount/input - 是HDFS中的输入路径
/usr/joe/wordcount/output - 是HDFS中的输出路径
用示例文本文件做为输入：

$ bin/hadoop dfs -ls /usr/joe/wordcount/input/
/usr/joe/wordcount/input/file01
/usr/joe/wordcount/input/file02

$ bin/hadoop dfs -cat /usr/joe/wordcount/input/file01
Hello World Bye World

$ bin/hadoop dfs -cat /usr/joe/wordcount/input/file02
Hello Hadoop Goodbye Hadoop

运行应用程序：

$ bin/hadoop jar /usr/joe/wordcount.jar org.myorg.WordCount /usr/joe/wordcount/input /usr/joe/wordcount/output

输出是：

$ bin/hadoop dfs -cat /usr/joe/wordcount/output/part-00000
Bye 1
Goodbye 1
Hadoop 2
Hello 2
World 2

应用程序能够使用-files选项来指定一个由逗号分隔的路径列表，这些路径是task的当前工作目录。使用选项-libjars可以向map和reduce的classpath中添加jar包。使用-archives选项程序可以传递档案文件做为参数，这些档案文件会被解压并且在task的当前工作目录下会创建一个指向解压生成的目录的符号链接（以压缩包的名字命名）。 有关命令行选项的更多细节请参考 Commands manual。

使用-libjars和-files运行wordcount例子：
hadoop jar hadoop-examples.jar wordcount -files cachefile.txt -libjars mylib.jar input output

解释
WordCount应用程序非常直截了当。

Mapper(14-26行)中的map方法(18-25行)通过指定的 TextInputFormat(49行)一次处理一行。然后，它通过StringTokenizer 以空格为分隔符将一行切分为若干tokens，之后，输出< <word>, 1> 形式的键值对。

对于示例中的第一个输入，map输出是：
< Hello, 1>
< World, 1>
< Bye, 1>
< World, 1>

第二个输入，map输出是：
< Hello, 1>
< Hadoop, 1>
< Goodbye, 1>
< Hadoop, 1>

关于组成一个指定作业的map数目的确定，以及如何以更精细的方式去控制这些map，我们将在教程的后续部分学习到更多的内容。

WordCount还指定了一个combiner (46行)。因此，每次map运行之后，会对输出按照key进行排序，然后把输出传递给本地的combiner（按照作业的配置与Reducer一样），进行本地聚合。

第一个map的输出是：
< Bye, 1>
< Hello, 1>
< World, 2>

第二个map的输出是：
< Goodbye, 1>
< Hadoop, 2>
< Hello, 1>

Reducer(28-36行)中的reduce方法(29-35行) 仅是将每个key（本例中就是单词）出现的次数求和。

因此这个作业的输出就是：
< Bye, 1>
< Goodbye, 1>
< Hadoop, 2>
< Hello, 2>
< World, 2>

代码中的run方法中指定了作业的几个方面， 例如：通过命令行传递过来的输入/输出路径、key/value的类型、输入/输出的格式等等JobConf中的配置信息。随后程序调用了JobClient.runJob(55行)来提交作业并且监控它的执行。

我们将在本教程的后续部分学习更多的关于JobConf， JobClient， Tool和其他接口及类(class)。

Map/Reduce - 用户界面
这部分文档为用户将会面临的Map/Reduce框架中的各个环节提供了适当的细节。这应该会帮助用户更细粒度地去实现、配置和调优作业。然而，请注意每个类/接口的javadoc文档提供最全面的文档；本文只是想起到指南的作用。

我们会先看看Mapper和Reducer接口。应用程序通常会通过提供map和reduce方法来实现它们。

然后，我们会讨论其他的核心接口，其中包括： JobConf，JobClient，Partitioner， OutputCollector，Reporter， InputFormat，OutputFormat等等。

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

核心功能描述
应用程序通常会通过提供map和reduce来实现 Mapper和Reducer接口，它们组成作业的核心。

Mapper
Mapper将输入键值对(key/value pair)映射到一组中间格式的键值对集合。

Map是一类将输入记录集转换为中间格式记录集的独立任务。 这种转换的中间格式记录集不需要与输入记录集的类型一致。一个给定的输入键值对可以映射成0个或多个输出键值对。

Hadoop Map/Reduce框架为每一个InputSplit产生一个map任务，而每个InputSplit是由该作业的InputFormat产生的。

概括地说，对Mapper的实现者需要重写 JobConfigurable.configure(JobConf)方法，这个方法需要传递一个JobConf参数，目的是完成Mapper的初始化工作。然后，框架为这个任务的InputSplit中每个键值对调用一次 map(WritableComparable, Writable, OutputCollector, Reporter)操作。应用程序可以通过重写Closeable.close()方法来执行相应的清理工作。

输出键值对不需要与输入键值对的类型一致。一个给定的输入键值对可以映射成0个或多个输出键值对。通过调用 OutputCollector.collect(WritableComparable,Writable)可以收集输出的键值对。

应用程序可以使用Reporter报告进度，设定应用级别的状态消息，更新Counters（计数器），或者仅是表明自己运行正常。

框架随后会把与一个特定key关联的所有中间过程的值（value）分成组，然后把它们传给Reducer以产出最终的结果。用户可以通过 JobConf.setOutputKeyComparatorClass(Class)来指定具体负责分组的 Comparator。

Mapper的输出被排序后，就被划分给每个Reducer。分块的总数目和一个作业的reduce任务的数目是一样的。用户可以通过实现自定义的 Partitioner来控制哪个key被分配给哪个 Reducer。

用户可选择通过 JobConf.setCombinerClass(Class)指定一个combiner，它负责对中间过程的输出进行本地的聚集，这会有助于降低从Mapper到 Reducer数据传输量。

这些被排好序的中间过程的输出结果保存的格式是(key-len, key, value-len, value)，应用程序可以通过JobConf控制对这些中间结果是否进行压缩以及怎么压缩，使用哪种 CompressionCodec。

需要多少个Map？
Map的数目通常是由输入数据的大小决定的，一般就是所有输入文件的总块（block）数。

Map正常的并行规模大致是每个节点（node）大约10到100个map，对于CPU 消耗较小的map任务可以设到300个左右。由于每个任务初始化需要一定的时间，因此，比较合理的情况是map执行的时间至少超过1分钟。

这样，如果你输入10TB的数据，每个块（block）的大小是128MB，你将需要大约82,000个map来完成任务，除非使用 setNumMapTasks(int)（注意：这里仅仅是对框架进行了一个提示(hint)，实际决定因素见这里）将这个数值设置得更高。

Reducer
Reducer将与一个key关联的一组中间数值集归约（reduce）为一个更小的数值集。

用户可以通过 JobConf.setNumReduceTasks(int)设定一个作业中reduce任务的数目。

概括地说，对Reducer的实现者需要重写 JobConfigurable.configure(JobConf)方法，这个方法需要传递一个JobConf参数，目的是完成Reducer的初始化工作。然后，框架为成组的输入数据中的每个<key, (list of values)>对调用一次 reduce(WritableComparable, Iterator, OutputCollector, Reporter)方法。之后，应用程序可以通过重写Closeable.close()来执行相应的清理工作。

Reducer有3个主要阶段：shuffle、sort和reduce。

Shuffle
Reducer的输入就是Mapper已经排好序的输出。在这个阶段，框架通过HTTP为每个Reducer获得所有Mapper输出中与之相关的分块。

Sort
这个阶段，框架将按照key的值对Reducer的输入进行分组 （因为不同mapper的输出中可能会有相同的key）。

Shuffle和Sort两个阶段是同时进行的；map的输出也是一边被取回一边被合并的。

Secondary Sort
如果需要中间过程对key的分组规则和reduce前对key的分组规则不同，那么可以通过 JobConf.setOutputValueGroupingComparator(Class)来指定一个Comparator。再加上 JobConf.setOutputKeyComparatorClass(Class)可用于控制中间过程的key如何被分组，所以结合两者可以实现按值的二次排序。

Reduce
在这个阶段，框架为已分组的输入数据中的每个 <key, (list of values)>对调用一次 reduce(WritableComparable, Iterator, OutputCollector, Reporter)方法。

Reduce任务的输出通常是通过调用 OutputCollector.collect(WritableComparable, Writable)写入 文件系统的。

应用程序可以使用Reporter报告进度，设定应用程序级别的状态消息，更新Counters（计数器），或者仅是表明自己运行正常。

Reducer的输出是没有排序的。

需要多少个Reduce？
Reduce的数目建议是0.95或1.75乘以 (<no. of nodes> * mapred.tasktracker.reduce.tasks.maximum)。

用0.95，所有reduce可以在maps一完成时就立刻启动，开始传输map的输出结果。用1.75，速度快的节点可以在完成第一轮reduce任务后，可以开始第二轮，这样可以得到比较好的负载均衡的效果。

增加reduce的数目会增加整个框架的开销，但可以改善负载均衡，降低由于执行失败带来的负面影响。

上述比例因子比整体数目稍小一些是为了给框架中的推测性任务（speculative-tasks） 或失败的任务预留一些reduce的资源。

无Reducer
如果没有归约要进行，那么设置reduce任务的数目为零是合法的。

这种情况下，map任务的输出会直接被写入由 setOutputPath(Path)指定的输出路径。框架在把它们写入FileSystem之前没有对它们进行排序。

Partitioner
Partitioner用于划分键值空间（key space）。

Partitioner负责控制map输出结果key的分割。Key（或者一个key子集）被用于产生分区，通常使用的是Hash函数。分区的数目与一个作业的reduce任务的数目是一样的。因此，它控制将中间过程的key（也就是这条记录）应该发送给m个reduce任务中的哪一个来进行reduce操作。

HashPartitioner是默认的 Partitioner。

Reporter
Reporter是用于Map/Reduce应用程序报告进度，设定应用级别的状态消息， 更新Counters（计数器）的机制。

Mapper和Reducer的实现可以利用Reporter 来报告进度，或者仅是表明自己运行正常。在那种应用程序需要花很长时间处理个别键值对的场景中，这种机制是很关键的，因为框架可能会以为这个任务超时了，从而将它强行杀死。另一个避免这种情况发生的方式是，将配置参数mapred.task.timeout设置为一个足够高的值（或者干脆设置为零，则没有超时限制了）。

应用程序可以用Reporter来更新Counter（计数器）。

OutputCollector
OutputCollector是一个Map/Reduce框架提供的用于收集 Mapper或Reducer输出数据的通用机制 （包括中间输出结果和作业的输出结果）。

Hadoop Map/Reduce框架附带了一个包含许多实用型的mapper、reducer和partitioner 的类库。

作业配置
JobConf代表一个Map/Reduce作业的配置。

JobConf是用户向Hadoop框架描述一个Map/Reduce作业如何执行的主要接口。框架会按照JobConf描述的信息忠实地去尝试完成这个作业，然而：

一些参数可能会被管理者标记为 final，这意味它们不能被更改。
一些作业的参数可以被直截了当地进行设置（例如： setNumReduceTasks(int)），而另一些参数则与框架或者作业的其他参数之间微妙地相互影响，并且设置起来比较复杂（例如： setNumMapTasks(int)）。
通常，JobConf会指明Mapper、Combiner(如果有的话)、 Partitioner、Reducer、InputFormat和 OutputFormat的具体实现。JobConf还能指定一组输入文件 (setInputPaths(JobConf, Path...) /addInputPath(JobConf, Path)) 和(setInputPaths(JobConf, String) /addInputPaths(JobConf, String)) 以及输出文件应该写在哪儿 (setOutputPath(Path))。

JobConf可选择地对作业设置一些高级选项，例如：设置Comparator； 放到DistributedCache上的文件；中间结果或者作业输出结果是否需要压缩以及怎么压缩； 利用用户提供的脚本(setMapDebugScript(String)/setReduceDebugScript(String)) 进行调试；作业是否允许预防性（speculative）任务的执行 (setMapSpeculativeExecution(boolean))/(setReduceSpeculativeExecution(boolean)) ；每个任务最大的尝试次数 (setMaxMapAttempts(int)/setMaxReduceAttempts(int)) ；一个作业能容忍的任务失败的百分比 (setMaxMapTaskFailuresPercent(int)/setMaxReduceTaskFailuresPercent(int)) ；等等。

当然，用户能使用 set(String, String)/get(String, String) 来设置或者取得应用程序需要的任意参数。然而，DistributedCache的使用是面向大规模只读数据的。

任务的执行和环境
TaskTracker是在一个单独的jvm上以子进程的形式执行 Mapper/Reducer任务（Task）的。

子任务会继承父TaskTracker的环境。用户可以通过JobConf中的 mapred.child.java.opts配置参数来设定子jvm上的附加选项，例如： 通过-Djava.library.path=<> 将一个非标准路径设为运行时的链接用以搜索共享库，等等。如果mapred.child.java.opts包含一个符号@taskid@， 它会被替换成map/reduce的taskid的值。

下面是一个包含多个参数和替换的例子，其中包括：记录jvm GC日志； JVM JMX代理程序以无密码的方式启动，这样它就能连接到jconsole上，从而可以查看子进程的内存和线程，得到线程的dump；还把子jvm的最大堆尺寸设置为512MB， 并为子jvm的java.library.path添加了一个附加路径。

<property>
  <name>mapred.child.java.opts</name>
  <value>
     -Xmx512M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@taskid@.gc
     -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false
  </value>
</property>

用户或管理员也可以使用mapred.child.ulimit设定运行的子任务的最大虚拟内存。mapred.child.ulimit的值以（KB)为单位，并且必须大于或等于-Xmx参数传给JavaVM的值，否则VM会无法启动。

注意：mapred.child.java.opts只用于设置task tracker启动的子任务。为守护进程设置内存选项请查看 cluster_setup.html

${mapred.local.dir}/taskTracker/是task tracker的本地目录， 用于创建本地缓存和job。它可以指定多个目录（跨越多个磁盘），文件会半随机的保存到本地路径下的某个目录。当job启动时，task tracker根据配置文档创建本地job目录，目录结构如以下所示：

${mapred.local.dir}/taskTracker/archive/ :分布式缓存。这个目录保存本地的分布式缓存。因此本地分布式缓存是在所有task和job间共享的。
${mapred.local.dir}/taskTracker/jobcache/$jobid/ : 本地job目录。
${mapred.local.dir}/taskTracker/jobcache/$jobid/work/: job指定的共享目录。各个任务可以使用这个空间做为暂存空间，用于它们之间共享文件。这个目录通过job.local.dir 参数暴露给用户。这个路径可以通过API JobConf.getJobLocalDir()来访问。它也可以被做为系统属性获得。因此，用户（比如运行streaming）可以调用System.getProperty("job.local.dir")获得该目录。
${mapred.local.dir}/taskTracker/jobcache/$jobid/jars/: 存放jar包的路径，用于存放作业的jar文件和展开的jar。job.jar是应用程序的jar文件，它会被自动分发到各台机器，在task启动前会被自动展开。使用api JobConf.getJar() 函数可以得到job.jar的位置。使用JobConf.getJar().getParent()可以访问存放展开的jar包的目录。
${mapred.local.dir}/taskTracker/jobcache/$jobid/job.xml： 一个job.xml文件，本地的通用的作业配置文件。
${mapred.local.dir}/taskTracker/jobcache/$jobid/$taskid： 每个任务有一个目录task-id，它里面有如下的目录结构：
${mapred.local.dir}/taskTracker/jobcache/$jobid/$taskid/job.xml： 一个job.xml文件，本地化的任务作业配置文件。任务本地化是指为该task设定特定的属性值。这些值会在下面具体说明。
${mapred.local.dir}/taskTracker/jobcache/$jobid/$taskid/output 一个存放中间过程的输出文件的目录。它保存了由framwork产生的临时map reduce数据，比如map的输出文件等。
${mapred.local.dir}/taskTracker/jobcache/$jobid/$taskid/work： task的当前工作目录。
${mapred.local.dir}/taskTracker/jobcache/$jobid/$taskid/work/tmp： task的临时目录。（用户可以设定属性mapred.child.tmp 来为map和reduce task设定临时目录。缺省值是./tmp。如果这个值不是绝对路径， 它会把task的工作路径加到该路径前面作为task的临时文件路径。如果这个值是绝对路径则直接使用这个值。 如果指定的目录不存在，会自动创建该目录。之后，按照选项 -Djava.io.tmpdir='临时文件的绝对路径'执行java子任务。 pipes和streaming的临时文件路径是通过环境变量TMPDIR='the absolute path of the tmp dir'设定的）。 如果mapred.child.tmp有./tmp值，这个目录会被创建。
下面的属性是为每个task执行时使用的本地参数，它们保存在本地化的任务作业配置文件里：

名称 类型 描述
mapred.job.id String job id
mapred.jar String job目录下job.jar的位置
job.local.dir String job指定的共享存储空间
mapred.tip.id String task id
mapred.task.id String task尝试id
mapred.task.is.map boolean 是否是map task
mapred.task.partition int task在job中的id
map.input.file String map读取的文件名
map.input.start long map输入的数据块的起始位置偏移
map.input.length long map输入的数据块的字节数
mapred.work.output.dir String task临时输出目录

task的标准输出和错误输出流会被读到TaskTracker中，并且记录到 ${HADOOP_LOG_DIR}/userlogs

DistributedCache 可用于map或reduce task中分发jar包和本地库。子jvm总是把 当前工作目录 加到 java.library.path 和 LD_LIBRARY_PATH。 因此，可以通过 System.loadLibrary或 System.load装载缓存的库。有关使用分布式缓存加载共享库的细节请参考 native_libraries.html

作业的提交与监控
JobClient是用户提交的作业与JobTracker交互的主要接口。

JobClient 提供提交作业，追踪进程，访问子任务的日志记录，获得Map/Reduce集群状态信息等功能。

作业提交过程包括：

检查作业输入输出样式细节
为作业计算InputSplit值。
如果需要的话，为作业的DistributedCache建立必须的统计信息。
拷贝作业的jar包和配置文件到FileSystem上的Map/Reduce系统目录下。
提交作业到JobTracker并且监控它的状态。
作业的历史文件记录到指定目录的"_logs/history/"子目录下。这个指定目录由hadoop.job.history.user.location设定，默认是作业输出的目录。因此默认情况下，文件会存放在mapred.output.dir/_logs/history目录下。用户可以设置hadoop.job.history.user.location为none来停止日志记录。

用户使用下面的命令可以看到在指定目录下的历史日志记录的摘要。
$ bin/hadoop job -history output-dir
这个命令会打印出作业的细节，以及失败的和被杀死的任务细节。
要查看有关作业的更多细节例如成功的任务、每个任务尝试的次数（task attempt）等，可以使用下面的命令
$ bin/hadoop job -history all output-dir

用户可以使用 OutputLogFilter 从输出目录列表中筛选日志文件。

一般情况，用户利用JobConf创建应用程序并配置作业属性， 然后用 JobClient 提交作业并监视它的进程。

作业的控制
有时候，用一个单独的Map/Reduce作业并不能完成一个复杂的任务，用户也许要链接多个Map/Reduce作业才行。这是容易实现的，因为作业通常输出到分布式文件系统上的，所以可以把这个作业的输出作为下一个作业的输入实现串联。

然而，这也意味着，确保每一作业完成(成功或失败)的责任就直接落在了客户身上。在这种情况下，可以用的控制作业的选项有：

runJob(JobConf)：提交作业，仅当作业完成时返回。
submitJob(JobConf)：只提交作业，之后需要你轮询它返回的 RunningJob句柄的状态，并根据情况调度。
JobConf.setJobEndNotificationURI(String)：设置一个作业完成通知，可避免轮询。
作业的输入
InputFormat 为Map/Reduce作业描述输入的细节规范。

Map/Reduce框架根据作业的InputFormat来：

检查作业输入的有效性。
把输入文件切分成多个逻辑InputSplit实例， 并把每一实例分别分发给一个 Mapper。
提供RecordReader的实现，这个RecordReader从逻辑InputSplit中获得输入记录， 这些记录将由Mapper处理。
基于文件的InputFormat实现（通常是 FileInputFormat的子类） 默认行为是按照输入文件的字节大小，把输入数据切分成逻辑分块（logical InputSplit ）。 其中输入文件所在的FileSystem的数据块尺寸是分块大小的上限。下限可以设置mapred.min.split.size 的值。

考虑到边界情况，对于很多应用程序来说，很明显按照文件大小进行逻辑分割是不能满足需求的。 在这种情况下，应用程序需要实现一个RecordReader来处理记录的边界并为每个任务提供一个逻辑分块的面向记录的视图。

TextInputFormat 是默认的InputFormat。

如果一个作业的Inputformat是TextInputFormat， 并且框架检测到输入文件的后缀是.gz和.lzo，就会使用对应的CompressionCodec自动解压缩这些文件。 但是需要注意，上述带后缀的压缩文件不会被切分，并且整个压缩文件会分给一个mapper来处理。

InputSplit
InputSplit 是一个单独的Mapper要处理的数据块。

一般的InputSplit 是字节样式输入，然后由RecordReader处理并转化成记录样式。

FileSplit 是默认的InputSplit。 它把 map.input.file 设定为输入文件的路径，输入文件是逻辑分块文件。

RecordReader
RecordReader 从InputSlit读入<key, value>对。

一般的，RecordReader 把由InputSplit 提供的字节样式的输入文件，转化成由Mapper处理的记录样式的文件。 因此RecordReader负责处理记录的边界情况和把数据表示成keys/values对形式。

作业的输出
OutputFormat 描述Map/Reduce作业的输出样式。

Map/Reduce框架根据作业的OutputFormat来：

检验作业的输出，例如检查输出路径是否已经存在。
提供一个RecordWriter的实现，用来输出作业结果。 输出文件保存在FileSystem上。
TextOutputFormat是默认的 OutputFormat。

任务的Side-Effect File
在一些应用程序中，子任务需要产生一些side-file，这些文件与作业实际输出结果的文件不同。

在这种情况下，同一个Mapper或者Reducer的两个实例（比如预防性任务）同时打开或者写 FileSystem上的同一文件就会产生冲突。因此应用程序在写文件的时候需要为每次任务尝试（不仅仅是每次任务，每个任务可以尝试执行很多次）选取一个独一无二的文件名(使用attemptid，例如task_200709221812_0001_m_000000_0)。

为了避免冲突，Map/Reduce框架为每次尝试执行任务都建立和维护一个特殊的 ${mapred.output.dir}/_temporary/_${taskid}子目录，这个目录位于本次尝试执行任务输出结果所在的FileSystem上，可以通过 ${mapred.work.output.dir}来访问这个子目录。 对于成功完成的任务尝试，只有${mapred.output.dir}/_temporary/_${taskid}下的文件会移动到${mapred.output.dir}。当然，框架会丢弃那些失败的任务尝试的子目录。这种处理过程对于应用程序来说是完全透明的。

在任务执行期间，应用程序在写文件时可以利用这个特性，比如 通过 FileOutputFormat.getWorkOutputPath()获得${mapred.work.output.dir}目录， 并在其下创建任意任务执行时所需的side-file，框架在任务尝试成功时会马上移动这些文件，因此不需要在程序内为每次任务尝试选取一个独一无二的名字。

注意：在每次任务尝试执行期间，${mapred.work.output.dir} 的值实际上是 ${mapred.output.dir}/_temporary/_{$taskid}，这个值是Map/Reduce框架创建的。 所以使用这个特性的方法是，在 FileOutputFormat.getWorkOutputPath() 路径下创建side-file即可。

对于只使用map不使用reduce的作业，这个结论也成立。这种情况下，map的输出结果直接生成到HDFS上。

RecordWriter
RecordWriter 生成<key, value> 对到输出文件。

RecordWriter的实现把作业的输出结果写到 FileSystem。

其他有用的特性
Counters
Counters 是多个由Map/Reduce框架或者应用程序定义的全局计数器。 每一个Counter可以是任何一种 Enum类型。同一特定Enum类型的Counter可以汇集到一个组，其类型为Counters.Group。

应用程序可以定义任意(Enum类型)的Counters并且可以通过 map 或者 reduce方法中的 Reporter.incrCounter(Enum, long)或者 Reporter.incrCounter(String, String, long) 更新。之后框架会汇总这些全局counters。

DistributedCache
DistributedCache 可将具体应用相关的、大尺寸的、只读的文件有效地分布放置。

DistributedCache 是Map/Reduce框架提供的功能，能够缓存应用程序所需的文件 （包括文本，档案文件，jar文件等）。

应用程序在JobConf中通过url(hdfs://)指定需要被缓存的文件。 DistributedCache假定由hdfs://格式url指定的文件已经在 FileSystem上了。

Map-Redcue框架在作业所有任务执行之前会把必要的文件拷贝到slave节点上。 它运行高效是因为每个作业的文件只拷贝一次并且为那些没有文档的slave节点缓存文档。

DistributedCache 根据缓存文档修改的时间戳进行追踪。 在作业执行期间，当前应用程序或者外部程序不能修改缓存文件。

distributedCache可以分发简单的只读数据或文本文件，也可以分发复杂类型的文件例如归档文件和jar文件。归档文件(zip,tar,tgz和tar.gz文件)在slave节点上会被解档（un-archived）。 这些文件可以设置执行权限。

用户可以通过设置mapred.cache.{files|archives}来分发文件。 如果要分发多个文件，可以使用逗号分隔文件所在路径。也可以利用API来设置该属性： DistributedCache.addCacheFile(URI,conf)/ DistributedCache.addCacheArchive(URI,conf) and DistributedCache.setCacheFiles(URIs,conf)/ DistributedCache.setCacheArchives(URIs,conf) 其中URI的形式是 hdfs://host:port/absolute-path#link-name 在Streaming程序中，可以通过命令行选项 -cacheFile/-cacheArchive 分发文件。

用户可以通过 DistributedCache.createSymlink(Configuration)方法让DistributedCache 在当前工作目录下创建到缓存文件的符号链接。 或者通过设置配置文件属性mapred.create.symlink为yes。 分布式缓存会截取URI的片段作为链接的名字。 例如，URI是 hdfs://namenode:port/lib.so.1#lib.so， 则在task当前工作目录会有名为lib.so的链接， 它会链接分布式缓存中的lib.so.1。

DistributedCache可在map/reduce任务中作为 一种基础软件分发机制使用。它可以被用于分发jar包和本地库（native libraries）。 DistributedCache.addArchiveToClassPath(Path, Configuration)和 DistributedCache.addFileToClassPath(Path, Configuration) API能够被用于 缓存文件和jar包，并把它们加入子jvm的classpath。也可以通过设置配置文档里的属性 mapred.job.classpath.{files|archives}达到相同的效果。缓存文件可用于分发和装载本地库。

Tool
Tool 接口支持处理常用的Hadoop命令行选项。

Tool 是Map/Reduce工具或应用的标准。应用程序应只处理其定制参数， 要把标准命令行选项通过 ToolRunner.run(Tool, String[]) 委托给 GenericOptionsParser处理。

Hadoop命令行的常用选项有：
-conf <configuration file>
-D <property=value>
-fs <local|namenode:port>
-jt <local|jobtracker:port>

IsolationRunner
IsolationRunner 是帮助调试Map/Reduce程序的工具。

使用IsolationRunner的方法是，首先设置 keep.failed.tasks.files属性为true （同时参考keep.tasks.files.pattern）。

然后，登录到任务运行失败的节点上，进入 TaskTracker的本地路径运行 IsolationRunner：
$ cd <local path>/taskTracker/${taskid}/work
$ bin/hadoop org.apache.hadoop.mapred.IsolationRunner ../job.xml

IsolationRunner会把失败的任务放在单独的一个能够调试的jvm上运行，并且采用和之前完全一样的输入数据。

Profiling
Profiling是一个工具，它使用内置的java profiler工具进行分析获得(2-3个)map或reduce样例运行分析报告。

用户可以通过设置属性mapred.task.profile指定系统是否采集profiler信息。 利用api JobConf.setProfileEnabled(boolean)可以修改属性值。如果设为true， 则开启profiling功能。profiler信息保存在用户日志目录下。缺省情况，profiling功能是关闭的。

如果用户设定使用profiling功能，可以使用配置文档里的属性 mapred.task.profile.{maps|reduces} 设置要profile map/reduce task的范围。设置该属性值的api是 JobConf.setProfileTaskRange(boolean,String)。 范围的缺省值是0-2。

用户可以通过设定配置文档里的属性mapred.task.profile.params 来指定profiler配置参数。修改属性要使用api JobConf.setProfileParams(String)。当运行task时，如果字符串包含%s。 它会被替换成profileing的输出文件名。这些参数会在命令行里传递到子JVM中。缺省的profiling 参数是 -agentlib:hprof=cpu=samples,heap=sites,force=n,thread=y,verbose=n,file=%s。

调试
Map/Reduce框架能够运行用户提供的用于调试的脚本程序。 当map/reduce任务失败时，用户可以通过运行脚本在任务日志（例如任务的标准输出、标准错误、系统日志以及作业配置文件）上做后续处理工作。用户提供的调试脚本程序的标准输出和标准错误会输出为诊断文件。如果需要的话这些输出结果也可以打印在用户界面上。

在接下来的章节，我们讨论如何与作业一起提交调试脚本。为了提交调试脚本， 首先要把这个脚本分发出去，而且还要在配置文件里设置。

如何分发脚本文件：
用户要用 DistributedCache 机制来分发和链接脚本文件

如何提交脚本：
一个快速提交调试脚本的方法是分别为需要调试的map任务和reduce任务设置 "mapred.map.task.debug.script" 和 "mapred.reduce.task.debug.script" 属性的值。这些属性也可以通过 JobConf.setMapDebugScript(String) 和 JobConf.setReduceDebugScript(String) API来设置。对于streaming， 可以分别为需要调试的map任务和reduce任务使用命令行选项-mapdebug 和 -reducedegug来提交调试脚本。

脚本的参数是任务的标准输出、标准错误、系统日志以及作业配置文件。在运行map/reduce失败的节点上运行调试命令是：
$script $stdout $stderr $syslog $jobconf

Pipes 程序根据第五个参数获得c++程序名。 因此调试pipes程序的命令是
$script $stdout $stderr $syslog $jobconf $program

默认行为
对于pipes，默认的脚本会用gdb处理core dump， 打印 stack trace并且给出正在运行线程的信息。

JobControl
JobControl是一个工具，它封装了一组Map/Reduce作业以及他们之间的依赖关系。

数据压缩
Hadoop Map/Reduce框架为应用程序的写入文件操作提供压缩工具，这些工具可以为map输出的中间数据和作业最终输出数据（例如reduce的输出）提供支持。它还附带了一些 CompressionCodec的实现，比如实现了 zlib和lzo压缩算法。 Hadoop同样支持gzip文件格式。

考虑到性能问题（zlib）以及Java类库的缺失（lzo）等因素，Hadoop也为上述压缩解压算法提供本地库的实现。更多的细节请参考 这里。

中间输出
应用程序可以通过 JobConf.setCompressMapOutput(boolean)api控制map输出的中间结果，并且可以通过 JobConf.setMapOutputCompressorClass(Class)api指定 CompressionCodec。

作业输出
应用程序可以通过 FileOutputFormat.setCompressOutput(JobConf, boolean) api控制输出是否需要压缩并且可以使用 FileOutputFormat.setOutputCompressorClass(JobConf, Class)api指定CompressionCodec。

如果作业输出要保存成 SequenceFileOutputFormat格式，需要使用 SequenceFileOutputFormat.setOutputCompressionType(JobConf, SequenceFile.CompressionType)api，来设定 SequenceFile.CompressionType (i.e.RECORD / BLOCK - 默认是RECORD)。

例子：WordCount v2.0
这里是一个更全面的WordCount例子，它使用了我们已经讨论过的很多Map/Reduce框架提供的功能。

运行这个例子需要HDFS的某些功能，特别是 DistributedCache相关功能。因此这个例子只能运行在 伪分布式 或者 完全分布式模式的 Hadoop上。

源代码
  WordCount.java
1. package org.myorg;
2.  
3. import java.io.*;
4. import java.util.*;
5.  
6. import org.apache.hadoop.fs.Path;
7. import org.apache.hadoop.filecache.DistributedCache;
8. import org.apache.hadoop.conf.*;
9. import org.apache.hadoop.io.*;
10. import org.apache.hadoop.mapred.*;
11. import org.apache.hadoop.util.*;
12.  
13. public class WordCount extends Configured implements Tool {
14.  
15.    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
16.  
17.      static enum Counters { INPUT_WORDS }
18.  
19.      private final static IntWritable one = new IntWritable(1);
20.      private Text word = new Text();
21.  
22.      private boolean caseSensitive = true;
23.      private Set<String> patternsToSkip = new HashSet<String>();
24.  
25.      private long numRecords = 0;
26.      private String inputFile;
27.  
28.      public void configure(JobConf job) {
29.        caseSensitive = job.getBoolean("wordcount.case.sensitive", true);
30.        inputFile = job.get("map.input.file");
31.  
32.        if (job.getBoolean("wordcount.skip.patterns", false)) {
33.          Path[] patternsFiles = new Path[0];
34.          try {
35.            patternsFiles = DistributedCache.getLocalCacheFiles(job);
36.          } catch (IOException ioe) {
37.            System.err.println("Caught exception while getting cached files: " + StringUtils.stringifyException(ioe));
38.          }
39.          for (Path patternsFile : patternsFiles) {
40.            parseSkipFile(patternsFile);
41.          }
42.        }
43.      }
44.  
45.      private void parseSkipFile(Path patternsFile) {
46.        try {
47.          BufferedReader fis = new BufferedReader(new FileReader(patternsFile.toString()));
48.          String pattern = null;
49.          while ((pattern = fis.readLine()) != null) {
50.            patternsToSkip.add(pattern);
51.          }
52.        } catch (IOException ioe) {
53.          System.err.println("Caught exception while parsing the cached file '" + patternsFile + "' : " + StringUtils.stringifyException(ioe));
54.        }
55.      }
56.  
57.      public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
58.        String line = (caseSensitive) ? value.toString() : value.toString().toLowerCase();
59.  
60.        for (String pattern : patternsToSkip) {
61.          line = line.replaceAll(pattern, "");
62.        }
63.  
64.        StringTokenizer tokenizer = new StringTokenizer(line);
65.        while (tokenizer.hasMoreTokens()) {
66.          word.set(tokenizer.nextToken());
67.          output.collect(word, one);
68.          reporter.incrCounter(Counters.INPUT_WORDS, 1);
69.        }
70.  
71.        if ((++numRecords % 100) == 0) {
72.          reporter.setStatus("Finished processing " + numRecords + " records " + "from the input file: " + inputFile);
73.        }
74.      }
75.    }
76.  
77.    public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
78.      public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
79.        int sum = 0;
80.        while (values.hasNext()) {
81.          sum += values.next().get();
82.        }
83.        output.collect(key, new IntWritable(sum));
84.      }
85.    }
86.  
87.    public int run(String[] args) throws Exception {
88.      JobConf conf = new JobConf(getConf(), WordCount.class);
89.      conf.setJobName("wordcount");
90.  
91.      conf.setOutputKeyClass(Text.class);
92.      conf.setOutputValueClass(IntWritable.class);
93.  
94.      conf.setMapperClass(Map.class);
95.      conf.setCombinerClass(Reduce.class);
96.      conf.setReducerClass(Reduce.class);
97.  
98.      conf.setInputFormat(TextInputFormat.class);
99.      conf.setOutputFormat(TextOutputFormat.class);
100.  
101.      List<String> other_args = new ArrayList<String>();
102.      for (int i=0; i < args.length; ++i) {
103.        if ("-skip".equals(args[i])) {
104.          DistributedCache.addCacheFile(new Path(args[++i]).toUri(), conf);
105.          conf.setBoolean("wordcount.skip.patterns", true);
106.        } else {
107.          other_args.add(args[i]);
108.        }
109.      }
110.  
111.      FileInputFormat.setInputPaths(conf, new Path(other_args.get(0)));
112.      FileOutputFormat.setOutputPath(conf, new Path(other_args.get(1)));
113.  
114.      JobClient.runJob(conf);
115.      return 0;
116.    }
117.  
118.    public static void main(String[] args) throws Exception {
119.      int res = ToolRunner.run(new Configuration(), new WordCount(), args);
120.      System.exit(res);
121.    }
122. }
123.  

运行样例
输入样例：

$ bin/hadoop dfs -ls /usr/joe/wordcount/input/
/usr/joe/wordcount/input/file01
/usr/joe/wordcount/input/file02

$ bin/hadoop dfs -cat /usr/joe/wordcount/input/file01
Hello World, Bye World!

$ bin/hadoop dfs -cat /usr/joe/wordcount/input/file02
Hello Hadoop, Goodbye to hadoop.

运行程序：

$ bin/hadoop jar /usr/joe/wordcount.jar org.myorg.WordCount /usr/joe/wordcount/input /usr/joe/wordcount/output

输出：

$ bin/hadoop dfs -cat /usr/joe/wordcount/output/part-00000
Bye 1
Goodbye 1
Hadoop, 1
Hello 2
World! 1
World, 1
hadoop. 1
to 1

注意此时的输入与第一个版本的不同，输出的结果也有不同。

现在通过DistributedCache插入一个模式文件，文件中保存了要被忽略的单词模式。

$ hadoop dfs -cat /user/joe/wordcount/patterns.txt
\.
\,
\!
to

再运行一次，这次使用更多的选项：

$ bin/hadoop jar /usr/joe/wordcount.jar org.myorg.WordCount -Dwordcount.case.sensitive=true /usr/joe/wordcount/input /usr/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

应该得到这样的输出：

$ bin/hadoop dfs -cat /usr/joe/wordcount/output/part-00000
Bye 1
Goodbye 1
Hadoop 1
Hello 2
World 2
hadoop 1

再运行一次，这一次关闭大小写敏感性（case-sensitivity）：

$ bin/hadoop jar /usr/joe/wordcount.jar org.myorg.WordCount -Dwordcount.case.sensitive=false /usr/joe/wordcount/input /usr/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

输出：

$ bin/hadoop dfs -cat /usr/joe/wordcount/output/part-00000
bye 1
goodbye 1
hadoop 2
hello 2
world 2

程序要点
通过使用一些Map/Reduce框架提供的功能，WordCount的第二个版本在原始版本基础上有了如下的改进：

展示了应用程序如何在Mapper (和Reducer)中通过configure方法 修改配置参数(28-43行)。
展示了作业如何使用DistributedCache 来分发只读数据。 这里允许用户指定单词的模式，在计数时忽略那些符合模式的单词(104行)。
展示Tool接口和GenericOptionsParser处理Hadoop命令行选项的功能 (87-116, 119行)。
展示了应用程序如何使用Counters(68行)，如何通过传递给map（和reduce） 方法的Reporter实例来设置应用程序的状态信息(72行)。
Java和JNI是Sun Microsystems, Inc.在美国和其它国家的注册商标。

本文来自CSDN博客，转载请标明出处：http://blog.csdn.net/superxgl/archive/2010/01/11/5171929.aspx