大数据专栏-USB迷|专注于互联网分享

大数据专栏

大数据概论

大数据的概念

大数据特点

大数据应用场景

大数据发展前景

业务流程分析

Hadoop入门

Hadoop简介

Hadoop核心组件：

Hadoop生态圈

Hadoop产生背景

Hadoop创始人和logo

Hadoop发展历程

Hadoop特性

Hadoop优点

分布式系统概述

分布式集群

负载均衡概念

什么是Nginx？

什么是反向代理？

最常见负载均衡策略：

负载均衡和分布式的区别

Linux系统环境准备

Hadoop集群部署方式：

任务一：安装JDK

2. 解压压缩包

3. 配置环境变量（1）

3. 配置环境变量（2）

任务二：配置SSH免密登录

01 SSH概念

02 SSH组成

03 SSH实现过程

配置SSH免密登录

（1）下载SSH服务并启动

（2）首先生成密钥对，使用命令：

HDFS伪分布式集群搭建

HDFS集群主要配置文件

Hadoop默认提供了两种配置文件

HDFS集群测试

01 格式化文件系统

YARN伪分布式集群搭建

YARN集群主要配置文件

YARN集群测试

Hadoop集群初体验

启动Hadoop集群

查看进程启动情况

大数据概论

大数据的概念

何谓“大数据”（Big Data），如果从字面意思看来，“大数据”指的是巨量数据。

1TB=1024GB=2^10GB

PB、EB、ZB、YB、BB之间的换算单位都是1024

麦肯锡的定义

“大数据”是指在一定时间内无法用传统数据库软件工具采集、存储、管理和分析其内容的数据集合。

研究机构Gartner

“大数据”是需要新处理模式才能具有更强的决策力、洞察发现力和流程优化能力的海量、高增长率和多样化的信息资产。

技术角度

通过大数据处理、智能分析让数据带来高价值。

大数据特点

（1）规模性（Volume）

（2）多样性

大数据可以分为三类：

一是结构化数据
二是非结构化数据
三是半结构化数据

（3）高速性

（4）价值性

（5）真实性

大数据应用场景

01 电商大数据——精准营销法宝

02 金融大数据——财源滚滚来

03 医疗大数据——看病更高效

04 零售大数据——最懂消费者

05 交通大数据——畅通出行

06 舆情监控大数据——名侦探柯南

大数据发展前景

国家政策
国际方面
高校方面

业务流程分析

Hadoop入门

Hadoop简介

Hadoop是Apache软件基金会旗下的一个开源的分布式计算平台。

Hadoop 提供的功能：利用服务器集群，根据用户的自定义业务逻辑，对海量数据进行分布式处理（海量数据的存储和海量数据的分析计算问题。）

Hadoop核心组件：

Common（基础组件）:Hadoop Common是Hadoop体系最底层的一个模块，为Hadoop各个子模块提供各种工具，比如系统配置工具Configuration、远程调用RPC、序列化机制和日志操作等等，是其他模块的基础。
HDFS（Hadoop Distributed File System 分布式文件系统） ：HDFS是以分布式进行存储的文件系统，主要负责集群数据的存储与读取
MapReduce（Map 和 Reduce 分布式运算编程框架） ：MapReduce是一种计算模型，用于大规模数据集（大于1TB）的并行计算。
“Map”对数据集上的独立元素进行指定的操作，生成键值对（例如：手机通讯录中，键：小明，值：13333333333（小明号码），这就是所谓键值对，不要想太复杂了）形式中间结果；
“Reduce”则对之间结果中相同“键”的所有“值”进行规约，以得到最终结果。
YARN（Yet Another Resources N 运算资源调度系统）：Hadoop2.X中的资源管理器，它可以为上层应用提供统一的资源管理和调度

Hadoop生态圈

除了核心的HDFS和MapReduce以外，Hadoop生态圈还包括Hive、ZooKeeper、HBase、Sqoop、Flume等功能组件。

HDFS是Hadoop分布式文件系统缩写，它是Hadoop的基石。HDFS是一个具备高度容错性的文件系统，适合部署在廉价的机器上，它能提供高吞吐量的数据访问，非常适合大规模数据集上的应用。
MapReduce是一种编程模型，利用函数式编程思想，将对数据集的过程分为Map和Reduce两个阶段。MapReduce的这种编程模型非常适合进行分布式计算。Hadoop提供MapReduce的计算框架，实现了这种编程模型，用户可以通过Java\C++\Python\PHP等多种语言进行编程。
Pig与Hive类似，也是对大数据集进行分析和评估的工具，不同于Hive的是Pig提供了一种高层的，面向领域的抽象语言Pig Latin.同样Pig也可以将Pig Latin转化为MapReduce作业。相比与SQL，Pig Latin更加灵活，但学习成本更高。
Zookeeper（也称为动物饲养员）作为一个分布式服务框架，是基于Fast Paxos算法实现，解决分布式系统中一致性的问题。提供了配置维护，名字服务，分布式同步，组服务等。

HBase（分布式列存数据库）来源于Google的Bigtable论文，HBase是一个分布式的，面向列簇的开源数据库。采用了Bigtable的数据模型--列簇。HBase擅长大规模数据的随机、实时读写访问。

Sqoop（数据ETL/同步工具）是SQL-to-Hadoop的缩写，主要用于传统数据和Hadoop之前传输数据。数据的导入和导出本质上是MapReduce程序，充分利用了MR的并行化和容错性，Sqoop利用数据库技术描述数据架构，用于关系数据库（MySQL）、数据仓库和Hadoop之间转移数据。

Hadoop产生背景

Hadoop创始人和logo

Hadoop是由Apache Lucence创始人Doug Cutting创建的，Lucence是一个应用广泛的文本搜索系统库。

Hadoop发展历程

2002~2004 年，第一轮互联网泡沫刚刚破灭，很多互联网从业人员都失业了。
我们的“主角" Doug Cutting 也不例外，他只能写点技术文章赚点稿费来养家糊口。但是 Doug Cutting 不甘寂寞，怀着对梦想和未来的渴望，与他的好朋友 Mike Cafarella 一起开发出一个开源的搜索引擎 Nutch，并历时一年把这个系统做到能支持亿级网页的搜索。
但是当时的网页数量远远不止这个规模，所以两人不断改进，想让支持的网页量再多一个数量级。）

Hadoop特性

Hadoop优点

（1）高可靠性：数据存储多个备份；Hadoop会自动重新部署数据处理请求失败的计算任务。

（2）高扩展性：Hadoop集群可以很容易进行节点的扩展，扩大集群。

（3）高效性：Hadoop节点间移动数据，保证各个节点的动态平衡，因此处理速度非常快。

（4）高容错性：数据存储多个备份；Hadoop会重新运行失败的任务。

（5）低成本：Hadoop 开源。

（6）可构建在廉价的机器上：大部分普通商用服务器就可以满足要求。

（7）Hadoop基本框架用Java语言编写：Hadoop含有使用Java语言编写的框架。FEN

分布式系统概述

分布式集群

集群和分布式的区别？

（1）从解决问题的角度看：分布式是以缩短单个任务的执行时间来提升效率的；集群则是通过提高单位时间内执行的任务数来提升效率。

（2）从软件部署的角度看：分布式是指将不同的业务分布在不同的地方；集群则是将几台服务器集中在一起，实现同一业务。

总结：一个分布式系统，是通过多个节点（厨师和配菜师）组成的，各节点（厨师与厨师之间，配菜师与配菜师之间）都是集群化，并且各集群（厨师群，配菜师群）还是分布式的。

负载均衡概念

负载均衡是指将请求分摊到多个操作单元也就是分开部署的服务器上，Nginx是常用的反向代理服务器，可以用来做负载均衡。

什么是Nginx？

Nginx 是俄罗斯人编写的十分轻量级的 HTTP 服务器，它是一个高性能的HTTP和反向代理服务器。优点：性能高、轻量级、易操作

什么是反向代理？

反向代理（ReverseProxy）是指以代理服务器来接受internet上的连接请求，然后将请求转发给内部网络上的服务器，并将从服务器上得到的结果返回给internet上请求连接的客户端。即真实的服务器不能直接被外部网络访问,想要访问必须通过代理。

反向代理的作用：
（1）防止主服务器被恶意攻击
（2）为负载均衡提供实现支持

最常见负载均衡策略：

（1）轮询：平均分配，人人都有、一人一次。

（2）加权轮询：根据服务器的不同处理能力，给每个服务器分配不同的权值，使其能够接受相应权值数的服务请求。

（3）最快响应：这也是一种动态负载均衡策略，它的本质是根据每个节点对过去一段时间内的响应情况来分配，响应越快分配的越多。

（4）Hash法

哈希算法主要将对请求的IP地址或者URL计算一个哈希值，然后与集群节点的数量进行取模来决定将请求分发给哪个集群节点。

负载均衡和分布式的区别

负载均衡：请求是不可拆分的，是独立的一个请求，被服务器当中的任何一台接收并处理即可。

（例如：总共做2道菜，一个酸辣土豆丝（一个独立的请求），一个鱼香肉丝，均匀地分配给2个厨师）

分布式：任何一个任务（请求）需要节点相互协作共同完成，是可以拆分的。

（例如：一个酸辣土豆丝（一个独立的请求），让厨师和配菜师配合共同完成）

Linux系统环境准备

Hadoop集群部署方式：

（1）单机模式：又称为独立模式，在该模式下，无须运行任何守护进程，所有的程序都在单个JVM上执行。

（2）伪分布式模式：Hadoop程序的守护进程运行在一台主机节点上，通常使用伪分布式模式来调试Hadoop分布式程序的代码，以及程序执行是否正确，伪分布式模式是完全分布式模式的一个特例。

（3）完全分布式模式：Hadoop的守护进程分别运行在由多个主机搭建的集群上，不同节点担任不同的角色，在实际工作应用开发中，通常使用该模式构建企业级Hadoop系统。

任务一：安装JDK

2. 解压压缩包

tar命令：用于打包并压缩和解包并解压缩文件

使用格式：

○ 打包并压缩：tar -zcvf 打包压缩名文件名/目录

○ 解包并解压缩：tar -zxvf *.tar.gz格式的打包压缩文件

常用选项：

○ z：gzip，通过 gzip 格式压缩或者解压缩

○ -C：指定需要解压的目录，若是未指定，则解压到当前目录

3. 配置环境变量（1）

配置环境变量就是在整个运行环境都可以使用的变量，而路径添加到PATH类似于在Windows平台下将程序添加到注册表，添加某个路径到PATH环境变量后，执行该路径下的文件就不需要输入完整的命令路径而只需要输入命令的文件名。

3. 配置环境变量（2）

Linux环境变量和Windows的环境变量一样，分系统环境变量和用户环境变量，系统环境变量对所有用户有效，而用户环境变量只对当前用户有效。

系统环境变量：对于添加给所有用户的环境变量，直接编辑 "/etc/bashrc"或者"/etc/profile"；
用户环境变量：对于添加给某一个用户的环境变量，可以编辑用户/home目录下的 " 用户名/.bashrc "或者"用户名/.bash_profile"。

（1）在此处我们配置系统环境变量，使用命令：

（2）在最后加入以下两行内容：

注意：

export 是把这两个变量导出为全局变量。
大小写必须严格区分。

效果图如下所示

添加完成，使用:wq保存退出

（3）让配置文件立即生效，使用如下命令：

（4）检测JDK是否安装成功，使用命令查看JDK版本：

执行此命令后，若是出现JDK版本信息说明配置成功：

任务二：配置SSH免密登录

01 SSH概念

SSH 为 Secure Shell（安全外壳协议）的缩写。

SSH 是一种网络协议，用于计算机之间的加密登录。很多 ftp、 pop 和 telnet 在本质上都是不安全的，因为它们在网络上用明文传送口令和数据，别有用心的人非常容易就可以截获这些口令和数据。

SSH就是专为远程登录会话和其他网络服务提供安全性的协议。

02 SSH组成

SSH 是由客户端和服务端的软件组成的。

服务端是一个守护进程(sshd)，他在后台运行并响应来自客户端的连接请求。
客户端包含 ssh 程序以及像 scp（远程拷贝）、 slogin（远程登陆）、 sftp（安全文件传输）等其他的应用程序。

03 SSH实现过程

（1）首先需要在客户机上生成root用户的密钥对文件（公钥和私钥）；
（2）然后在客户机上将生成的公钥文件复制到公钥库文件authorized_keys中，然后将公钥库文件authorized_keys放置到/root/.ssh目录下；
（3）将客户机的公钥库文件authorized_keys传送到服务端的同级目录下，即/root/.ssh目录下；
（4）这样以后每次在客户机上使用root登录服务端的时候，就不需要输入密码了。

配置SSH免密登录

（1）下载SSH服务并启动

SSH服务（openssh-server和openssh-clients）已经为大家下载好，所以此处直接启动即可：

SSH服务启动后，默认开启22（SSH的默认端口）端口号，使用以下命令进行查看：

执行命令，可以看到22号端口已经开启，证明我们SSH服务启动成功：

只要将SSH服务启动成功，我们就可以进行远程连接访问了。

开发人员比较常用的远程连接工具有Xshell、SecureCRT等。

（2）首先生成密钥对，使用命令：

上面一种是简写形式，提示要输入信息时不需要输入任何东西，直接回车三次即可

语法解析：

ssh-keygen ：生成、管理和转换认证密钥。
-t：指定密钥类型，包括RSA和DSA两种密钥，默认RSA。

验证密钥对

从打印信息中可以看出，私钥id_rsa和公钥id_rsa.pub都已创建成功，并放在/root/.ssh目录中：

（3）将公钥放置到授权列表文件 authorized_keys 中，使用命令：

验证

（4）修改授权列表文件 authorized_keys 的权限，使用命令：

设置拥有者可读可写，其他人无任何权限（不可读、不可写、不可执行）。

（5）验证免密登录是否配置成功，使用如下命令：

查看本机主机名

查看本机IP地址：

（6）远程登录成功后，若想退出，可以使用exit命令：

HDFS伪分布式集群搭建

Hadoop是Apache基金会面向全球开源的产品之一，任何用户都可以从Apache Hadoop官网/dist/hadoop/common/下载使用。

hadoop2.7.7的安装包，存放在/root/software目录下，使用如下命令进行解压即可使用：

将其解压到当前目录下，即/root/software中，

接下来，进入Hadoop的安装目录，通过 ll 命令查看Hadoop目录结构，如下图所示：

（1）bin：存放操作Hadoop相关服务（HDFS、YARN）的脚本，但是通常使用sbin目录下的脚本。

（2）etc：存放Hadoop配置文件。

（3）include：对外提供的编程库头文件（具体动态库和静态库在lib目录中）。

（4）lib：该目录包含了Hadoop对外提供的编程动态库和静态库。

（5）libexec：各个服务对应的shell配置文件所在的目录。

（6）sbin：该目录存放Hadoop管理脚本，主要包含HDFS和YARN中各类服务的启动/关闭脚本。

（7）share：Hadoop各个模块编译后的jar包所在的目录。

HDFS集群主要配置文件

Hadoop默认提供了两种配置文件

（1）一种是只读的默认配置文件，包括core-default.xml、hdfs-default.xml、mapred-default.xml和yarn-default.xml，这些文件包含了Hadoop系统各种默认配置参数；

（2）另一种是Hadoop集群自定义配置时编辑的配置文件，包括hadoop-env.sh、yarn-env.sh、core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml和slaves共7个文件，可以根据需要在这些文件中对默认配置文件中的参数进行修改，Hadoop会优先选择这些配置文件中的参数。

01 配置环境变量hadoop-env.sh

首先需要复制一下本机安装的JDK的实际位置（避免写错最好不要手写），可以使用下列方式打印JDK的安装目录：

复制完成，使用如下命令打开hadoop-env.sh文件

找到JAVA_HOME参数位置，修改为本机安装的JDK的实际位置：

02 配置核心组件core-site.xml

该文件是Hadoop的核心配置文件，其目的是配置HDFS地址、端口号，以及临时文件目录。

使用如下命令打开“core-site.xml”文件：

将下面的配置内容添加到 <configuration></configuration> 中间：

该文件主要用于配置 HDFS 相关的属性，例如复制因子（即数据块的副本数）、NameNode 和 DataNode 用于存储数据的目录等。在完全分布式模式下，默认数据块副本是 3 份。使用如下命令打开“hdfs-site.xml”文件：

将下面的配置内容添加到 <configuration></configuration> 中间：

04 配置slaves文件

该文件用于记录Hadoop集群所有从节点（HDFS的DataNode和YARN的NodeManager所在主机）的主机名，用来配合一键启动脚本启动集群从节点（并且还需要保证关联节点配置了SSH免密登录）。

打开该配置文件：

我们看到其默认内容为localhost，因为我们搭建的是伪分布式集群，就只有一台主机，所以从节点也需要放在此主机上，所以此配置文件无需修改。

HDFS集群测试

01 格式化文件系统

执行格式化指令后，必须出现有“successfully formatted”信息才表示格式化成功

特别注意：上述格式化指令只需要在HDFS集群初次启动前执行即可。

02 启动和关闭HDFS集群

1. 单节点逐个启动和关闭

（1）在本机上使用以下指令启动NameNode进程

启动完成之后，使用 jps 指令查看NameNode进程的启动情况，效果如下图所示：

jps命令：显示系统当前运行的Java程序及其进程号。
其中424和350是进程的PID，也就是进程号。

（2）在本机上使用以下指令启动DataNode进程：

（3）再本机上使用以下指令启动SecondaryNameNode进程：

另外，当需要停止相关服务进程时，只需要将上述指令中的start更改为stop即可。

2. 脚本一键启动和关闭

启动集群最常使用的方式是使用脚本一键启动，前提是需要配置slaves配置文件和SSH免密登录。

一键启动HDFS集群

若想一键关闭HDFS集群，只需要将start改为stop即可，即stop-dfs.sh。

03 查看进程启动情况

在本机上执行 jps 命令，在打印结果中会看到 4 个进程，分别是 NameNode、SecondaryNameNode、Jps、和DataNode，如果出现了这 4 个进程表示进程启动成功。

04 通过UI查看HDFS运行状态

通过本机的浏览器访问http://localhost:50070或http://本机IP地址:50070查看HDFS集群状态，效果如下图所示：

YARN伪分布式集群搭建

YARN集群主要配置文件

下表是Hadoop集群搭建涉及的主要配置文件以及相关功能的描述。

01 配置环境变量yarn-env.sh

打开“yarn-env.sh”文件

找到JAVA_HOME参数位置，将前面的#去掉，将其值修改为本机安装的JDK的实际位置：

02 配置计算框架mapred-site.xml

该文件是MapReduce的核心配置文件，用于指定MapReduce运行时框架。此处应该指定 yarn，另外的可用值还有 local （本地的作业运行器）和 classic（MR1运行模式），默认为 local。在$HADOOP_HOME/etc/hadoop/目录中默认没有该文件，需要先通过如下命令将文件复制并重命名为“mapred-site.xml”：

接着，打开“mapred-site.xml”文件进行修改

将下面的配置内容添加到 <configuration></configuration> 中间：

03 配置YARN系统yarn-site.xml

本文件是YARN框架的核心配置文件，用于配置 YARN 进程及 YARN 的相关属性。

需要设置 ResourceManager 守护进程所在主机和NodeManager 上运行的辅助服务。

将下面的配置内容加入 <configuration></configuration> 中间

YARN集群测试

01 启动和关闭YARN集群

在启动YARN集群之前，我们首先使用脚本一键启动的方式启动HDFS集群。命令如下所示：

1. 单节点逐个启动和关闭

单节点逐个启动的方式，需要参照以下方式逐个启动YARN集群服务需要的相关服务进程，具体步骤如下：

（1）在本机上使用以下指令启动ResourceManager进程：

（2）在本机上使用以下指令启动NodeManager进程：

另外，当需要停止相关服务进程时，只需要将上述指令中的start更改为stop即可。

01 启动和关闭YARN集群

在本机上使用如下方式一键启动YARN集群：

02 查看进程启动情况

在本机上执行 jps 命令，在打印结果中多了2 个进程，分别是 ResourceManager和NodeManager，如果出现了这 2 个进程表示进程启动成功

03 通过UI查看YARN运行状态

通过本机的浏览器访问http://localhost:8088或http://本机IP地址:8088查看YARN集群状态，效果如下图所示：

Hadoop集群初体验

启动Hadoop集群

01 HDFS集群

在本机上使用如下方式一键启动HDFS集群：

启动顺序是：NameNode（主节点）——》DataNode（从节点）——》SecondaryNameNode

02 YARN集群

在本机上使用如下方式一键启动YARN集群：

启动顺序是：ResourceManager（主节点）——》NodeManager（从节点）

查看进程启动情况

在本机上执行 jps 命令，在打印结果中会看到 6 个进程，分别是 NodeManager、SecondaryNameNode 、ResourceManager、Jps、DataNode 和 NameNode，如果出现了这 6 个进程表示进程启动成功。如下图所示：

WordCount单词统计案例

（1）首先，通过本机的浏览器访问http://localhost:50070或http://本机IP地址:50070打开HDFS的Web UI界面。其次，选择Utilities——》Browse the file system查看文件系统里的数据文件，可以看到新建的HDFS上没有任何数据文件，如下图所示：

（2）在本机的/root目录下，新建一个名为data的文件夹，然后再执行“vim word.txt”指令新建一个word.txt文本文件，并编写一些单词内容，如下图所示：

（3）接着，在HDFS上创建/wordcount/input目录，并将word.txt文件上传至该目录下：

执行完上述指令后，再次查看HDFS的Web UI界面，会发现/wordcount/input目录创建成功并上传了指定的word.txt文件，如下图所示：

4）进入$HADOOP_HOME/share/hadoop/mapreduce/目录下，使用 ll 指令查看文件夹内容：

这里可以直接使用hadoop-mapreduce-examples-2.7.7.jar示例包，对HDFS上的word.txt文件进行单词统计，在jar包位置执行如下命令：

hadoop jar hadoop-mapreduce-examples-2.7.7.jar ：表示执行一个Hadoop的jar包程序；
wordcount：表示执行jar包程序中的单词统计功能；
/wordcount/input/word.txt：表示进行单词统计的HDFS文件路径；
/wordcount/output：表示进行单词统计后的输出HDFS结果路径。

执行完上述指令后，示例包中的MapReduce程序开始执行，效果图如下所示：

因为MapReduce程序分为Map端和Reduce端，当Map端和Reduce端都执行到100%，并显示job completed successfully时，才代表程序执行成功。

也可以通过YARN集群的Web UI界面查看运行状态，在本机的浏览器上访问http://localhost:8088或http://本机IP地址:8088。效果图如下所示：

（5）在“单词统计”示例程序执行成功后，再次刷新并查看HDFS的Web UI界面，效果如下图所示：

从上图可以看出，MapReduce程序执行成功后，在HDFS上自动创建了指定的输出目录/wordcount/output，并且输出了 _SUCCESS 和 part-r-00000 结果文件。

其中，_SUCCESS文件用于表示此次任务成功执行的标识，而part-r-00000表示单词统计的结果。

（6）接着，我们使用HDFS Shell的相关指令查看part-r-00000的内容，具体指令如下所示：

效果如下图所示：

Java API操作HDFS文件

"FileSystem.rename（Path arg0，Path arg1）"可对HDFS文件或目录进行重命名，其中arg0和arg1均为HDFS文件或目录的完整路径。具体实现如下：

  // 单元方法：重命名文件或者文件夹
    @Test
    public void renameFileOrDir() throws IllegalArgumentException, IOException {
        //重命名文件
        fs.rename(new Path("/123/README.txt"), new Path("/123/read.txt"));
        //重命名文件夹
        fs.rename(new Path("/123/1"), new Path("/123/data"));
    }

@Before和@After可以重复使用，所以不用重复书写，只写测试方法即可。

注意：测试方法上必须使用@Test，如果没有加程序会发生异常。

在测试方法中使用FileSystem.rename（Path arg0，Path arg1）对HDFS文件或目录进行重命名。其中，

arg0：为要重命名的HDFS中的文件或目录。
arg1：HDFS中的文件或目录的新名字。

而且这两个参数的类型都是Path类型，而不是String类型，所以都需要使用new Path(String str)将String类型路径转换为Path类型。

执行结果如下所示：

从执行结果中可以看出，我们成功将根目录下123目录中的/README.txt文件重命名为read.txt，将根目录下123目录下名为1的目录重命名为data。

注意：运行完程序后一定要注意刷新根目录，否则看不到执行结果。

查看文件/目录状态

通过"FileStatus.listStatus(Path f)"可查看指定HDFS中某个目录下所有文件或文件夹，具体实现如下：

      //单元方法：查看文件及文件夹信息
    @Test
    public void listStatus() throws FileNotFoundException, IllegalArgumentException, IOException{
        //使用listStatus()方法获得参数中指定目录下文件和文件夹的元数据信息（文件（夹）名称、路径、长度等），存放在一个数组中
        FileStatus[] listStatus = fs.listStatus(new Path("/123"));
        String flag="";
        for(FileStatus status:listStatus){
            if(status.isDirectory()){
                flag="Directory";
            }else {
                flag="File";
            }
            System.out.println(flag+":"+status.getPath().getName());
        } 
    }

@Before和@After可以重复使用，所以不用重复书写，只写测试方法即可。

注意：测试方法上必须使用@Test，如果没有加程序会发生异常。

在测试方法中使用FileSystem.listStatus(Path f)查看指定HDFS中某个目录下所有文件或文件夹。

该方法的返回值是一个FileStatus类型的数组，数组中存放的是该指定目录下所有文件和文件夹的元数据信息（文件（夹）名称、路径、长度等）。

要想得到该目录下所有文件和文件夹的名称，我们需要使用for循环遍历数组。

任务一：下载文件

格式：FileSystem.copyToLocalFile（Path src，Patch dst）
解析：将HDFS文件下载到本地的指定位置上，与HDFS Shell操作中get命令和copyToLocal命令相对应。

其中src和dst均为文件的完整路径，且都是Path类型，其中，

src：为起始路径，即要下载的文件所在的HDFS路径。
dst：为目标路径，即要下载到本地的目标路径。

另外需要注意的是，即使在dst路径下有下载的同名文件，程序也不会出错，它会覆盖之前的同名文件。

任务二：重命名文件/目录

格式：FileSystem.rename（Path arg0，Path arg1）
解析：对HDFS文件或目录进行重命名。

其中，arg0和arg1均为文件的完整路径，且都是Path类型，其中，

arg0：为要重命名的HDFS中的文件或目录。
arg1：HDFS中的文件或目录的新名字。

任务三：查看文件/目录状态

格式：FileStatus.listStatus(Path f)
解析：查看指定HDFS中某个目录下所有文件或文件夹，其中f为文件夹的完整路径（注意其路径的类型是Path类型，不是平时常用的String类型）。

该方法的返回值是一个FileStatus类型的数组，数组中存放的是该指定目录下所有文件和文件夹的元数据信息（文件（夹）名称、路径、长度等）。

要想得到该目录下所有文件和文件夹的名称，我们需要使用for循环遍历数组。

删除文件/目录

通过"FileSystem.delete（Path f，Boolean recursive）"可删除指定的HDFS文件或目录，其中f为需要删除文件或目录的完整路径，recursive用来确定是否进行递归删除，若是删除文件则为false,若是删除的是目录则为true。具体实现如下：

public class HDFSDemo {
	FileSystem fs = null;

	// 每次执行单元测试前都会执行该方法
	@Before
	public void setUp() throws IOException, InterruptedException, URISyntaxException {
		Configuration conf = new Configuration();
		// 不需要配置“fs.defaultFS”参数，直接传入URI和用户身份，最后一个参数是安装Hadoop集群的用户，我的是“root”
		fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "root");
	}

    //单元方法：删除文件或者文件夹
    @Test
    public void deleteFileOrDir() throws IllegalArgumentException, IOException{
        //删除文件，第二参数：是否递归，若是文件或者空文件夹时可以为false，若是非空文件夹则需要为true
        fs.delete(new Path("/123/read.txt"),false);
        //删除文件夹
        fs.delete(new Path("/123/data"), true);
    }

	// 每次执行单元测试后都会执行该方法，关闭资源
	@After
	public void tearDown() {
		if (null != fs) {
			try {
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
}

查看指定路径中文件和块信息

"FileSystem.listFiles（Path f，Boolean recursive）"可递归获取指定HDFS目录下的所有文件的绝对路径，其中f为指定路径，recursive用来确定是否进行递归列出。具体实现如下：

 //单元方法：查看目录信息，只显示该目录下的文件信息
    @Test
    public void listFiles() throws FileNotFoundException, IllegalArgumentException, IOException{
        //使用迭代器递归获取该目录下的所有文件
        RemoteIterator<LocatedFileStatus> listfile=fs.listFiles(new Path("/123"), true);
        while(listfile.hasNext()){
            LocatedFileStatus fileStatus=listfile.next();
            System.out.println("文件路径："+fileStatus.getPath());
            System.out.println("文件名称："+fileStatus.getPath().getName());
            System.out.println("块的大小："+fileStatus.getBlockSize());
            System.out.println("文件所有者："+fileStatus.getOwner());
            System.out.println("文件所属组："+fileStatus.getGroup());
            System.out.println("文件权限："+fileStatus.getPermission());
            System.out.println("副本个数："+fileStatus.getReplication());
            System.out.println("文件长度："+fileStatus.getLen());
            System.out.println("-----块的信息-----");
            BlockLocation[] blockLocations=fileStatus.getBlockLocations();
            for(BlockLocation bLocation:blockLocations){
                System.out.println("块的长度:"+bLocation.getLength()+"\t块起始偏移量:"+bLocation.getOffset());
                //块所在的DataNode节点
                String[] hosts = bLocation.getHosts();
                System.out.print("DataNode: ");
                for(String str:hosts){
                    System.out.print(str+"\t");
                }
                System.out.println();
            }
            System.out.println("------------------------");
        }
    }

任务一：删除文件/目录

格式：FileSystem.delete（Path f，Boolean recursive）
解析：删除指定的HDFS文件或目录，与HDFS Shell操作中rm命令相对应。

其中，

f：为需要删除文件或目录的完整路径；
recursivef：用来确定是否进行递归删除，若是删除文件或者空目录则为false，若是删除的是非空目录则为true。

注意参数f的类型是Path类型，而不是String类型，所以需要使用new Path(String str)将String类型路径转换为Path类型。

任务二：查看指定路径中文件状态和块信息

格式：FileSystem.listFiles（Path f，Boolean recursive）
解析：递归获取指定HDFS目录下的所有文件的绝对路径，其中f为指定路径，recursive用来确定是否进行递归列出。

该方法的返回值是一个迭代器，所以我们需要遍历迭代器中的元素，使用hasNext()判断是否有下一个元素， next()返回该元素。之后我们可以使用相应方法获取该元素的各种信息。

MapReduce简介

重温 Hadoop 四大组件：

HDFS：分布式文件系统
MapReduce：分布式运算编程框架
YARN： Hadoop 的资源调度系统
Common：以上三大组件的底层支撑组件，主要提供基础工具包和 RPC 框架等

什么是 MapReduce？

MapReduce 是一个分布式运算程序的编程框架，是用户开发“基于 Hadoop 的数据分析应用”的核心框架。

MapReduce 核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个 Hadoop 集群上。

什么是框架？
框架是一个半成品，已经对基础的代码进行了封装并提供相应的API，开发者在使用框架是直接调用封装好的API可以省去很多代码编写，从而提高工作效率和开发速度。

MapReduce 核心功能是将用户编写的业务逻辑代码和自带默认组件（也就是框架部分）整合成一个完整的分布式运算程序，并发运行在一个 Hadoop 集群上。

并发运行：让计算机同时运行几个程序或同时运行同一个程序多个进程或线程

为什么需要 MapReduce?

1. 海量数据在单机上处理因为硬件资源限制，无法胜任；

2. 而一旦将单机版程序扩展到集群来分布式运行，将极大增加程序的复杂度和开发难度；

3. 引入 MapReduce 框架后，开发人员可以将绝大部分工作集中在业务逻辑的开发上，而将分布式计算中的复杂性交由框架来处理。

MapReduce 程序运行演示

01 PI程序

进入 $HADOOP_HOME/share/hadoop/mapreduce/目录下，执行如下命令：

hadoop jar hadoop‐mapreduce‐examples‐2.7.7.jar pi 10 10

如下图所示：

解析：

hadoop jar hadoop-mapreduce-examples-2.7.7.jar：表示执行一个Hadoop的jar包程序；
pi：表示执行jar包程序中计算PI值的功能；
第1个10：表示运行10次map任务；
第2个10：表示每个map任务，投掷的次数

2个参数的乘积就是总的投掷次数。（类似扔飞镖）

PI值=4 * 投掷在圆里的次数/总投掷次数

从执行结果可以看出，MapReduce程序执行成功后，计算出了PI值为3.20，这与实际PI值有一定的偏差。

若想计算更为精确的PI值我们需要将map的个数和投掷次数增大。

02 wordcount 程序

将 $HADOOP_HOME/README.txt 文件上传到 HDFS 作为数据源：

hadoop fs -put README.txt /

运行结果：

准备数据源，可以自己创建数据源，这里我们直接使用Hadoop安装包中自带的README.txt文件作为数据源。

使用-put命令将文件上传到HDFS的根目录下，再使用ls命令查看是否上传成功。

执行 wordcount 程序：

hadoop jar hadoop-mapreduce-examples-2.7.7.jar wordcount /README.txt /wordcount

解析：

hadoop jar hadoop-mapreduce-examples-2.7.7.jar ：表示执行一个Hadoop的jar包程序；
wordcount：表示执行jar包程序中的单词统计功能；
/README.txt：表示进行单词统计的HDFS文件路径；
/wordcount：表示进行单词统计后的HDFS输出结果路径，不需要提前手动创建，程序执行过程中会自动创建该输出路径。

执行完成，使用 cat 命令查看运行结果：

Task two：MapReduce program running demonstration

MapReduce程序执行成功后，在HDFS上自动创建了指定的输出目录/wordcount，并且输出了 _SUCCESS 和 part-r-00000 结果文件。

其中，_SUCCESS文件用于表示此次任务成功执行的标识，而part-r-00000表示单词统计的结果。

接着，我们使用HDFS Shell的cat指令查看part-r-00000的内容，具体指令如下所示：

hadoop fs -cat /wordcount/part-r-00000

从运行结果可以看出，MapReduce示例程序成功统计出了/README.txt文本中的单词数量，并进行了结果输出。

源码解析

打开“/root/software/hadoop-2.7.7-src/hadoop-mapreduce-project/hadoop-mapreduce-examples”，如下图所示：

Task two：MapReduce program running demonstration

首先打开“ /root/software/hadoop-2.7.7-src/hadoop-mapreduce-project/hadoop-mapreduce-examples ”，如下图所示：

其中，pom.xml文件是该项目的maven管理配置文件，src里面包含我们所要查看的源代码。

查看PI和wordcount程序的源代码

使用vi编辑器打开“pom.xml”，找到第 127 行，它告诉了我们例子程序的主程序入口：

使用vi编辑器打开文件后，在命令模式下使用:set nu命令为该文本设置行号。

org.apache.hadoop.examples.ExampleDriver：其中org.apache.hadoop.examples为包名，ExampleDriver为类名。

查看PI和wordcount程序的源代码

进入该目录“ src/main/java/org/apache/hadoop/examples ”，如下图所示：

其中，org/apache/hadoop/examples为包的路径。

打开主入口程序“ExampleDriver.java”，告诉我们pi和wordcount对应的实际程序分别是WordCount.class和QuasiMonteCarlo.class：

所以我们在执行PI和wordcount时，使用的是hadoop jar xxx wordcount 和 hadoop jar xxx pi。

wordcount：表示执行 jar 包程序中的单词统计功能，运行的就是WordCount.class程序。
pi：表示执行jar包程序中计算PI值的功能，运行的就是QuasiMonteCarlo.class程序。

（1）首先分析主函数main()方法，系统运行时会加载Hadoop默认的一些配置。

Configuration conf = new Configuration();

（2）程序运行时会接收从命令行传来的一些参数。默认第一个参数为程序要处理的数据文件夹路径，即in文件夹路径。默认第二个参数为程序结果要输出到的文件夹路径，即out文件夹路径。如果命令行的参数少于2个时程序会停止运行。

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length < 2) {
      System.err.println("Usage: wordcount <in> [<in>...] <out>");
      System.exit(2);
    }

(3) 初始化一个Job任务，这个Job任务需要加载Hadoop的一些配置，并给这个Job 命名为“word count”。

Job job = Job.getInstance(conf, "word count");

（4）setJarByClass()方法使用了WordCount.class的类加载器来寻找包含该类的Jar包，然后设置该Jar包为作业所用的Jar包。

job.setJarByClass(WordCount.class);

（5）setMapperClass()方法设置了该Job任务所使用的Mapper类（拆分）。
setCombinerClass()方法设置了该Job任务所使用的Combiner类（中间结果合并）。
setReducerClass()方法设置了该Job任务所使用的Reducer类（合并）。

job.setMapperClass(TokenizerMapper.class);  
job.setCombinerClass(IntSumReducer.class);  
job.setReducerClass(IntSumReducer.class);

从上述代码可以发现，Combiner处理类和Reducer处理类使用的是同一个类即IntSumReducer类。为什么这两个风马牛不相及的处理类可以使用同一个类呢？这仅仅是巧合吗？答案是否定的。详细知识在后续讲述。

（6）设置Reducer的键输出的类型为Text类型，值输出的类型为IntWritable类型。例如本程序输出的单词和其出现的次数<单词,次数>。

Text类似于Java的String类型。
IntWritable类型类似于Java的int类型。

job.setOutputKeyClass(Text.class);  
 job.setOutputValueClass(IntWritable.class);

（7）设置程序输入和输出的路径。本示例是从命令行中接收参数。第一个参数为输入路径。第二个参数为输出路径。

FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));

（8）最后一行如果Job运行成功了，我们的程序就会正常退出。

System.exit(job.waitForCompletion(true) ? 0 : 1);

（9）Map类中map()方法分析。用户自定义Map类需要继承Mapper类，并实现map()方法。

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>

从上述代码可以发现，有有四种形式的参数分别用来指定Map的输入key值的类型、Map的输入value值的类型、Map的输出key值类型、Map的输出value值类型。

由于本示例中使用的是默认的TextInputFormat输入类型。所以Map输入键的类型为LongWritable的父类型Object，Map的输入值的类型为Text。

因为Map要输出的键值对类型为<单词,次数>，所以Map的输出key值类型为Text，输出value值的类型为IntWritable。

public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }

map()方法为用户想要实现的特定的功能。在本示例中，map()方法对输入的记录以空格为单位进行切分，然后使用Context对象进行输出。Context包含运行时的上下文信息。

（10）Reduce类中reduce方法分析。用户自定义Reduce需要继承Reducer类，并实现reduce()方法。

public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable>

此类也是一种规范类型，它同样有四种形式的参数用来指定Reduce的输入key值类型、输入value值类型、输出key类型、输出value类型。但，Reduce的输入key值的类型必须和Map输出key值类型相同。Reduce的输人value值的类型必须和Map输出value值类型相同。

public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }

当数据输入到Reduce端时，key为具体的单词，而values是对应单词的计数值所组成的列表，Map的输出就是Reduce的输入，所以reduce()方法只要遍历values并求和，即可得到某个单词的总次数。

通过查看 WordCount 程序 MapReduce 源码，得出以下几点结论：

该程序有一个 main() 方法，来启动任务的运行，其中 Job 对象存储了该程序运行的必要信息，比如指定 Mapper 类和 Reducer 类：

job.setMapperClass(TokenizerMapper.class);  # 继承 Mapper 类
job.setReducerClass(IntSumReducer.class); # 继承 Redcuer 类

总结：
MapReduce 程序的业务编码分为两个大部分，一部分配置程序的运行信息，一部分编写该 MapReduce 程序的业务逻辑，并且业务逻辑的 map 阶段和 reduce 阶段的代码分别继承 Mapper 类和 Reducer 类。

Map端规范

用户编写的程序分成三个部分： Mapper， Reducer， Driver
Mapper 的输入数据是 KV 对的形式（KV 的类型可自定义，示例中默认是<Object, Text>）
Mapper 的输出数据是 KV 对的形式（KV 的类型可自定义）
Mapper 中的业务逻辑写在 map() 方法中
MapTask 进程对每一个 <K,V> 调用一次map()方法

Reduce端规范

Reducer 的输入数据类型对应 Mapper 的输出数据类型，也是 KV（一定要注意，因为Map端的输出作为Reduce端的输入，所以数据类型一致）
Reducer 的输出数据是 KV 对的形式（KV 的类型可自定义）
Reducer 的业务逻辑写在 reduce() 方法中
ReduceTask 进程对每一组相同 k 的 <k,v> 组调用一次 reduce() 方法
用户自定义的 Mapper 和 Reducer 都要继承各自的父类（使用extends实现继承）

Driver端规范

整个程序需要一个 Drvier 来进行提交，提交的是一个描述了各种必要信息的 Job 对象

任务一：MapReduce概述

重温 Hadoop 四大组件：

HDFS：分布式文件系统
MapReduce：分布式运算编程框架
YARN： Hadoop 的资源调度系统
Common：以上三大组件的底层支撑组件，主要提供基础工具包和 RPC 框架等

MapReduce 是什么？

MapReduce是一个分布式运算程序的编程框架，是用户开发“基于 Hadoop 的数据分析应用”的核心框架。

MapReduce 核心功能是什么？

将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个 Hadoop 集群上。

任务二：MapReduce程序运行演示

Hadoop 的发布包中内置了一个 hadoop-mapreduce-examples-2.7.7.jar，这个 jar 包中有各种 MapReduce 示例程序，其中非常有名的就是 PI 程序和 wordcount。此 jar 包存放在$HADOOP_HOME/share/hadoop/mapreduce/ 目录里。
注意在运行MapReduce程序前，必须要成功开启Hadoop集群（HDFS集群和YARN集群）。

任务三：MapReduce示例编写规范

Map端规范

用户编写的程序分成三个部分： Mapper， Reducer， Driver
Mapper 的输入数据是 KV 对的形式（KV 的类型可自定义，示例中默认是<Object, Text>）
Mapper 的输出数据是 KV 对的形式（KV 的类型可自定义）
Mapper 中的业务逻辑写在 map() 方法中
MapTask 进程对每一个 <K,V> 调用一次map()方法

Reduce端规范

Reducer 的输入数据类型对应 Mapper 的输出数据类型，也是 KV（一定要注意，因为Map端的输出作为Reduce端的输入，所以数据类型一致）
Reducer 的输出数据是 KV 对的形式（KV 的类型可自定义）
Reducer 的业务逻辑写在 reduce() 方法中
ReduceTask 进程对每一组相同 k 的 <k,v> 组调用一次 reduce() 方法
用户自定义的 Mapper 和 Reducer 都要继承各自的父类（使用extends实现继承）

Driver端规范

整个程序需要一个 Drvier 来进行提交，提交的是一个描述了各种必要信息的 Job 对象

YARN 概述

YARN 概念

YARN（Yet Another Resource Negotiator）是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台。

Hadoop 分为三个层次：

HDFS 用于存储
YARN 用于资源管理
MapReduce 为计算引擎

换句话说，HDFS 是 Hadoop 的存储层，而 MapReduce 是 Hadoop 的处理框架，可提供任何自动并行化和分发。

从图中可以看出，除了 MapReduce ，Tez、HBase、Storm、Spark等也都在计算引擎层。Spark、Storm 等运算框架都可以整合在 YARN 上运行，只要他们各自的框架中有符合 YARN 规范的资源请求机制即可。

（1）HDFS 1.0存在的问题

NameNode 单点故障：在 Hadoop1.x 框架中，NameNode 设计为单个节点。它需要记录整个集群所有block及其副本的元数据信息和操作日志（EditsLog）。NameNode 的可用性决定整个集群的存储系统的可用性。
NameNode 压力过大，内存受限：在集群启动时，NameNode 需要将集群所有block 的元数据信息加载到内存。

（2）MapReduce1.0 存在的问题

JobTracker 单点故障：JobTracker 是单个节点，一旦该节点失败，整个集群的作业就全部失败。
不支持 MapReduce 之外的其它计算框架：Hadoop1.0 仅能支持 MapReduce 计算框架，不支持 Spark、Storm 等计算框架。
JobTracker 访问压力大，影响系统扩展性：在 Hadoop1.0 中，JobTracker 负责所有作业的调度、各个作业的生命周期、各个作业中所有 task 的跟踪和失败重启等。

YARN 的诞生

当集群中作业很多时，所有作业的 MapTask 和 ReduceTask 都要与 JobTracker 进行交互，JobTracker 的状态就变成下面这种情况：

很显然，这是 Hadoop 1.0 的一个瓶颈。

（1）HDFS 2.0

多个NameNode分管不同的目录：针对 Hadoop 1.0 中 NameNode 制约 HDFS 的扩展性问题，提出HDFS Federation 联邦以及高可用（HA）集群 。

HDFS Federation 联邦：简单理解为多个 HDFS 集群聚合到一起，更准确地理解是有多个 NameNode 节点的 HDFS 集群。
高可用（HA）集群：可以同时启动 2个 NameNode，其中一个处于工作（Active）状态，另一个处于随时待命（Standby）状态。

（2）YARN/MapReduce 2.0

为解决 Hadoop1.0 中 JobTracker 单点故障问题，Hadoop2.0 开始将 Hadoop1.0 中的 JobTracker 拆分成了两个独立的服务：

一个全局的资源管理器 ResourceManager：负责整个系统的资源管理和分配
每个应用程序特有的 ApplicationMaster：负责单个应用程序的管理

YARN 的基本组成结构：

ResourceManager（资源调度器）：Global（全局）资源管理和任务调度
NodeManager（节点管理器）：单个节点的资源管理和监控
ApplicationMaster（应用程序管理器）：单个作业的资源管理和任务监控
Container（容器）：资源申请的单位和任务运行的容器

ResourceManager 负责对各个 NodeManager 上资源进行统一管理和调度。

当用户提交一个应用程序（Application）时，需要提供一个用以跟踪和管理这个程序的 ApplicationMaster，它负责向 ResourceManager 申请资源，并要求 NodeManger 启动可以占用一定资源的任务。

ResourceManager（RM）

ResourceManager 是基于应用程序对集群资源的需求进行调度的 YARN 集群主控节点，负责协调和管理整个集群（所有 NodeManager）的资源，响应用户提交的不同类型应用程序的解析、调度、监控等工作。

ResourceManager 会为每一个应用程序（Application）启动一个 ApplicationMaster，并且 ApplicationMaster 分散在各个 NodeManager 节点。

ResourceManager主要由两个组件构成：调度器（Scheduler）和应用程序管理器（Applications Manager， ASM）。

Scheduler（调度器）：调度器根据应用程序（Application）的资源需求进行资源分配，不参与应用程序（Application）具体的执行和监控等工作。资源分配的单位就是 Container（容器）。FIFO、 Fair Scheduler 和 Capacity Scheduler。
Applications Manager（应用程序管理器）：负责管理整个系统中所有应用程序（Application），包括应用程序（Application）提交，与调度器（Scheduler）协商资源，启动和监控 ApplicationMaster 的运行状态并在失败时重新启动它等。

ResourceManager 有以下作用：

1）处理客户端请求

2）监控 NodeManager

3）启动和监控 ApplicationMaster

4）资源的分配与调度

NodeManager（NM）

NodeManager 是 YARN 集群当中真正资源的提供者，是真正执行应用程序（Application）的容器（Container）的提供者。

NodeManager 有以下作用：

（1）管理单个节点上的资源

（2）处理来自 ResourceManager 的命令

（3）处理来自 ApplicationMaster 的命令

ApplicationMaster（AM）

管理单个应用程序（Application）的生命周期，包括向资源调度器 ResourceManager 申请执行任务的资源容器（Container），运行任务，监控整个任务的执行，跟踪整个任务的状态，处理任务失败以及异常情况。

1）负责数据的切分

（2）为应用程序（Application）申请资源并分配给内部的任务

（3）任务的监控与容错

Container（容器）

Container 是一个抽象出来的逻辑资源单位。它封装了一个节点上的 CPU、内存、磁盘、网络等信息， MapReduce 程序的所有 task 都是在一个容器里执行完成的，容器的大小是可以动态调整的。

任务一：YARN 概念

YARN是什么？

ARN（Yet Another Resource Negotiator）是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台。

Hadoop分为三个层次：

HDFS 用于存储
YARN 用于资源管理
MapReduce 为计算引擎（Spark、Storm 等运算框架都可以整合在 YARN 上运行，只要他们各自的框架中有符合 YARN 规范的资源请求机制即可。）

任务二：YARN 的诞生

2.1 Hadoop 1.0

1. HDFS 1.0 存在的问题

NameNode 单点故障
NameNode 压力过大，内存受限

2. MapReduce1.0 存在的问题

JobTracker 单点故障
不支持 MapReduce 之外的其它计算框架
JobTracker 访问压力大，影响系统扩展性

2.2 Hadoop 2.0

1. HDFS 2.0

多个NameNode分管不同的目录

HDFS Federation 联邦
高可用（HA）集群

2. YARN/MapReduce 2.0

为解决 Hadoop1.0 中 JobTracker 单点故障问题，Hadoop2.0 开始将 Hadoop1.0 中的 JobTracker 拆分成了两个独立的服务：

一个全局的资源管理器 ResourceManager
每个应用程序特有的 ApplicationMaster

任务三：YARN 基本架构

1. ResourceManager（RM）

由两个组件构成：调度器（Scheduler）和应用程序管理器（Applications Manager， ASM）

Scheduler（调度器）：根据应用程序（Application）的资源需求进行资源分配。
Applications Manager（应用程序管理器）：管理整个系统中所有应用程序（Application）。

ResourceManager 有以下作用：

处理客户端请求
监控 NodeManager
启动和监控 ApplicationMaster
资源的分配与调度

2. NodeManager（NM）

NodeManager 有以下作用：

管理单个节点上的资源
处理来自 ResourceManager 的命令
处理来自 ApplicationMaster 的命令

3. ApplicationMaster（AM）

ApplicationMaster 有以下作用：

负责数据的切分
为应用程序（Application）申请资源并分配给内部的任务
任务的监控与容错

4. Container（容器）

封装了一个节点上的 CPU、内存、磁盘、网络等信息， MapReduce 程序的所有 task 都是在一个容器里执行完成的。

WordCount 的业务逻辑

MapTask 阶段处理每个数据分块的单词统计分析，思路是将每一行文本拆分成一个个的单词，每遇到一个单词则把其转换成一个 key-value 对，比如单词 Car，就转换成<’Car’,1>发送给 ReduceTask 去汇总。

ReduceTask 阶段将接收 MapTask 的结果，按照 key 对 value 做汇总计数。

回顾 MapReduce Map 端编码规范：

1. 用户自定义的 Mapper 需要继承父类 Mapper

2. Mapper 的输入数据是 KV 对的形式（KV 的类型可自定义）

3. Mapper 的输出数据是 KV 对的形式（KV 的类型可自定义）

4. Mapper 中的业务逻辑写在 map() 方法中

5. MapTask 进程对每一个<K,V>调用一次map()方法

eclipse 成功连接到 Hadoop 集群后，选择“File”->“New”->“Project”->“Map/Reduce Project”创建名为 MyMR 的项目名，在此项目下创建名为com.hongyaa.mr的包名，在此包下创建名为 WordCountMapper.java 的类，如下图所示：

首先编写 Map 端编程框架，自定义的 WordCountMapper 需要继承父类 Mapper，输入数据和输出数据都是KV 对的形式。具体框架代码如下：

public class WordCountMapper extends Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {

}

KEYIN：读到的key是一行文本的起始偏移量，所以key的类型是Long，对应 Hadoop 中的 LongWritable
VALUEIN：读到的value是一行文本的内容，所以value的类型是String，对应 Hadoop 中的 Text
KEYOUT：在此WordCount程序中，我们输出的key是单词，所以是String，对应 Hadoop 中的 Text
VALUEOUT：在此WordCount程序中,我们输出的value是单词的数量,所以是Integer，对应 Hadoop 中的 IntWritable

将框架中的KV对对应的类型修改完成后的代码如下所示：

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

}

已知 Mapper 中的业务逻辑写在 map() 方法中，在此 map()方法中我们需要实现的是将读取到的每一行文本拆分成一个个的单词，每遇到一个单词则把其转换成一个 key-value 对，比如单词 Car，就转换成<’Car’,1>发送给 ReduceTask 去汇总。具体代码如下所示：

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
	//（1）将MapTask传给我们的一行文本内容先转换成String
	String line = value.toString();  
	//（2）根据空格将这一行切分成单词
	String[] words = line.split(" ");
	//（3）将单词输出为<单词，1>
	for (String word : words) {
		//将单词作为key，将次数1作为value，以便后续的数据分发，可以根据单词分发，将相同单词分发到同一个ReduceTask中
		context.write(new Text(word), new IntWritable(1));
	}
}

已知 Mapper 中的业务逻辑写在 map() 方法中，在此 map()方法中我们需要实现的是将读取到的每一行文本拆分成一个个的单词，每遇到一个单词则把其转换成一个 key-value 对，比如单词 Car，就转换成<’Car’,1>发送给 ReduceTask 去汇总。具体代码如下所示：

@Override  
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
 //（1）将MapTask传给我们的一行文本内容先转换成String  
 String line = value.toString();  
 //（2）根据空格将这一行切分成单词  
 String[] words = line.split(" ");  
 //（3）将单词输出为<单词，1>  
 for (String word : words) {  
 //将单词作为key，将次数1作为value，以便后续的数据分发，可以根据单词分发，将相同单词分发到同一个ReduceTask中  
 context.write(new Text(word), new IntWritable(1));  
 }  
}

context对象：

使用context对象进行Map端的输出。context对象就是一个保存所有信息的中介桥梁，与Java中的session有着异曲同工之妙，Context记录map执行的上下文信息。

Map端的详细执行步骤如下：

（1）将文件按照blocksize进行拆分，默认是128M，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成<key,value>对，如下图所示。这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数（Windows和Linux环境会不同）。

（2）将分割好的<key,value>对交给用户定义的map()方法进行处理，生成新的<key,value>对，如下图所示。

使用 value.toString()..split(" ")对每行文本按照分隔符空格（源数据单词的分隔符为空格，我们使用Hadoop安装包自带的README.txt作为数据源）进行分隔。
之后得到一个数组，使用for循环遍历数组，将每个单词读取出来作为Map端的输出key，单词每出现一次就计数为1，将这个计数1作为该key对应的value。

（3）Map端的Shuffle过程。
得到map()方法输出的<key,value>对后，Mapper会将它们按照key值进行排序。如下图所示：

Java数据类型和Hadoop数据类型相互转换的方法

（1）java数据类型转换成hadoop数据类型：

第一种方法：

IntWritable number= new IntWritable(1);//将java的int型变量1封装成hadoop的整型类IntWritable的对象。  
Text text= new Text("hello world!"); //同理，将java的String类型字符串封装成hadoop的文本类Text的对象。

第二种方法：

Text text= new Text();  
text.set(“hello world!”); //Text对象text调用set（String str）方法，将传入的java String类型的字符串转化成Text类型。

（2）Hadoop数据类型转换成java数据类型

对于Text类型：

Text text= new Text();  
text.set(“hello world!”);  
text.toString(); //Text类对象调用toString方法即转换成java String类

对于除了Text的类型：

IntWritable number= new IntWritable(1);  
number.get(); //除了Text类对象调用toString方法外，其余hadoop数据类型要调用get()方法来转换成相应的java数据类型。

首先编写 Map 端编程框架，自定义的 WordCountMapper 需要继承父类 Mapper，输入数据和输出数据都是KV 对的形式。

关键字extends，表示对父类的继承，可以实现父类，也可以调用父类初始化。而且会覆盖父类定义的变量或者函数。
这里我们就需要重写父类中的map()函数，将自定义业务逻辑写在map()函数中。

具体框架代码如下：

public class WordCountMapper extends Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {  
  
}

KEYIN:是指框架读取到的数据的key的类型，在默认的InputFormat下，读到的key是一行文本的起始偏移量（第一行起始偏移量从0开始，第二行起始偏移量从第一行末尾的偏移量+1开始），所以key的类型是Long，对应 Hadoop 中的 LongWritable
VALUEIN：是指框架读取到的数据的value的类型，在默认的InputFormat下，读到的value是一行文本的内容，所以value的类型是String，对应 Hadoop 中的 Text

Inputformat: InputFormat()方法是用来生成可供map处理的<key,value>对的。

KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此WordCount程序中，我们输出的key是单词，所以是String，对应 Hadoop 中的 Text
VALUEOUT：用户自定义逻辑方法返回数据中value的类型,由用户业务逻辑决定,在此WordCount程序中,我们输出的value是单词的数量,所以是Integer，对应 Hadoop 中的 IntWritable

String，Long等jdk自带的数据类型，在序列化时，效率比较低，Hadoop为了提高序列化效率，自定义了一套序列化框架。所以，在Hadoop的程序中，如果该数据需要进行序列化（写磁盘，或者网络传输），就一定要用实现了Hadoop序列化框架的数据类型。

Long‐‐‐‐‐‐‐>LongWritable
String‐‐‐‐‐>Text
Integer‐‐‐‐>IntWritable

将框架中的KV对对应的类型修改完成后的代码如下所示：

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {  
  
}

LongWritable：Map的输入key，Map读到的key是一行文本的起始偏移量
Text：Map的输入value，Map 读到的value是一行文本的内容
Text：Map的输出key，在此WordCount程序中，我们输出的key是单词
IntWritable：Map的输出value，在此WordCount程序中，我们输出的value是单词的数量

WordCountMapper.java 的完整代码如下所示：

package com.hongyaa.mr.wordcount;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
	/**
	 * Map阶段的业务逻辑需写在自定义的map()方法中
	 * MapTask会对每一行输入数据调用一次我们自定义的map()方法
	 */
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		//（1）将MapTask传给我们的一行文本内容先转换成String
		String line = value.toString();  
		//（2）根据空格将这一行切分成单词
		String[] words = line.split(" ");
		//（3）将单词输出为<单词，1>
		for (String word : words) {
			//将单词作为key，将次数1作为value，以便后续的数据分发，可以根据单词分发，将相同单词分发到同一个ReduceTask中
			context.write(new Text(word), new IntWritable(1));
		}
	}
}

任务一：WordCount的业务逻辑

在Hadoop中，每个MapReduce任务都被初始化为一个Job，每个Job又可以分为两种阶段：map阶段和reduce阶段。

MapTask 阶段

MapTask 阶段处理每个数据分块的单词统计分析，思路是将每一行文本拆分成一个个的单词，每遇到一个单词则把其转换成一个 key-value 对，比如单词 Car，就转换成<’Car’,1>发送给 ReduceTask 去汇总。

ReduceTask 阶段

ReduceTask 阶段将接收 MapTask 的结果，按照 key 对 value 做汇总计数。

任务二：WordCount Map 端程序编写

Map端编程框架

首先编写 Map 端编程框架，自定义的 WordCountMapper 需要继承父类 Mapper，输入数据和输出数据都是KV 对的形式。

具体框架代码如下：
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {  
  
}
KEYIN:是指框架读取到的数据的key的类型，在默认的InputFormat下，读到的key是一行文本的起始偏移量（第一行起始偏移量从0开始，第二行起始偏移量从第一行末尾的偏移量+1开始），所以key的类型是Long，对应 Hadoop 中的 LongWritable

VALUEIN：是指框架读取到的数据的value的类型，在默认的InputFormat下，读到的value是一行文本的内容，所以value的类型是String，对应 Hadoop 中的 Text

KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此wordcount程序中，我们输出的key是单词，所以是String，对应 Hadoop 中的 Text

VALUEOUT：用户自定义逻辑方法返回数据中value的类型,由用户业务逻辑决定,在此wordcount程序中,我们输出的value是单词的数量,所以是Integer，对应 Hadoop 中的 IntWritable

map()方法

Mapper 中的业务逻辑写在 map() 方法中，map()方法的编程框架如下：
protected void map(LongWritable key, Text value, Context context){
}
key：对应Map端的输入key，即一行文本的起始偏移量
value：对应Map端的输入value，即一行文本的内容
context：接收Map端的输出的key-value信息

map()方法的具体实现
String[] words = value.toString().split(" ");
将maptask传给我们的文本内容value先转换成String（读入的value是Text类型，将Hadoop Text类型转换为Java String类型），之后再按空格切割单词。
for (String word : words) {
context.write(new Text(word), new IntWritable(1));
}
使用for循环遍历单词数组，读取出每个单词，将每个单词作为key，将次数1作为value，使用context对象的write()方法写入context中，帮助我们分发给reducetask。

WordCount 示例的业务逻辑

MapReduce编程模型

MapReduce采用分而治之的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果。简单来说，MapReduce就是“任务的分解和结果的汇总”。

在分布式计算中，MapReduce框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以及网络通信等复杂问题，把处理过程高度抽象为两个函数：map和reduce，map负责把任务分解成多个任务，reduce负责把分解后多任务处理的结果汇总起来。

WordCount处理过程

MapTask 阶段处理每个数据分块的单词统计分析，思路是将每一行文本拆分成一个个的单词，每遇到一个单词则把其转换成一个 key-value 对，比如单词 Car，就转换成<’Car’,1>发送给 ReduceTask 去汇总。
ReduceTask 阶段将接收 MapTask 的结果，按照 key 对 value 做汇总计数。

上节课我们实现了Map端的实现，把数据转换成了<’Car’,1>的形式，现在我们需要实现的Reduce阶段，主要是接收Map端的数据，按照key（每个单词）对value（1）做汇总计数，这样就得出每个单词出现的总次数了。

WordCount 示例的业务逻辑

MapReduce编程模型

WordCount处理过程

MapTask 阶段处理每个数据分块的单词统计分析，思路是将每一行文本拆分成一个个的单词，每遇到一个单词则把其转换成一个 key-value 对，比如单词 Car，就转换成<’Car’,1>发送给 ReduceTask 去汇总。
ReduceTask 阶段将接收 MapTask 的结果，按照 key 对 value 做汇总计数。

Reduce 端编码规范

Reduce 端规范

用户编写的程序分成三个部分： Mapper， Reducer， Driver
用户自定义的 Reducer 需要继承父类 Reducer
Reducer 的输入数据类型对应 Mapper 的输出数据类型，也是 KV（一定要注意此条）
Reducer 的输出数据是KV 对的形式（KV 的类型可自定义）
Reducer 的业务逻辑写在reduce() 方法中
ReduceTask 进程对每一组相同 k 的<k,v>组调用一次 reduce() 方法

创建WordCountReducer.java

为了更方便地调试程序，我们需要实现通过 eclipse 直接连接我们的 Hadoop 集群，直接查看 Hadoop 集群的文件信息。此步骤在6.1时详细讲述过了，这里不再赘述。

要想编写代码首先要创建相应的Project（项目），之后再创建Package（包）以及class（类）。

创建项目Project：选择 “File”->“New”->“Project”->“Map/Reduce Project”->“Next”，弹出“New MapReduce Project Wizard”对话框，为“Project name”起个名字，可以任意取名，这里我起名为MyMR。
创建包名Package：在之前创建的MapReduce项目MyMR下创建 Package包com.hongyaa.mr，包名需要全部用小写字母，并且可以有多级，之前用点分隔；
创建类名class：在Package包下创建Class类，这里类名叫做WordCountReducer.java，在此类下书写以下程序。

类名命名规范：首字母大写，通常由多个单词合成一个类名，要求每个单词的首字母也要大写，使用驼峰式命名。

上节课我们创建了MyMR项目、com.hongyaa.mr包和WordCountMapper.java类，在WordCountMapper.java类的同级目录下创建WordCountReducer.java类。

首先编写 Reduce 端编程框架，自定义的 WordCountReducer 需要继承父类 Reducer，输入数据和输出数据都是KV 对的形式，其中输入的key-value对数据类型对应Map端输出的key-value对。

关键字extends，表示对父类的继承，可以实现父类，也可以调用父类初始化。而且会覆盖父类定义的变量或者函数。
这里我们就需要重写父类中的map()函数，将自定义业务逻辑写在map()函数中。

具体框架代码如下：

public class WordCountReducer extends Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {

}

KEYIN：对应 Mapper 端输出的 KEYOUT，即单个单词，所以是 String，对应 Hadoop 中的 Text
VALUEIN：对应 Mapper 端输出的 VALUEOUT，即单词的数量，所以是Integer，对应 Hadoop 中的 IntWritable
KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此WordCount程序中，我们输出的key是单词，所以是String，对应 Hadoop 中的 Text
VALUEOUT：用户自定义逻辑方法返回数据中value的类型,由用户业务逻辑决定,在此WordCount程序中,我们输出的value是单词的出现的总次数，所以是Integer，对应 Hadoop 中的 IntWritable

Long‐‐‐‐‐‐‐>LongWritable
String‐‐‐‐‐>Text
Integer‐‐‐‐>IntWritable

将框架中的KV对对应的类型修改完成后的代码如下所示：

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

}

Text：Reduce端的输入key，对应Map端的输出key，即单个单词
IntWritable：Reduce端的输出value，对应Map端的输出value，即单词出现的次数1
Text：Reduce的输出key，在此WordCount程序中，我们输出的key是每个单词
IntWritable：Reduce的输出value，在此WordCount程序中，我们输出的value是每个单词出现的总次数

已知 Reducer 中的业务逻辑写在 reduce() 方法中，在此 reduce()方法中我们需要接收 MapTask 的输出结果，然后按照 key（单词）对 value（数量1）做汇总计数。具体代码如下所示：

/**
 * <Deer,1><Deer,1><Deer,1><Deer,1><Deer,1>
 * <Car,1><Car,1><Car,1><Car,1>
 * 框架在Map处理完成之后，将所有key-value对缓存起来，进行分组，然后传递一个组<key,values{}>，调用一次reduce()方法
 * <Deer,{1,1,1,1,1,1.....}>
 * 入参key，是一组相同单词kv对的key
 */
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
		throws IOException, InterruptedException {
	//（1）做每个key（单词）的结果汇总
	int sum = 0;
	// 遍历values，进行累加求和
	for (IntWritable v : values) {
		sum += v.get();
	}
	//（2）输出每个key（单词）和其对应的总次数
	context.write(key, new IntWritable(sum));
}

context对象：

使用context对象进行Reduce端的输出。

Reduce端的详细执行步骤如下：

（1）Reduce端的Shuffle过程。

Reducer对从Mapper接收的数据按照key值进行归并排序。如下图所示：

（2）自定义reduce()方法处理Reducer任务输入的<key,value>对。

将归并排序好的数据交由用户自定义的reduce()方法处理，之后会得到新的<key,value>对，将其作为WordCount的输出结果。如下图所示：

Java数据类型和Hadoop数据类型相互转换的方法

（1）java数据类型转换成hadoop数据类型：

IntWritable number= new IntWritable(1);//将java的int型变量1封装成hadoop的整型类IntWritable的对象。  
Text text= new Text("hello world!"); //同理，将java的String类型字符串封装成hadoop的文本类Text的对象。

（2）Hadoop数据类型转换成java数据类型

对于除了Text的类型，例如IntWritable 类型：

IntWritable number= new IntWritable(1);  
number.get(); /

回顾 MapReduce Driver 端编码规范：整个程序需要一个 Drvier 来进行提交，提交的是一个描述了各种必要信息的 job 对象

接下来进入 WordCount Driver 端程序的编写，在 com.hongyaa.mr 包下创建名为 WordCount.java 的类。

我们已经创建了MyMR项目、com.hongyaa.mr包和WordCountMapper.java类和WordCountReducer.java类，即Mapper和Reducer端都已经写好了，就差Driver端了，在同级目录下创建WordCount.java 类。

（1）创建 Job

Driver 端为该 WordCount 程序运行的入口，相当于 YARN 集群（分配运算资源）的客户端，需要创建一个 Job 类对象来管理 MapReduce 程序运行时需要的相关运行参数，最后将该 Job 类对象提交给 YARN。

Job对象指定作业执行规范，我们可以用它来控制整个作业的运行。接下来，我们分步讲述作业从提交到执行的整个过程。

1. 创建 Job

Job 的创建比较容易，其实就是 new 一个实例，先创建一个配置文件的对象，然后将配置文件对象作为参数，构造一个 Job 对象就可以了。不过new实例的方式已经过时，我们可以使用新的API创建Job。具体代码如下：

// 创建配置文件对象
Configuration conf = new Configuration();
// 新建一个 job 任务
Job job = Job.getInstance(conf);

对配置文件的配置及解析是每个框架的基本且必不可少的部分，首先我们主要对Hadoop中的配置文件的解析类Configuration的基本结构及主要方法进行介绍。

1. Configuration是什么？

Configuration做为Hadoop的一个基础功能承担着重要的责任，为YARN、HDFS、MapReduce等 提供参数的配置、配置文件的分布式传输(实现了Writable接口) 等重要功能。

2. 什么使用Configuration？

Configuration是 Hadoop 的公用类，这个类是作业的配置信息类，任何作用的配置信息必须通过Configuration传递，因为通过Configuration可以实现在多个mapper和多个reducer任务之间共享信息。

使用Configuration类的一般过程是：
（1）使用 new 构造Configuration对象；
（2）然后就可以使用get()或set()方法访问或设置配置项，资源会在第一次使用的时候自动加载到对象中。

Configuration conf=new Configuration();  # 创建Configuration对象
conf.set("fs.defaultFS", "hdfs://localhost:9000");  # 使用set()方法设置单个配置项

我们在 Hadoop 集群上运行这个作业 时，要把代码打包成一个Jar文件，只需要在Job对象的setJarByClass()方法中传递一个类即可，Hadoop会利用这个类来查找包含它的Jar文件，进而找到相关的Jar文件。具体代码如下：

// 将 job 所用到的那些类（class）文件，打成jar包 
job.setJarByClass(WordCount.class);

说明：若是直接在eclipse中运行此程序，这句代码可以不用写，但是要大成Jar包，使用hadoop jar命令去运行时此句代码就必须要写，否则会抛出ClassNotFoundException异常。

3）设置各个环节的函数

指定我们自定义的 mapper 类和 reducer 类，通过 Job 对象进行设置，将自定义的函数和具体的作业联系起来。具体代码如下：

// 指定 mapper 类和 reducer 类
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);

分别指定 MapTask 和 ReduceTask 的输出key-value类型。如果 MapTask 的输出的key-value类型与 ReduceTask 的输出key-value类型一致，则可以只指定ReduceTask 的输出key-value类型。 具体代码如下：

// 指定 MapTask 的输出key-value类型（可以省略）
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

// 指定 ReduceTask 的输出key-value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

在此WordCount示例中，MapTask输出的key-value类型分别为Text和IntWritable，ReduceTask输出的key-value类型也是Text和IntWritable，此时可以不用指定MapTask 的输出key-value类型。

因为 setOutputKeyClass 和 setOutputValueClass 默认是同时设置 Map 和 Reduce 的输出类型。

注意：必须key和value的类型都要一致才行，只有一个相同是不可以省略的。

在设置输入输出文件目录时，可以选择使用绝对目录，就是直接在语句中写入目录；也可以使用参数输入，即在运行程序时，再在控制台输入目录。具体代码如下：

// 指定该 mapreduce 程序数据的输入和输出路径，此处输入、输出为固定文件目录
Path inPath=new Path("/wordcount/input");
Path outpath=new Path("/wordcount/output");
FileInputFormat.setInputPaths(job,inPath);
FileOutputFormat.setOutputPath(job, outpath);

// 此处为参数
FileInputFormat.setInputPaths(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

在eclipse中直接运行MapReduce程序时可以使用第一种方式设定输入输出文件目录。若是打成Jar包，使用hadoop jar运行MapReduce程序时，则建议使用第二种方式。

我们Hadoop中自带的WordCount示例使用的就是第二种方式，所以运行是的时候需要指定输入和输出路径。

hadoop jar xxx.jar wordcount 输入路径 输出路径

若是在程序中指定了固定的目录，运行的时候就无需再重新指定输入输出路径了，直接使用hadoop jar xxx.jar wordcount就可以运行。

单个任务的提交可以直接使用如下语句：

job.waitForCompletion(true);

Job运行是通过job.waitForCompletion(true)，true表示将运行进度等信息及时输出给用户，false的话只是等待作业结束。

WordCount.java 的完整代码如下所示：

package com.hongyaa.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {
	/**
	 * 该MR程序运行的入口，相当于yarn集群（分配运算资源）的客户端，需要在此封装MR程序的相关运行参数，指定jar包，最后提交给yarn
	 * 
	 * 其中用一个Job类对象来管理程序运行时所需要的很多参数： 比如，指定哪个类作为map阶段的业务逻辑类，哪个类作为reduce阶段的业务逻辑类；
	 * 指定wordcount job程序的jar包所在路径...以及其他各种需要的参数。
	 */
	public static void main(String[] args) throws Exception {
		// （1）创建配置文件对象
		Configuration conf = new Configuration();

		// （2）新建一个 job 任务
		Job job = Job.getInstance(conf);

		// （3）将 job 所用到的那些类（class）文件，打成jar包 （打成jar包在集群运行必须写）
		job.setJarByClass(WordCount.class);

		// （4）指定 mapper 类和 reducer 类
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);

		// （5）指定 MapTask 的输出key-value类型（可以省略）
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		// （6）指定 ReduceTask 的输出key-value类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		// （7）指定该 mapreduce 程序数据的输入和输出路径
		Path inPath=new Path("/wordcount/input");
		Path outpath=new Path("/wordcount/output");
                // 获取 fs 对象
		FileSystem fs=FileSystem.get(conf);
		if(fs.exists(outpath)){
			fs.delete(outpath,true);
		}
		FileInputFormat.setInputPaths(job,inPath);
		FileOutputFormat.setOutputPath(job, outpath);

		// （8）最后给YARN来运行，等着集群运行完成返回反馈信息，客户端退出
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}
}

system.exit(0)和system.exit(1)的区别：

system.exit(0)：正常退出，程序正常执行结束退出。
system.exit(1)：是非正常退出，就是说无论程序正在执行与否，都退出。

本地运行模式要点

1. 本地运行模式（eclipse 开发环境下本地运行，好处是方便调试和测试）

要点一： MapReduce 程序是被提交给 LocalJobRunner 在本地以单进程的形式运行
要点二：数据输入输出可以在本地，也可以在 HDFS
要点三：怎么实现本地运行？
- 在你的 MapReduce 程序当中不要带集群的配置文件（本质就是由 mapreduce.framework.name 和 yarn.resourcemanager.hostname 这两个参数决定）

LocalJobRunner解析：
Hadoop作业分本地模式和集群模式两种执行模式，JobClient 初始化时会读取配置项 mapreduce.framework.name(默认为local)，如果该配置项的值为local，则 Hadoop 采用本地模式执行作业，否则采用集群模式执行。

本地模式使用LocalJobRuner提交并执行作业。对LocalJobRunner实例调用submitJob( )方法会创建Job(LocalJobRunner的内部类)实例，该实例完成作业的执行

若是在本地，由以下2个参数决定：

conf.set("mapreduce.framework.name", "local"); // 指定MapReduce运行时框架为本地作业运行器，该参数的默认值就是local
conf.set("fs.defaultFS", "file:///");  // 获取本地文件系统实例，该参数的默认值就是file:///

需要注意的是，这里设置的是在Linux本地运行，相应的输入输出路径要设置成Linux本地的路径（路径可以任意）：

Path inPath=new Path("/root/software/wordcount/input");  // 需要提前创建，并放入数据源文件
Path outpath=new Path("/root/software/wordcount/output");  // 不需提前创建，程序运行时会自动创建
FileInputFormat.setInputPaths(job,inPath);
FileOutputFormat.setOutputPath(job, outpath);

若是在HDFS上，则由以下2个参数决定：

conf.set("mapreduce.framework.name", "local");    
conf.set("fs.defaultFS", "hdfs://localhost:9000"); // HDFS集群中NameNode的URI，获取DistributedFileSystem实例

此时我们操作的就不是Linux本地了，而是HDFS，相应的输入输出路径要设置成HDFS的路径（路径可以任意）：

Path inPath=new Path("/wordcount/input");
Path outpath=new Path("/wordcount/output");
FileInputFormat.setInputPaths(job,inPath);
FileOutputFormat.setOutputPath(job, outpath);

集群运行模式要点

2. 集群运行模式（打 jar 包，提交任务到集群运行）

将 MapReduce 程序提交给 YARN 集群 ResourceManager，分发到多个节点上并发执行。处理的输入数据和输出结果都应位于 HDFS 文件系统。

要点一：首先要把代码打成 jar 包放在 Linux 本地
要点二：用 hadoop jar 的命令去提交代码到 YARN 集群运行
要点三：处理的输入数据和输出结果都应位于 HDFS 文件系统

集群运行模式需要配置以下参数：

conf.set("mapreduce.framework.name", "yarn");// 指定MapReduce运行时框架为YARN   
conf.set("yarn.resourcemanager.hostname", "localhost"); // 指定 ResourceManager 守护进程所在主机  
conf.set("fs.defaultFS", "hdfs://localhost:9000");// HDFS集群中NameNode的URI，获取DistributedFileSystem实例

以上三个参数在Hadoop集群搭建的时候在配置文件中都已经配置过了，如果我们打jar包在Hadoop集群中运行时，其实可以不用配置以上这三个参数。

具体打 jar 包运行的步骤如下所示：　　

（1）配置完以上参数后，右键项目MyMR，选择“Export”：

2）选择“Java”——》“JAR file”，之后点击“Next”：

（3）在弹出的对话框中，勾选MyMR项目中的 src，然后在JAR file中选择保存的路径名（包含最终的jar包的名字），之后点击“Finish”：

（4）将Hadoop安装包中自带的README.txt文件上传到HDFS的/wordcount/input目录下，将其作为数据源。进入jar包所在的目录，使用以下命令运行jar包：

hadoop jar wordcount.jar com.hongyaa.mr.wordcount

解析：

hadoop jar：用来执行一个Hadoop的jar包程序
wordcount.jar：执行的jar包的名字
com.hongyaa.mr.wordcount：包名.类名，此处的类名为我们主类的名字

因为我们在代码中已经设定了固定的输入输出路径，所以在运行时就可以不用再指定输入输出路径了。

hadoop jar wordcount.jar com.hongyaa.mr.wordcount执行示意图：

从图中可以看到Job的ID，该Job提交到Yarn集群运行，并且运行成功（Map端100%，Reduce端100%）。

（5）在本机的浏览器上访问**http://localhost:8088**进入YARN集群的Web UI界面，在此界面中查看该Job是否成功提交到YARN集群。

从上图可以看出该Job成功提交给YARN集群运行，并且运行成功。

（6）使用 HDFS Shell操作查看运行结果，运行结果储存在HDFS的/wordcount/output目录下：

hadoop fs -cat /wordcount/output/part-r-00000

运行结果如下所示：

什么是 Combiner？

Combiner 是 MapReduce 程序中 Mapper 和 Reducer 之外的一种组件，它的作用是在 MapTask 之后给 MapTask 的结果进行局部汇总，以减轻 ReduceTask 的计算负载，减少网络传输。

Combiner 最基本的是实现本地key的聚合，对 Map 输出的 key 排序，value 进行迭代，有本地 Reduce 之称，实际上就是继承 Reducer 类，本质上就是一个 Reducer。

这里引用别人的一个例子：map与reduce的例子

map理解为销售人员，reduce理解为销售经理。

每个人（map）只管销售,赚了多少钱销售人员不统计，也就是说这个销售人员没有Combine，那么这个销售经理就累垮了，因为每个人都没有统计，它需要统计所有人员卖了多少件，赚钱了多少钱。

这样是不行的，所以销售经理（reduce）为了减轻压力，每个人（map）都必须统计自己卖了多少钱，赚了多少钱（Combine），然后经理所做的事情就是统计每个人统计之后的结果。这样经理就轻松多了。所以Combine在map所做的事情，减轻了reduce的事情。

数据格式的转换

MapReduce框架的运作基于键值对，即数据的输入是键值对，生成的结果也是存放在集合里的键值对，其中键值对的值也是一个集合，一个MapReduce任务的执行过程以及数据输入输出的类型如下所示，这里我们定义list表示集合：

map: (K1, V1) → list(K2, V2) 
combiner: (K2, list(V2)) → list(K3, V3) 
reduce: (K3, list(V3)) → list(K4, V4)

map()函数操作所产生的键值对会作为combine函数的输入，经combine函数处理后再送到reduce函数进行处理，减少了写入磁盘的数据量，同时也减少了网络中键值对的传输量。

在Map端，用户自定义实现的Combine优化机制类Combiner在执行Map端任务的节点本身运行，相当于对map函数的输出做了一次reduce（按照key对value进行汇总）。

集群上的可用带宽往往是有限的，产生的中间临时数据量很大时就会出现性能瓶颈，因此尽量避免Map端任务和Reduce端任务之间大量的数据传输是很重要的。使用Combine机制的意义就在于使Map端输出更紧凑，使得写到本地磁盘和传给Reduce端的数据更少。

选用Combine机制下的Combiner虽然减少了IO，但是等于多做了一次reduce，所以应该查看作业日志来判断combine函数的输出记录数是否明显少于输入记录的数量，以确定这种减少和花费额外的时间来运行Combiner相比是否值得。

注：现在想想，如果在 WordCount 中不用 Combiner，那么所有的结果都是 reduce 完成，效率会相对低下。使用 Combiner 之后，先完成的 Map 会在本地聚合，提升速度。对于 WordCount 的例子，value 就是一个叠加的数字，所以 Map 一结束就可以进行 Reduce 的 value 叠加，而不必要等到所有的 Map 结束再去进行 Reduce 的 value 叠加。

添加设置 Combiner 的实现步骤

Combiner 的意义就是对每一个 MapTask 的输出进行局部汇总，以减小网络传输量。

具体实现步骤：

（1）自定义一个 Combiner 继承 Reducer，重写 reduce 方法，其实现的功能和之前我们写的WordCountReducer是一样的，所以reduce方法中的实现也是一样的。
（2）在 Job 中设置：job.setCombinerClass(xxx.class)（也就是Driver端，即WordCount.java中设置，与指定 mapper 类和 reducer 类的代码放在一起）

具体代码实现

具体代码实现：
（1）Combiner 的代码（和 Reduce 端的代码一致，本质上就是一个 Reducer）

/**
 * Combiner
 */
public static class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Reducer<Text, IntWritable, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		int sum = 0;
		for (IntWritable v : values) {
			sum += v.get();
		}
		context.write(key, new IntWritable(sum));
	}
}

（2）在 Job 中设置 Combiner 类

job.setCombinerClass(WordCountCombiner.class);

在WordCount示例中，WordCountCombiner与WordCountReducer实现的功能是一样的，所以我们可以不用另外写WordCountCombiner.java了，可以直接使用WordCountReducer.java，只需要在Driver端指定清楚即可。如下所示：

job.setMapperClass(WordCountMapper.class);  
job.setCombinerClass(WordCountReducer.class);
job.setReducerClass(WordCountReducer.class);

说明：

“词频统计”是一个可以展示 Combiner 用处的基础例子，上面的 WordCount 程序为每一行数据生成了一个（word，1）键值对。
所以如果在同一个文档内“Car”出现了3次，（"Car"，1）键值对会被生成3次，这些键值对会被送到 Reduce 端。
通过使用 Combiner，这些键值对可以被压缩为一个（"Car"，3）送往 Reduce 端。现在每一个节点针对每一个词只会发送一个值到 Reduce 端，大大减少了 Shuffle 过程所需要的带宽并加速了作业的执行。
这里面最棒的就是我们不用写任何额外的代码就可以享用此功能！如果你的 Reduce 是可交换及可组合的，那么它也就可以作为一个 Combiner。你只要在 Driver 中添加job.setCombinerClass（xxx.class）;这行代码就可以在“词频统计”程序中启用Combiner。

使用 Combiner 的注意事项

（1）Combiner 和 Reducer 的区别在于运行的位置：

Combiner 是在每一个 MapTask 所在的节点运行
Reducer 是接收全局所有 Mapper 的输出结果

（2）Combiner 的输出 KV 跟 Reducer 的输入 KV 类型相对应。
（3）不要以为在写 MapReduce 程序时设置了 Combiner 就认为 Combiner 一定会起作用，实际情况是这样的吗？答案是否定的。

Hadoop 文档中也有说明 Combiner 可能被执行也可能不被执行，可能执行一次也可能执行多次。
那么在什么情况下不执行呢？ 如果当前集群在很繁忙的情况下 Job 就是设置了也不会执行 Combiner，这时集群本身负载量很大，会尽量提早执行完 Map，空出资源。所以， Combiner 使用的原则是：有或没有都不能影响业务逻辑，都不能影响最终结果。

（4）Combiner的适用场景比如说在 “汇总统计”、“求最大值” 时，就可以使用 Combiner，但是在“求平均数”的时候就不适用了。

在求取平均数时，因为添加的Combiner组件是与reduce组件具有相同的逻辑，会提前求一次平均值后传给reduce类，导致求取的平均值错误。

总结：
Combiner 可以理解为是在map端的reduce的操作，对单个map任务的输出结果数据进行合并的操作。
Combiner是作为一个优化手段，可选项，不是所有的MR程序都适合Combiner。
Combiner的优化是一定不能够改变最终的输出的结果。

好处：

减少网络的传输
减轻磁盘IO负载

. MapTask 并行度

MapTask 并行度

MapTask 并行度决定 Map 阶段的任务处理并发度，进而影响到整个 Job 的处理速度。

那么， MapTask 并行实例是否越多越好呢？其并行度又是如何决定呢？

一个 Job 的 Map 阶段并行度由客户端在提交 Job 时决定，客户端对 Map 阶段并行度的规划的基本逻辑为：

将待处理数据执行逻辑切片（即按照一个特定切片大小（默认是128M），将待处理数据划分成逻辑上的多个 split），然后每一个 split 分配一个 MapTask 并行实例处理。

这段逻辑及形成的切片规划描述文件，是由 FileInputFormat 实现类的 getSplits() 方法完成的。该方法返回的是 List<InputSplit>， InputSplit 封装了每一个逻辑切片的信息，包括长度和位置信息，而 getSplits() 方法返回一组 InputSplit。

getSplits()方法主要完成数据切分的功能，它会尝试着将输入数据切分成 InputSplit，并放入集合List中返回。
InputSplit有以下特点：
（1）逻辑分片：它只是在逻辑上对输入数据进行分片，并不会在磁盘上将其切分成分片进行存储。 InputSplit 只记录了分片的元数据信息，比如起始位置、长度以及所在的节点列表等。
（2）可序列化：在 Hadoop 中，对象序列化主要有两个作用：进程间通信和永久存储。此处， InputSplit 支持序列化操作主要是为了进程间通信。作业被提交到 ResourceManager 之前， Client 会调用作业 InputFormat 中的 getSplits() 函数，并将得到的 InputSplit 序列化到文件中。这样，当作业提交到 ResourceManager 端对作业初始化时，可直接读取该文件，解析出所有 InputSplit，并创建对应的 MapTask。而 createRecordReader 则根据InputSplit ，将其解析成一个个 key/value 对。

FileInputFormat 切片机制

1. FileInputFormat 中默认的切片机制：

（1）简单地按照文件的内容长度进行切片
（2）切片大小，默认等于 block 大小，即128M
（3）切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

比如待处理数据有两个文件：

File1.txt  200M  
File2.txt  100M

经过 getSplits() 方法处理之后，形成的切片信息是：

File1.txt-split1  0-128M  
File1.txt-split2  129M-200M  
File2.txt-split1  0-100M

他是针对每一个文件的单独切片，不是数据集整体。

2. FileInputFormat 中切片的大小的参数配置：

计算切片大小：long splitSize = computeSplitSize(Math.max(minSize, Math.min(maxSize, blockSize)))，翻译一下就是求这三个值的中间值。

切片主要由这几个值来运算决定：

blocksize：默认是 128M，可通过 dfs.blocksize 修改
minSize：默认是 1，可通过 mapreduce.input.fileinputformat.split.minsize 修改
maxsize：默认是 Long.MaxValue，可通过mapreduce.input.fileinputformat.split.maxsize 修改

因此，默认情况下，切片大小等于 blocksize。

如果 maxsize 调的比 blocksize 小，则切片会小于 blocksize，而且就等于配置的这个参数的值；
如果 minsize 调的比 blocksize 大，则切片会大于 blocksize。但是，不论怎么调参数，都不能让多个小文件“划入”一个 split。

.setNumReduceTasks(4);  //默认值是 1，手动设置为 4

ReduceTask 的数量默认为 1，我们手动设置为 4，表示运行 4 个 ReduceTask，相应的输出结果会有4个，如下图所示：

注意：这四个输出文件中的内容是不一样的，从文件大小上也可以看出其是不一样的，也可以使用cat命令查看里面的内容进行验证。

那么每一条结果去到哪个结果文件中呢？

数据的key.hashcode%ReduceTask数==本ReduceTask号

ReduceTask号默认从0开始。之后再讲解自定义分区的时候会用到，到时候会再详细讲述。

如果设置为 0，表示不运行 ReduceTask 任务，也就是没有 Reduce 阶段，只有 Map 阶段，Map 阶段的输出结果作为最终的输出结果。

job.setNumReduceTasks(0);唯一影响的是map结果的输出方式：

当job.setNumReduceTasks(0);时，即没有reduce阶段，此时唯一影响的就是map结果的输出方式。

如果有reduce阶段，map的结果被flush到硬盘，作为reduce的输入； reduce的结果将被OutputFormat的RecordWriter写到指定的地方（setOutputPath），作为整个程序的输出。
如果没有reduce阶段，map的结果将直接被OutputFormat的RecordWriter写到指定的地方（setOutputPath），作为整个程序的输出。

总结：有reduce时reduce的结果作为整个程序的输出；无reduce时，map的结果作为整个程序的输出。

如果数据分布不均匀，就有可能在 Reduce 阶段产生数据倾斜。

1. 对于maptask，1个逻辑切片对应1个block块，数据比较均匀，并行度比较高。

2. 对于reducetask存在数据倾斜的风险：数据倾斜指的是每个reducetask的数据分配不均匀，产生数据倾斜。
* 有的reducatesk分配的任务比较多，就会造成当前的reducetask处理的时间长；
* 有的reducatesk分配的任务比较少，就会造成当前的reducetask处理的时间短。

3. 存在数据倾斜的原因：分区算法中对map端输出的数据分配的不均匀。

4. 如何避免数据倾斜：合理的设计分区算法。

默认的分区公式是：数据的key.hashcode%ReduceTask数==本ReduceTask号，之后我们会学习自定义分区，到时候再讲述如何设计分区。

注意： ReduceTask 数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有 1 个 ReduceTask（因为我们想要的汇总是一个整体的汇总）。

MapReduce 结构

一个完整的 MapReduce 程序在分布式运行时有两类实例进程：

MRAppMaster（MapReduce Application Master）：负责整个程序的过程调度及状态协调。

MRAppMaster是MapReduce的ApplicationMaster实现，它使得MapReduce可以直接运行在YARN上，它主要作用在于管理作业的生命周期。

YarnChild（MapTask）：负责 Map 阶段的整个数据处理流程，阶段并发任务
YarnChild（ReduceTask）：负责 Reduce 阶段的整个数据处理流程，阶段汇总任务

什么是YarnChild？
答：MRAppMaster运行程序时向 ResouceManager 请求的MapTask/ReduceTask，也是运行程序的容器。其实它就是一个运行程序的进程。

以上两个阶段 MapTask 和 ReduceTask 的进程都是 YarnChild，并不是说这 MapTask 和 ReduceTask 就跑在同一个 YarnChild 进程里。

查看进程2

运行完 MapTask 阶段，此 YarnChild 进程会关闭，随后再运行 ReduceTask 阶段，此时还会开启一个名为 YarnChild 的进程，但是通过查看进程号发现，此时的 YarnChild 进程是一个新的进程，与 MapTask 阶段的 YarnChild 不是同一个进程，如下图所示：

等整个 MapReduce 程序运行完成后，YarnChild 和 MRAppMaster 进程都会自己关闭。

MapReduce 程序的运行流程1-2

MapReduce 程序的运行流程

1.一个 MapReduce 程序启动的时候，最先启动的是 MRAppMaster， MRAppMaster 启动后根据本次 Job 的描述信息，计算出需要的 MapTask 实例数量，然后向集群申请机器启动相应数量的 MapTask 进程；

MapTask的数量是由输入文件的切片数决定的，切片的大小默认等于 block，即128M。

2.MapTask 进程启动之后，根据给定的数据切片(哪个文件的哪个偏移量范围)范围进行数据处理，主体流程为：

利用客户指定的 InputFormat 来获取 RecordReader 读取数据，形成输入 KV 对

InputFormat

Hadoop在 org.apache.hadoop.mapreduce.lib.input 包里提供了一些InputFormat的实现：
* FileInputFormat: 这是一个抽象基类，可以作为任何基于文本输入的父类（WordCount中使用的就是此类）
* SequenceFileInputFormat: 这是一个高效的二进制文件格式
* TextInputFormat: 它用于普通文本文件

RecordReader

Hadoop在 org.apache.hadoop.mapreduce.lib.input 包里也提供了一些常见的RecordReader实现：
* LineRecordReader: 这是RecordReader类对文本文件的默认实现，它将行号设置为key并将该行内容视为value。
* **SequenceFileRecordReader**: 该类从二进制文件SequenceFile读取键值

将输入 KV 对传递给客户定义的 map() 方法，做逻辑运算，并将 map() 方法输出的 KV 对收集到缓存
将缓存中的 KV 对按照 K 分区排序后不断溢写到磁盘文件（此过程是在MapTask端的shuffle阶段，排序时按照字典顺序排序）

MapReduce 程序的运行流程3-4

MapReduce 程序的运行流程

3.MRAppMaster 监控到所有 MapTask 进程任务完成之后（真实情况是，某些 MapTask 进程处理完成后，就会开始启动 ReduceTask 去已完成的 MapTask 处 fetch（拿取）数据），会根据客户指定的参数启动相应数量的 ReduceTask 进程，并告知 ReduceTask 进程要处理的数据范围（数据分区）；

ReduceTask 数量的决定是可以直接手动设置：

```java
job.setNumReduceTasks(数量); 	//默认值是 1

4.ReduceTask 进程启动之后，根据 MRAppMaster 告知的待处理数据所在位置，从若干台 MapTask 运行所在机器上获取到若干个 MapTask 输出结果文件，并在本地进行重新归并排序，然后按照相同 key 的 KV 为一个组，调用客户定义的 reduce() 方法进行逻辑运算，并收集运算输出的结果 KV，然后调用客户指定的 OutputFormat 将结果数据输出到外部存储。
注意：输出文件的个数与ReduceTask的个数有关。

WordCount 实现流程

总结 WordCount 程序的详细实现流程：

本节总结-导图
任务一：MapTask 并行度决定机制

1. MapTask 并行度

MapTask 并行度决定 Map 阶段的任务处理并发度，进而影响到整个 Job 的处理速度。那么， MapTask 并行实例是否越多越好呢？其并行度又是如何决定呢？

一个 Job 的 Map 阶段并行度由客户端在提交 Job 时决定，客户端对 Map 阶段并行度的规划的基本逻辑为：
将待处理数据执行逻辑切片（即按照一个特定切片大小，将待处理数据划分成逻辑上的多个 split），然后每一个 split 分配一个 MapTask 并行实例处理。

2. FileInputFormat 切片机制

FileInputFormat 中默认的切片机制：
（1）简单地按照文件的内容长度进行切片
（2）切片大小，默认等于 block 大小
（3）切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

计算切片大小：long splitSize = computeSplitSize(Math.max(minSize, Math.min(maxSize, blockSize)))，翻译一下就是求这三个值的中间值。

切片主要由这几个值来运算决定，都有默认值，当然都可以通过参数修改默认值：

blocksize：默认是 128M。
minSize：默认是 1。
maxsize：默认是 Long.MaxValue。

因此，默认情况下，切片大小等于 blocksize。

任务二：ReduceTask 并行度决定机制

ReduceTask 的并行度同样影响整个 Job 的执行并发度和执行效率，但与 MapTask 的并发数由切片数决定不同， ReduceTask 数量的决定是可以直接手动设置：
job.setNumReduceTasks(数量);  //默认值是 1
如果设置为 0，表示不运行 ReduceTask 任务，也就是没有 Reduce 阶段，只有 Map 阶段，Map 阶段的输出结果作为最终的输出结果。

任务三：MapReduce 程序的运行流程

1. MapReduce结构

一个完整的 MapReduce 程序在分布式运行时有两类实例进程：

MRAppMaster（MapReduce Application Master）：负责整个程序的过程调度及状态协调
YarnChild（MapTask）：负责 Map 阶段的整个数据处理流程，阶段并发任务
YarnChild（ReduceTask）：负责 Reduce 阶段的整个数据处理流程，阶段汇总任务

以上两个阶段 MapTask 和 ReduceTask 的进程都是 YarnChild，并不是说这 MapTask 和 ReduceTask 就跑在同一个 YarnChild 进程里。

2. MapReduce 程序的运行流程

一个 MapReduce 程序启动的时候，最先启动的是 MRAppMaster， MRAppMaster 启动后根据本次 Job 的描述信息向集群申请启动相应数量的 MapTask 进程；

MapTask 进程启动之后，读取输入数据，将其形成 KV 对形式传递给客户定义的 map() 方法，做逻辑运算，并将 map() 方法输出的 KV 对收集到缓存。将缓存中的 KV 对按照 K 分区排序后不断溢写到磁盘文件。

MRAppMaster 监控到只有有一个 MapTask 进程任务完成，就会启动 ReduceTask 去已完成的 MapTask 处 fetch 数据，会根据客户指定的参数启动相应数量的 ReduceTask 进程；

ReduceTask 获取到 MapTask 运行输出结果后，会在本地进行重新归并排序，然后按照相同 key 的 KV 为一个组，调用客户定义的 reduce() 方法进行逻辑运算，并收集运算输出的结果 KV，然后将结果数据输出到外部存储。

YARN 工作原理描述：

Client 提交作业到YARN上；
ResourceManager 选择一个 NodeManager ，启动一个 Container；
在 Container 中运行 ApplicationMaster 实例；
ApplicationMaster 根据实际需要向 ResourceManager 请求更多的 Container 资源（如果作业很小，ApplicationMaster 会选择在其自己的 JVM 中运行任务）；
ApplicationMaster 通过获取到的 Container 资源执行分布式计算。

5个独立实体

在最高层有5个独立实体：
（1）客户端(Client)：提交 MapReduce 作业。
（2）YARN 资源管理器 (ResourceManager)：负责协调集群上计算机资源的分配。
（3）YARN 节点管理器 (NodeManager)：负责启动和监视集群中机器上的计算容器（Container）。
（4）MapReduce 的应用管理器 (ApplicationMaster)：负责协调运行 MapReduce 作业的任务。它和 MapReduce 任务在容器（Container）中运行，这些容器由资源管理器分配并由节点管理器进行管理。
（5）分布式文件系统：一般为 HDFS，用来与其他实体间共享作业文件。

上一张PPT中的YARN作业提交流程可以大体总结为以下6步。

作业提交
作业初始化
任务分配
任务运行
进度和状态更新
任务完成

作业提交

步骤1：Client 调用 job.waitForCompletion() 方法，向整个集群提交 MapReduce 作业（在 YARN 中，作业一般称为 Application 应用程序）。

步骤2：新的作业ID（应用ID，Application ID）由资源管理器 (ResourceManager) 分配。

步骤3：作业的 Client 核实作业的输出，计算输入的 split，将作业的资源（包括 Jar 包，配置文件， split信息）拷贝给 HDFS。也就是通过 HDFS 共享程序的 jar 包，供 Task 进程读取。

步骤4：最后，通过调用资源管理器 (ResourceManager) 的 submitApplication() 来提交作业。

步骤5a：当资源管理器 (ResourceManager) 收到 submitApplciation() 的请求时，就将该请求发给调度器 (scheduler)，调度器分配容器 (Container)；

步骤5b：然后资源管理器在该容器内启动应用管理器(ApplicationMaster)进程（MapReduce 作业的应用管理器是一个主类为 MRAppMaster 的 Java 应用），由节点管理器 (NodeManager) 监控。

步骤6：应用管理器 (ApplicationMaster) 初始化作业（应用程序），初始化动作包括创建监听对象来监控作业的进度和执行情况，得到任务的进度和完成报告。

步骤7：应用管理器 (ApplicationMaster) 根据作业代码中指定的数据地址（数据源一般来自 HDFS）进行数据分片，以确定 Mapper 任务数，具体每个 Mapper 任务发往哪个计算节点。同时还会计算 Reduce 任务数，Reduce 任务数是在程序中通过job.setNumReduceTask()显示指定的。

应用管理器 (ApplicationMaster) 必须决定如何运行构成 MapReduce 作业的各个任务。如果作业很小，就选择和自己在同一个JVM上运行任务。

默认情况下，小作业就是少于10个 Mapper 且只有1个 Reduce 且输入大小小于一个HDFS块的作业。这样的作业称为Uberized，或者作为Uber任务执行。默认情况下，Uber 模式是关闭的。

步骤8：如果任务不适合Uber任务运行，应用管理器 (ApplicationMaster) 就会采用轮询的方式通过 RPC 协议向资源管理器 (ResourceManager) 申请和领取资源。

步骤9a：应用管理器 (ApplicationMaster) 申请到计算资源后（由 ResourceManager 的 Scheduler 负责分配），应用管理器通知节点管理器 (NodeManager) 启动一个容器 (Container)，这些容器用于执行 Map 任务或者 Reduce 任务。

步骤9b：启动容器 (Container) 后，即开始执行 Map 任务或者 Reduce 任务，任务由一个主类为 YarnChild 的 Java 应用执行。

步骤10：在运行任务之前首先本地化任务需要的资源，比如作业配置，jar文件以及分布式缓存的所有文件。

步骤11：最后，运行 Map 任务或者 Reduce 任务。

5. 进度和状态更新

一个作业和它的每个任务都有一个状态（Status），包括：作业或任务的状态（比如：运行中，运行成功，运行失败）、Map和Reduce的进度、作业计数器的值、状态消息或描述。

Task two：Yarn job submission process

MapReduce 的进度组成：

读入一条输入记录（在 Mapper 或 Reducer 中）；
写入一条输出记录（在 Mapper 或 Reducer 中）；
设置状态描述（通过 Reporter 或 TaskAttemptContext 的setStatus()方法）；
增加计数器的值（使用Reporter的 incrCounter()方法或Counter的increment()方法）；
调用 Reporter 或 TaskAttemptContext 的progress()方法。

进度和状态更新

在作业期间，客户端每秒钟轮询一次（轮询隔间可通过 mapreduce.client.progressmonitor.pollinterval设置），保证应用管理器 (ApplicationMaster) 接收到最新状态。客户端也可以使用 Job 的getStatus()方法得到一个JobStatus的实例，里面包含作业的所有状态信息。

作业完成

当应用管理器 (ApplicationMaster) 收到作业最后一个任务已完成的通知后，便把作业的状态设置为“成功”。然后，在Job轮询状态时，便知道任务已成功完成，于是Job打印一条消息告知用户，然后从 waitForCompletion() 方法返回。Job 的统计信息和计数值也在这个时候输出到控制台。

作业完成之后，应用管理器 (ApplicationMaster) 和任务容器 (Container) 会清理工作状态（这样中间输出将被删除）， OutputCommiter 的作业清理方法也会被调用。作业的信息会被作业历史服务器存储，以便日后用户需要时可以查询。

Job 失败

Task失败
应用管理器失败
节点管理器失败
资源管理器失败

Task 失败

（1）用户自定义的 MapTask 或者 ReduceTask 在运行过程中抛出异常，Task JVM 会将异常上报到应用管理器 (ApplicationMaster) 然后再退出。

（2）Task JVM 由于某种原因突然终止了，但是没有来得及向应用管理器 (ApplicationMaster) 汇报。

3）Task进程被挂起了。

（4）默认情况下，任何一个Task失败了四次，则整个Job被认为是失败的

应用管理器失败

MapReduce 的应用管理器 (ApplicationMaster) 失败了会再次尝试运行，默认会尝试运行两次。如果两次都失败则 Job 运行失败。mapreduce.am.max-attempts设置最多尝试次数。

应用管理器失败重启机制：

应用管理器 (ApplicationMaster) 会周期性向资源管理器(ResourceManager)发送心跳，如果资源管理器发现 ApplicationMaster 已经挂掉了，那么它会在一个新的容器(Container) 中开启另一个 ApplicationMaster 实例。
此时新的 ApplicationMaster 实例会利用 Job 的历史服务器去恢复正在运行的 Task 的状态。如果不想让 Job 继续执行，则可以将yarn.app.mapreduce.am.job.recovery.enable设置为 false 关闭此功能。

节点管理器失败

默认情况下，如果资源管理器(ResourceManager) 超过10分钟（yarn.resourcemanager.nm.liveness-monitor.expiry-interval-ms）没有收到节点管理器 (NodeManager) 的心跳消息，则认为它已经挂了，然后将它从自己的列表中删除。

资源管理器失败

默认配置下，资源管理器(ResourceManager) 存在单点故障，一旦失败，则所有 Job 都不被恢复。

为了高可用，可以同时运行两个资源管理器，另一作为备用。

客户端和节点管理器(NodeManager) 也要被配置成可以处理资源管理器失败的情况。

任务一：YARN 工作原理简述

Client 提交作业到YARN上；
ResourceManager 选择一个 NodeManager ，启动一个 Container；
在 Container 中运行 ApplicationMaster 实例；
ApplicationMaster 根据实际需要向 ResourceManager 请求更多的 Container 资源（如果作业很小，ApplicationMaster 会选择在其自己的 JVM 中运行任务）；
ApplicationMaster 通过获取到的 Container 资源执行分布式计算。

任务二：YARN 作业提交流程

在最高层有5个独立实体：

（1）客户端(Client)：提交 MapReduce 作业。
（2）YARN 资源管理器 (ResourceManager)：负责协调集群上计算机资源的分配。
（3）YARN 节点管理器 (NodeManager)：负责启动和监视集群中机器上的计算容器（Container）。
（4）MapReduce 的应用管理器 (ApplicationMaster)：负责协调运行 MapReduce 作业的任务。它和 MapReduce 任务在容器（Container）中运行，这些容器由资源管理器分配并由节点管理器进行管理。
（5）分布式文件系统：一般为 HDFS，用来与其他实体间共享作业文件。

主要流程

作业提交：Client 调用 job.waitForCompletion() 方法，向整个集群提交 MapReduce 作业。

作业初始化：资源管理器将请求发送给调度器 (scheduler)，调度器分配容器 (Container)，在该容器内启动应用管理器(ApplicationMaster)进程**，应用管理器 (ApplicationMaster) 初始化作业（应用程序）。

任务分配：应用管理器 (ApplicationMaster) 采用轮询的方式通过 RPC 协议向资源管理器 (ResourceManager) 申请和领取资源。

任务运行：应用管理器通知节点管理器 (NodeManager) 启动一个容器 (Container)，这些容器用于执行 Map 任务或者 Reduce 任务。

进度和状态更新：作业或任务的状态（比如：运行中，运行成功，运行失败）、Map和Reduce的进度等。

任务完成：把作业的状态设置为“成功”。

任务三：Job 失败

在真实环境中，Job失败是很常见的。通常失败包括Task失败、应用管理器失败、节点管理器失败、资源管理器失败。

实时课堂

留言提问

0

Shuffle工作机制

在 Hadoop 中数据从 Map 阶段传递给 Reduce 阶段的过程就叫 Shuffle，Shuffle 机制是整个 MapReduce 框架中最核心的部分。

Shuffle的本义是洗牌、混洗，把一组有一定规则的数据尽量转换成一组无规则的数据，越随机越好。MapReduce中的Shuffle更像是洗牌的逆过程，把一组无规则的数据尽量转换成一组具有一定规则的数据。

MapReduce计算模型为什么需要 Shuffle 过程？

我们都知道MapReduce计算模型一般包括两个重要的阶段：Map是映射，负责数据的过滤分发；Reduce是规约，负责数据的计算归并。Reduce的数据来源于Map，Map的输出即是Reduce的输入，Reduce需要通过Shuffle来获取数据。

具体来说：就是将 MapTask 输出的处理结果数据，分发给 ReduceTask，并在分发的过程中，对数据按 key 进行了分区和排序。

Shuffle 描述的是数据从 Map 端到 Reduce 端的过程，大致分为排序（sort）、溢写（spill）、合并（merge）、拉取拷贝（Copy）、合并排序（merge sort）这几个过程，大体流程如下：

Shuffle其实是 MapReduce 处理流程中的一个过程，并不是 MapReduce 的一个组件，这个过程是从Map输出数据，到Reduce接收处理数据之前，横跨Mapper和Reducer两端。整体来看，分为 3个操作：

Partition（分区，必要）
Sort （根据 key 排序，必要）
Combiner （进行局部 value 的合并，非必要）

Shuffle分为Mapper阶段和Reducer阶段，下面就两个阶段做具体分析。

01 Collect （收集）阶段

MapTask 收集我们的 map() 方法输出的 kv 对，放到内存缓冲区Kvbuffer中。每一个 MapTask 都有一个环形内存缓冲区，用于存储任务的输出，默认大小100MB（ mapreduce.task.io.sort.mb 属性）。

kvbuffer :字节数组，数据和数据的索引都会存在该数组中 。
kvmeta：只是kvbuffer中索引存储部分的一个视角。

为什么这么说？
因为索引往往是按整型存储（4个字节），所以使用kvmeta来重新组织该部分的字节
（kvmeta中的一个单元相当于4个字节，但是kvmeta并没有重新开辟内存，其指向的还是kvbuffer）。

每一个 MapTask 都有一个环形 Buffer，Map 将输出写入到这个环形缓冲区中。环形缓冲区是内存中的一种首尾相连的数据结构，专门用来存储 Key-Value 格式的数据，可以叫做Kvbuffer：

环形缓冲区是内存中的一种首尾相连的数据结构，专门用来存储Key-Value格式的数据。

这个数据结构其实就是个字节数组byte[]，叫Kvbuffer。

名如其义，但是这里面不光放置了数据，还放置了一些索引数据，给放置索引数据的区域起了一个Kvmeta的别名。

第一次写入数据的时候，队列头和队列尾都指向0，key-value对按照顺时针存储。当环形缓冲区的数据达到缓冲区大小的80%的时候（即80M），就会溢血到本地磁盘，当再次达到80%时，就会再次溢写到磁盘，直到最后一次，不管环形缓冲区还有多少数据，都会溢写到磁盘。

下图中总共涉及三个变量：kvstart、kvindex和kvend。

kvstart表示当前已写的数据的开始位置，kvindex表示写一个下一个可写的位置，因此，从kvstart到（kvindex-1）这部分数据就是已经写的数据，另外一个线程来Spill的时候，读取的数据就是这一部分。而写线程仍然从kvindex位置开始，并不冲突。

图中涉及三个变量：kvstart、kvindex和kvend。

kvstart表示当前已写的数据的开始位置；
kvindex表示写一个下一个可写的位置，因此，从kvstart到（kvindex-1）这部分数据就是已经写的数据，另外一个线程来Spill的时候，读取的数据就是这一部分。而写线程仍然从kvindex位置开始，并不冲突（如果写得太快而读得太慢，追了一圈后可以通过变量值判断，也无需加锁，只是等待）。

举例来说，下面的第一个图表示按顺时针加入索引，此时kvend=kvstart，但kvindex递增；

当触发Spill的时候，kvend=kvindex，Spill的值涵盖从kvstart到kvend-1区间的数据，kvindex不影响，继续按照进入的数据递增；

当进行完Spill的时候，kvindex增加，kvstart移动到kvend处，在Spill这段时间，kvindex可能已经往前移动了，但并不影响数据的读取。

因此，kvend实际上一般情况下不变，只有在要读取环形缓冲区中的数据时发生一次改变（即设置kvend=kvindex）。

有的同学会问，为什么不直接写入磁盘，要通过环形缓冲区写入磁盘呢？
答案：使用环形缓冲区，便于写入缓冲区和写出缓冲区同时进行。

提问：为什么要在达到缓冲区80%的时候溢出，为什么不等缓冲区满了再溢写到磁盘呢？
答案：会出现阻塞。
解析：如果写满了才溢出到磁盘，那么在溢出磁盘的过程中不能写入，写就被阻塞了，但是如果到了一定程度就溢出磁盘，那么缓冲区就一直有剩余空间可以写，这样就可以设计成读写不冲突，提高吞吐量。

2. sort （排序）阶段

当内存中的数据量达到一定的阈值80%，即80MB可通过 mapreduce.map.sort.spill.percent 配置），一个后台线程就会不断地将数据溢出（spill）到本地磁盘文件中，可能会溢出多个文件。

但溢写之前会有一个 sort 操作，这个 sort 操作先把 Kvbuffer 中的数据按照 partition 值和 key 两个关键字来排序，移动的只是索引数据，排序结果是 Kvmeta 中数据按照 partition（通过调用 Partitioner 的 getPartition() 方法就能知道该输出要送往哪个 Reducer）为单位聚集在一起，同一 partition 内的按照 key 有序（即先分区，然后再对分区中的数据按照key进行字典排序，保证同一个分区中的数据按照key有序）。

如果有 Combiner，还要对排序后的数据进行 Combiner。Combiner 就是一个简单Reducer操作，它在执行 Map 任务的节点本身运行，先对Map 的输出做一次简单Reduce，使得Map的输出更紧凑，更少的数据会被写入磁盘和传送到Reducer。临时文件会在MapTask结束后删除。

说明：写磁盘前，要partition、sort、Combiner。

3. spill（溢写）阶段

说明：写磁盘前，要partition、sort、Combiner。

当排序完成，便开始把数据刷到磁盘，刷磁盘的过程以分区为单位，一个分区写完，写下一个分区，分区内数据有序，最终实际上会多次溢写，然后生成多个溢出文件（每个溢出文件中都是以分区为单位存储）。

4. merge（合并）阶段

每次溢写会在磁盘上生成一个溢写文件，如果 Map 的输出结果很大，有多次这样的溢写发生，磁盘上相应的就会有多个溢写文件存在（例如图中就有2个溢出文件，分别是spill0.out和spill12.out）。

因为最终的文件只有一个，所以需要将这些溢写文件归并到一起，这个过程就叫做 Merge。

Merge 的过程也是相同分区的合并成一个片段（segment），最终所有的 segment 组装成一个最终文件，那么合并过程就完成了。

如下图所示，溢出文件spill0.out和spill12.out红色分区的数据全部放入file.out的红色分区，绿色分区的数据全部放入file.out的绿色分区，蓝色分区的数据全部放入file.out的蓝色分区。这样就合并成了一个最终的文件。

注意：每个MapTask有一个最终文件，并不是所有的MapTask合并成一个最终的文件。

整体Mapper阶段的流程如下：
input->map->buffer->split(partition-sort-combiner)->merge(partition-sort-combiner(file>=3))->数据落地。

至此，Map的操作就已经完成，Reduce端操作即将登场。

1. fetch copy（拉取拷贝）阶段

Map端就处理完了，接下来就是Reduce端了。Reduce端的Shuffle开始工作，而不是Reduce操作开始执行，在Shuffle阶段Reduce不会运行。Map完成后，会通过心跳将信息传给NodeManager，其进而通知ResourceManager，然后ReduceTask开始Shuffle工作。

1. fetch copy（拉取拷贝）阶段

Reduce 端默认有5个数据复制线程从 Map 端复制数据，其通过 Http 方式得到 Map 对应分区的输出文件。每个 MapTask 的完成时间可能不同，因此只要有一个任务完成，ReduceTask 就开始复制（copy）其输出。这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阈值的时候，就会将数据写到磁盘之上。

注意：Reduce端的Shuffle也有一个环形缓冲区，它的大小要比Map端的灵活（由JVM的heapsize设置），由Copy阶段获得的数据，会存放的这个缓冲区中，同样，当到达阈值时会发生溢写到磁盘操作，这个过程中如果设置了Combiner也是会执行的，这个过程会一直执行直到所有的Map输出都被复制过来，如果形成了多个磁盘文件还会进行合并，最后一次合并的结果作为reduce()方法的输入而不是写入到磁盘中。

2. merge sort（合并排序）阶段

在 ReduceTask 远程复制数据的同时，会在后台开启2个线程（一个是内存到磁盘的合并，一个是磁盘到磁盘的合并）对内存中和本地磁盘中的数据文件进行合并操作，以防止内存使用过多或磁盘上文件过多。

内存到磁盘的合并形式：一直在运行（Spill 阶段），直到结束；
磁盘到磁盘的合并形式：生成最终的文件（Merge 阶段）。

在对数据进行合并的同时，会进行排序操作，由于MapTask 阶段已经对数据进行了局部的排序，因此，ReduceTask 只需对所有数据进行一次归并排序即可。合并成大文件后，Reduce端的Shuffle过程也就结束了。

Reduce 阶段

最终合并的文件可能存在于磁盘中，也可能存在于内存中，但是默认情况下是位于磁盘中的。当 Reduce 的输入文件已定，整个 Shuffle 阶段就结束了，然后就是 Reduce 执行，把结果放到 HDFS 中（Reduce 阶段）。

至此整个 Shuffle 过程完成，最后总结几点：

Map 阶段的输出是写入本地磁盘而不是 HDFS，但是一开始数据并不是直接写入磁盘而是缓冲在内存中。
缓存的好处就是减少磁盘 I/O 的开销，提高合并和排序的速度。
内存缓冲区的大小默认是 100M（原则上说，缓冲区越大，磁盘 io 的次数越少，执行速度就越快。缓冲区的大小可以通过 io.sort.mb 参数调整），所以在编写 map 函数的时候要尽量减少内存的使用，为 Shuffle 过程预留更多的内存，因为该过程是最耗时的过程。

MapReduce 中的序列化

序列化概述

序列化（Serialization）：是指把结构化对象（Object）转化为字节流（ByteStream）。
反序列化（Deserialization）：是序列化的逆过程。即把字节流转回结构化对象。

序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储（持久化）和网络传输。
反序列化就是将收到字节序列（或其他数据传输协议）或者是硬盘的持久化数据，转换成内存中的对象。

当要在进程间传递对象或持久化对象的时候，就需要序列化对象成字节流，反之当要将接收到或从磁盘读取的字节流转换为对象，就要进行反序列化。

为什么需要序列化和反序列化？
总的来说可以归结为以下几点：
（1）永久性保存对象，保存对象的字节序列到本地文件或者数据库中；
（2）通过序列化以字节流的形式使对象在网络中进行传递和接收；
（3）通过序列化在进程间传递对象。

使用场景：

所有可在网络上传输的对象都必须是可序列化的。
所有需要保存到磁盘的Java对象都必须是可序列化的。

Java 的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息、header、继承体系等），不便于在网络中高效传输。

所以，Hadoop 自己开发了一套序列化机制（Writable），精简，高效。 Hadoop 中的序列化框架已经对基本类型和 null 提供了序列化的实现了。分别是：

Java 类型	Hadoop Writable 类型
byte	ByteWritable
short	ShortWritable
int	IntWritable
long	LongWritable
float	FloatWritable
double	DoubleWritable
string	Text
null	NullWritable

为什么不使用Java 的序列化？
扩展：Hadoop之父Doug Cutting（道格卡丁）解释道：“因为Java的序列化机制太过复杂了，而我认为需要有一个精简的机制，可以用于精确控制对象的读和写，这个机制将是Hadoop的核心。使用Java序列化虽然可以获得一些控制权，但用起来非常纠结。不用RMI（远程方法调用）也是出于类似的考虑。”

以流量统计项目案例为例：

（1）数据样例

13726238888 2481 24681 
13560436666 1116 954 
13726230503 2481 24681 
13826544101 264 0 
13926435656 132 1512 
13926251106 240 0 
18211575961 1527 2106

完整数据文件为/root/info/data/8/flow.txt。

（2）字段释义

字段中文释义	字段英文释义	数据类型
手机号	phone	String
上行流量	upflow	Long
下行流量	downflow	Long

（3）项目需求一

统计每一个用户（手机号）所耗费的总上行流量、总下行流量、总流量。

总流量=总上行流量+总下行流量

期望输出数据格式：

  手机号    总上行流量   总下行流量    总流量
13480253104  2494800  2494800  4989600

需求解析：其实此需求和之前我们写的WordCount有点类似，也是按照key进行求和，只不过此时的key不是每个单词，而是手机号，此时的value不是出现的次数1，而是手机号对应的 <上行流量，下行流量，上行流量+下行流量>

根据之前的WordCount示例我们可以联想到，我们需要将 <手机号，{上行流量，下行流量，上行流量+下行流量}> 作为Map端的输出，传递给Reduce端，让Reduce端按照key，即手机号做汇总操作。

所以此时的 Map 端的输出value不再是基础类型了，所以我们需要自定义 bean 类来封装流量信息，把 {上行流量，下行流量，上行流量+下行流量} 封装成一个对象。

序列化输出类 FlowBean 上一节课我们已经定义完成，本节课我们主要实现的是，利用 MapReduce 完成按照手机号对 value 进行汇总。

Map端规范

用户编写的程序分成三个部分： Mapper， Reducer， Driver
用户自定义的 Mapper 需要继承父类 Mapper
Mapper 的输入数据是 KV 对的形式（KV 的类型可自定义，示例中默认是<Object, Text>）
Mapper 的输出数据是 KV 对的形式（KV 的类型可自定义）
Mapper 中的业务逻辑写在 map() 方法中
MapTask 进程对每一个 <K,V> 调用一次map()方法

创建 Mapper 类

要想编写代码首先要创建相应的Project（项目），之后再创建Package（包）以及class（类）。

创建项目Project：选择 “File”->“New”->“Project”->“Map/Reduce Project”->“Next”，弹出“New MapReduce Project Wizard”对话框，为“Project name”起个名字，可以任意取名，这里我起名为MyMR。
创建包名Package：在之前创建的MapReduce项目MyMR下创建 Package包com.hongyaa.sum，包名需要全部用小写字母，并且可以有多级，之前用点分隔；
创建类名class：在Package包下创建Class类，这里类名叫做FlowSumMapper.java，在此类下书写以下程序。

类名命名规范：首字母大写，通常由多个单词合成一个类名，要求每个单词的首字母也要大写，使用驼峰式命名。

首先编写 Map 端编程框架，自定义的 FlowSumMapper 需要继承父类 Mapper，输入数据和输出数据都是KV 对的形式。

关键字extends，表示对父类的继承，可以实现父类，也可以调用父类初始化。而且会覆盖父类定义的变量或者函数。
这里我们就需要重写父类中的map()函数，将自定义业务逻辑写在map()函数中。

具体框架代码如下：

public class FlowSumMapper extends Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {  
  
}

KEYIN：是指框架读取到的数据的 key 的类型，在默认的InputFormat下，读到的key是一行文本的起始偏移量，所以key的类型是Long，对应 Hadoop 中的 LongWritable
VALUEIN：是指框架读取到的数据的value的类型，在默认的InputFormat下，读到的value是一行文本的内容，所以value的类型是String，对应 Hadoop 中的 Text

Inputformat: InputFormat()方法是用来生成可供map处理的<key,value>对的。

KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此程序中，我们输出的key是11位的手机号，所以是String，对应 Hadoop 中的 Text
VALUEOUT：用户自定义逻辑方法返回数据中value的类型，由用户业务逻辑决定，在此程序中，我们输出的 value 是我们封装好的流量信息类 FlowBean

Long‐‐‐‐‐‐‐>LongWritable
String‐‐‐‐‐>Text

已知 Mapper 中的业务逻辑写在 map() 方法中，在此 map()方法中我们需要实现的是将读取到的每一行文本拆分成3列，分别是手机号，上行流量和下行流量，对 {上行流量，下行流量} 进行封装，每遇到一个手机号就把其转换成一个 key-value 对，比如手机号 13726238888，就转换成<13726238888,{2481，24681，27162}>发送给 ReduceTask 去汇总。具体代码如下所示：

@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context)
		throws IOException, InterruptedException {
	//（1）将maptask传给我们的每行文本内容先转换成String（Hadoop数据类型转换成java数据类型）
	String line = value.toString(); 
	//（2）根据空格将这一行切分成单词
	String[] splits = line.split("\t");
	//（3）抽取业务所需的字段
	String telephone = splits[0]; // 手机号
	String upFlow = splits[1];  // 上行流量
	String downFlow = splits[2];  // 下行流量
	//（4）获取上行流量和下行流量，对其进行封装（将String类型转化成long类型，因为在FlowBean类中我们定义的upFlow和downFlow都是long类型）
	FlowBean fb = new FlowBean(Long.parseLong(upFlow), Long.parseLong(downFlow));
	//（5）将手机号作为key，将封装的流量信息类作为value
	context.write(new Text(telephone), fb);
}

context对象：

使用context对象进行Map端的输出。context对象就是一个保存所有信息的中介桥梁，与Java中的session有着异曲同工之妙，Context记录map执行的上下文信息。

Map端的详细执行步骤如下：

（1）将文件拆分成按照blocksize进行拆分，默认是128M，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成<key,value>对。这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数（Windows和Linux环境会不同）。

"13726230503	2481	24681" =====>  <0,"13726230503	2481	24681">

（2）将分割好的<key,value>对交给用户定义的map()方法进行处理，生成新的<key,value>对，如下图所示。

<0,"13726230503	2481	24681">=====> <"13726230503",{上行流量	下行流量	上行流量+下行流量}>

使用 value.toString()..split(" ")对每行文本按照分隔符"\t"（源数据单词的分隔符为"\t"）进行分隔。
之后得到一个数组，使用下标将手机号、上行流量和下行流量分别取出来，将每个手机号作为Map端的输出key，将封装好的{上行流量，下行流量}作为该key对应的value。

Map 端完整代码
FlowSumMapper.java 的完整代码如下所示：

	/**
	 * KEYIN:一行文本的起始偏移量
	 * VALUEIN:一行文本的内容
	 * KEYOUT:手机号
	 * VALUEOUT:封装的流量信息
	 * 
	 * <0,"13726230503	2481	24681">
	 *
	 */
public class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context)
			throws IOException, InterruptedException {
		//（1）将maptask传给我们的每行文本内容先转换成String
		String line = value.toString(); 
		//（2）根据空格将这一行切分成单词
		String[] splits = line.split("\t");
		//（3）抽取业务所需的字段
		String telephone = splits[0]; // 手机号
		String upFlow = splits[1];  // 上行流量
		String downFlow = splits[2];  // 下行流量
		//（4）获取上行流量和下行流量，对其进行封装（将String类型转化成long类型，因为在FlowBean类中我们定义的upFlow和downFlow都是long类型）
		FlowBean fb = new FlowBean(Long.parseLong(upFlow), Long.parseLong(downFlow));
		//（5）将手机号作为key，将封装的流量信息类作为value
		context.write(new Text(telephone), fb);
	}
}

	/**
	 * KEYIN:一行文本的起始偏移量
	 * VALUEIN:一行文本的内容
	 * KEYOUT:手机号
	 * VALUEOUT:封装的流量信息
	 * 
	 * <0,"13726230503	2481	24681">
	 *
	 */
public class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context)
			throws IOException, InterruptedException {
		//（1）将maptask传给我们的每行文本内容先转换成String
		String line = value.toString(); 
		//（2）根据空格将这一行切分成单词
		String[] splits = line.split("\t");
		//（3）抽取业务所需的字段
		String telephone = splits[0]; // 手机号
		String upFlow = splits[1];  // 上行流量
		String downFlow = splits[2];  // 下行流量
		//（4）获取上行流量和下行流量，对其进行封装（将String类型转化成long类型，因为在FlowBean类中我们定义的upFlow和downFlow都是long类型）
		FlowBean fb = new FlowBean(Long.parseLong(upFlow), Long.parseLong(downFlow));
		//（5）将手机号作为key，将封装的流量信息类作为value
		context.write(new Text(telephone), fb);
	}
}

Reduce 端规范

用户编写的程序分成三个部分： Mapper， Reducer， Driver
用户自定义的 Reducer 需要继承父类 Reducer
Reducer 的输入数据类型对应 Mapper 的输出数据类型，也是 KV（一定要注意此条）
Reducer 的输出数据是** KV 对**的形式（KV 的类型可自定义）
Reducer 的业务逻辑写在** reduce() 方法**中
ReduceTask 进程对每一组相同 k 的<k,v>组调用一次 reduce() 方法

要想编写代码首先要创建相应的Project（项目），之后再创建Package（包）以及class（类）。

创建项目Project：选择 “File”->“New”->“Project”->“Map/Reduce Project”->“Next”，弹出“New MapReduce Project Wizard”对话框，为“Project name”起个名字，可以任意取名，这里我起名为MyMR。
创建包名Package：在之前创建的MapReduce项目MyMR下创建 Package包com.hongyaa.sum，包名需要全部用小写字母，并且可以有多级，之前用点分隔；
创建类名class：在Package包下创建Class类，这里类名叫做FlowSumReducer.java，在此类下书写以下程序。

类名命名规范：首字母大写，通常由多个单词合成一个类名，要求每个单词的首字母也要大写，使用驼峰式命名。

刚刚我们已经创建了MyMR项目、com.hongyaa.sum包和FlowSumMapper.java类，在FlowSumMapper.java类的同级目录下创建FlowSumReducer.java类。

首先编写 Reduce 端编程框架，自定义的 FlowSumReducer 需要继承父类 Reducer，输入数据和输出数据都是KV 对的形式，其中输入的key-value对数据类型对应Map端输出的key-value对。

关键字extends，表示对父类的继承，可以实现父类，也可以调用父类初始化。而且会覆盖父类定义的变量或者函数。
这里我们就需要重写父类中的map()函数，将自定义业务逻辑写在map()函数中。

具体框架代码如下：

public class FlowSumReducer extends Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {

}

KEYIN：对应 Mapper 端输出的 KEYOUT，即每个手机号，所以是 String，对应 Hadoop 中的 Text
VALUEIN：对应 Mapper 端输出的 VALUEOUT，即封装的流量信息类，所以是 FlowBean
KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此程序中，我们输出的key是手机号，所以是String，对应 Hadoop 中的 Text
VALUEOUT：用户自定义逻辑方法返回数据中value的类型，由用户业务逻辑决定，在此程序中，我们输出的value是累加求和后封装的流量信息类，所以是 FlowBean

已知 Reducer 中的业务逻辑写在 reduce() 方法中，在此 reduce()方法中我们需要接收 MapTask 的输出结果，然后按照 key（手机号）对 value（上行流量、下行流量）做汇总计数。具体代码如下所示：

/*
 * （1）对输入的kv按照k进行分组排序（全局排序）
 * <13726230503,2481	24681	2481+24681><13726230503,2481	24681	2481+24681>。。。。
 * （2）将某一组kv中的第一个kv对中的key传给reduce()方法的变量key，将相同key的value放入迭代器中传给reducde()方法的变量values
 * 
 *<13726230503,list((2481	24681	2481+24681),(2481	24681	2481+24681))>
 *每一组kv对调用一次reduce()方法
 *
 */
@Override
protected void reduce(Text key, Iterable<FlowBean> values,
		Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
	long sumUpFlow = 0L; // 总上行流量
	long sumDownFlow = 0L; // 总下行流量
	
	for (FlowBean fb : values) {
		sumUpFlow += fb.getUpFlow();
		sumDownFlow += fb.getDownFlow();
	}

	// 获取总上行流量和总下行流量，对其进行封装
	FlowBean resultsum = new FlowBean(sumUpFlow, sumDownFlow);
	// 将手机号作为key，将封装的流量信息类作为value，写出最终结果
	context.write(key, resultsum);
}

context对象：

使用context对象进行Reduce端的输出。

Shuffle端（Map输出之后，Reduce接收之前）的详细执行步骤如下：

得到map方法输出的<key,value>对后，在Shuffle阶段（Map阶段的Shuffle）会将它们按照key值进行局部排序。

Reduce端的详细执行步骤如下：

Reducer先对从Mapper接收的数据进行排序（Shuffle中Reduce阶段的排序），再交由用户自定义的reduce方法进行处理，得到新的<key,value>对，并作为最终的输出结果。

<13726230503,2481	24681	2481+24681> <13726230503,2481	24681	2481+24681>
....分组排序后....
<13726230503,list((2481	24681	2481+24681),(2481	24681	2481+24681))>

FlowSumReducer.java 的完整代码如下所示：

	/**
	 * 进入Reduce端之前Maptask会对数据按照key进行局部排序（字典排序）
	 * 
	 * <13726230503,2481	24681	2481+24681>
	 * 。。。。
	 * 。。。
	 * 
	 * Reduce端的输入KEYIN, VALUEIN 对应Map端的输出 KEYOUT, VALUEOUT
	 * KEYOUT:手机号
	 * VALUEOUT：封装的流量信息
	 * 
	 *
	 */
public class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
	/*
	 * （1）对输入的kv按照k进行分组排序（全局排序）
	 * <13726230503,2481	24681	2481+24681><13726230503,2481	24681	2481+24681>。。。。
	 * （2）将某一组kv中的第一个kv对中的key传给reduce()方法的变量key，将相同key的value放入迭代器中传给reducde()方法的变量values
	 * 
	*<13726230503,list((2481	24681	2481+24681),(2481	24681	2481+24681))>
	*每一组kv对调用一次reduce()方法
	*
	*/

	@Override
	protected void reduce(Text key, Iterable<FlowBean> values,
			Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
		long sumUpFlow = 0L; // 总上行流量
		long sumDownFlow = 0L; // 总下行流量
		
		for (FlowBean fb : values) {
			sumUpFlow += fb.getUpFlow();
			sumDownFlow += fb.getDownFlow();
		}

		// 获取总上行流量和总下行流量，对其进行封装
		FlowBean resultsum = new FlowBean(sumUpFlow, sumDownFlow);
		// 将手机号作为key，将封装的流量信息类作为value，写出最终结果
		context.write(key, resultsum);
	}
}

	/**
	 * 进入Reduce端之前Maptask会对数据按照key进行局部排序（字典排序）
	 * 
	 * <13726230503,2481	24681	2481+24681>
	 * 。。。。
	 * 。。。
	 * 
	 * Reduce端的输入KEYIN, VALUEIN 对应Map端的输出 KEYOUT, VALUEOUT
	 * KEYOUT:手机号
	 * VALUEOUT：封装的流量信息
	 * 
	 *
	 */
public class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
	/*
	 * （1）对输入的kv按照k进行分组排序（全局排序）
	 * <13726230503,2481	24681	2481+24681><13726230503,2481	24681	2481+24681>。。。。
	 * （2）将某一组kv中的第一个kv对中的key传给reduce()方法的变量key，将相同key的value放入迭代器中传给reducde()方法的变量values
	 * 
	*<13726230503,list((2481	24681	2481+24681),(2481	24681	2481+24681))>
	*每一组kv对调用一次reduce()方法
	*
	*/

	@Override
	protected void reduce(Text key, Iterable<FlowBean> values,
			Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
		long sumUpFlow = 0L; // 总上行流量
		long sumDownFlow = 0L; // 总下行流量
		
		for (FlowBean fb : values) {
			sumUpFlow += fb.getUpFlow();
			sumDownFlow += fb.getDownFlow();
		}

		// 获取总上行流量和总下行流量，对其进行封装
		FlowBean resultsum = new FlowBean(sumUpFlow, sumDownFlow);
		// 将手机号作为key，将封装的流量信息类作为value，写出最终结果
		context.write(key, resultsum);
	}
}

回顾 MapReduce Driver 端编码规范：整个程序需要一个 Drvier 来进行提交，提交的是一个描述了各种必要信息的 job 对象

接下来进入 Driver 端程序的编写，在 com.hongyaa.sum 包下创建名为 FlowSum.java 的类。

我们已经创建了MyMR项目、com.hongyaa.sum包和FlowSumReducer.java类和FlowSumReducer.java类，即Mapper和Reducer端都已经写好了，就差Driver端了，在同级目录下创建FlowSum.java 类。

Driver 端为该 FlowSum 程序运行的入口，相当于 YARN 集群（分配运算资源）的客户端，需要创建一个 Job 类对象来管理 MapReduce 程序运行时需要的相关运行参数，最后将该 Job 类对象提交给 YARN。

Job对象指定作业执行规范，我们可以用它来控制整个作业的运行。接下来，我们来看一下作业从提交到执行的整个过程。

1. 创建 Job

Job 的创建比较容易，其实就是 new 一个实例，先创建一个配置文件的对象，然后将配置文件对象作为参数，构造一个 Job 对象就可以了。具体代码如下：

// 创建配置文件对象，指定mapreduce程序所需的 HDFS 相关参数
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");  // 只配置了此参数，代表程序在本地运行，不提交到YARN集群，切操作目录为HDFS
// 新建一个 job 任务
Job job = Job.getInstance(conf);

2. 打包作业
我们在 Hadoop 集群上运行这个作业 时，要把代码打包成一个Jar文件，只需要在Job对象的setJarByClass()方法中传递一个类即可，Hadoop会利用这个类来查找包含它的Jar文件，进而找到相关的Jar文件。具体代码如下：

// 将 job 所用到的那些类（class）文件，打成jar包 
job.setJarByClass(FlowSum.class);

3. 设置各个环节的函数
指定我们自定义的 mapper 类和 reducer 类，通过 Job 对象进行设置，将自定义的函数和具体的作业联系起来。具体代码如下：

// 指定 mapper 类和 reducer 类
job.setMapperClass(FlowSumMapper.class);
job.setReducerClass(FlowSumReducer.class);

4. 设置输入输出数据类型
分别指定 MapTask 和 ReduceTask 的输出key-value类型。如果 MapTask 的输出的key-value类型与 ReduceTask 的输出key-value类型一致，则可以只指定ReduceTask 的输出key-value类型。 具体代码如下：

// 指定 MapTask 的输出key-value类型（可以省略）
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(FlowBean.class);

// 指定 ReduceTask 的输出key-value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);

5. 设置输入输出文件目录
在设置输入输出文件目录时，可以选择使用绝对目录，就是直接在语句中写入目录。具体代码如下：

// 指定该 mapreduce 程序数据的输入和输出路径，此处输入、输出为固定文件目录
Path inPath=new Path("/flow/input");
Path outpath=new Path("/flow/output_sum");
FileInputFormat.setInputPaths(job,inPath);
FileOutputFormat.setOutputPath(job, outpath);

6. 提交并运行作业
单个任务的提交可以直接使用如下语句：

job.waitForCompletion(true);

Job运行是通过job.waitForCompletion(true)，true表示将运行进度等信息及时输出给用户，false的话只是等待作业结束。

任务一：流量统计项目案例

项目需求

统计每一个用户（手机号）所耗费的总上行流量、总下行流量、总流量。

期望输出数据格式：

  手机号    总上行流量   总下行流量    总流量
13480253104  2494800  2494800  4989600

需求解析

其实此需求和之前我们写的WordCount有点类似，也是按照key进行求和，只不过此时的key不是每个单词，而是手机号，此时的value不是出现的次数1，而是手机号对应的 <上行流量，下行流量，总流量>

根据之前的WordCount示例我们可以联想到，我们需要将 <手机号，{上行流量，下行流量，总流量}> 作为Map端的输出，传递给Reduce端，让Reduce端按照key，即手机号做汇总操作。

所以此时的 Map 端的输出value不再是基础类型了，所以我们需要自定义 bean 类来封装流量信息，把 {上行流量，下行流量，总流量} 封装成一个对象。

任务二：Map 端程序编写

首先编写 Map 端编程框架，自定义的 FlowSumMapper 需要继承父类 Mapper，输入数据和输出数据都是KV 对的形式。

具体框架代码如下：

public class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {  
  
}

KEYIN：指的是一行文本的起始偏移量，所以key的类型是 LongWritable
VALUEIN：读到的value是一行文本的内容，所以value的类型是 Hadoop 中的 Text
KEYOUT：在此程序中，我们输出的key是11位的手机号，所以类型是 Text
VALUEOUT：在此程序中，我们输出的 value 是我们封装好的流量信息类 FlowBean

map()方法的具体实现

String[] splits = value.toString().split("\t");

将maptask传给我们的文本内容value先转换成String（读入的value是Text类型，将Hadoop Text类型转换为Java String类型），之后再按"\t"进行切分。

FlowBean fb = new FlowBean(Long.parseLong(upFlow), Long.parseLong(downFlow));  // 封装流量信息
context.write(new Text(telephone), fb);

使用下标将手机号、上行流量和下行流量分别取出来，将每个手机号作为Map端的输出key，将封装好的{上行流量，下行流量}作为该key对应的value。

任务三：Reduce 端程序编写

具体框架代码如下：

public class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

}

KEYIN：对应 Mapper 端输出的 KEYOUT，即每个手机号，所以类型是 Text
VALUEIN：对应 Mapper 端输出的 VALUEOUT，即封装的流量信息类，所以是 FlowBean
KEYOUT：在此程序中，我们输出的key是手机号，所以类型是 Text
VALUEOUT：在此程序中，我们输出的value是累加求和后封装的流量信息类，所以是 FlowBean

reduce()方法的具体实现

for (FlowBean fb : values) {
	sumUpFlow += fb.getUpFlow();
	sumDownFlow += fb.getDownFlow();
}

遍历迭代对象values，将每个手机号对应的上行流量和下行流量全部读出，然后累加，就可以得出每个手机号对应的总上行流量和总下行流量了。

FlowBean resultsum = new FlowBean(sumUpFlow, sumDownFlow);   // 封装流量信息
context.write(key, resultsum);

将每个手机号作为key，将每个手机号对应的封装好的流量信息{总上行流量，总下行流量，总流量}作为value，使用context对象的write()方法写入context中，作为我们的最终输出结果。

任务四：Driver 端程序编写

1. 创建 Job

Job 的创建比较容易，其实就是 new 一个实例，先创建一个配置文件的对象，然后将配置文件对象作为参数，构造一个 Job 对象就可以了。

2. 打包作业

我们在 Hadoop 集群上运行这个作业时，要把代码打包成一个Jar文件，只需要在Job对象的setJarByClass()方法中传递一个类即可，Hadoop会利用这个类来查找包含它的Jar文件，进而找到相关的Jar文件。

3. 设置各个环节的函数

指定我们自定义的 mapper 类和 reducer 类，通过 Job 对象进行设置，将自定义的函数和具体的作业联系起来。

4. 设置输入输出数据类型

分别指定 MapTask 和 ReduceTask 的输出key-value类型。如果 MapTask 的输出的key-value类型与 ReduceTask 的输出key-value类型一致，则可以只指定ReduceTask 的输出key-value类型。

5. 设置输入输出文件目录

在设置输入输出文件目录时，可以选择使用绝对目录，就是直接在语句中写入目录；也可以使用参数输入，即在运行程序时，再在控制台输入目录。

6. 提交并运行作业

单个任务的提交可以直接使用job.waitForCompletion(true)语句，其中，参数true表示将运行进度等信息及时输出给用户，false的话只是等待作业结束。

流量统计项目案例介绍

以流量统计项目案例为例：

（1）数据样例

13726238888 2481 24681 
13560436666 1116 954 
13726230503 2481 24681 
13826544101 264 0 
13926435656 132 1512 
13926251106 240 0 
18211575961 1527 2106

完整数据文件为/root/info/data/8/flow.txt。

（2）字段释义

字段中文释义	字段英文释义	数据类型
手机号	phone	String
上行流量	upflow	Long
下行流量	downflow	Long

（3）项目需求二

得出上题结果的基础之上再加一个需求：将统计结果按照总流量倒序排序。

期望输出数据格式：

13502468823	101663100	1529437140	1631100240
13925057413	153263880	668647980	821911860
13726238888	34386660	342078660	376465320
...

（4）项目解析

基本思路：实现自定义的 bean 来封装流量信息，并将 bean 作为 Map 输出的 key 来传输。

MapReduce 程序在处理数据的过程中会对数据排序（Map 输出的 kv 对传输到 Reduce 之前，会排序），排序的依据是 Map 输出的 key，所以，我们如果要实现自己需要的排序规则，则可以考虑将排序因素放到 key 中。

排序是 MapReduce 框架中最重要的操作之一。 MapTask 和 ReduceTak 均会对数据按照 key 进行排序。该操作属于 Hadoop 的默认行为，任何应用程序中的数据均会被排序，而不管逻辑上是否需要。

默认排序是按照字典顺序排序。

字典排序是一种对于随机变量形成序列的排序方法。
其方法是按照字母排列顺序，或数字顺序由小到大形成的的序列。

shuffle过程：
对于MapTask，它会将处理的结果暂时放到环形缓冲区（默认大小100M）中，当环形缓冲区使用率达到一定阈值（环形缓冲区的80%，即80M）后，再对缓冲区中的数据进行一次排序（先分区再排序，保证同一 partition 内的按照 key 有序，默认是一个reduce，也就是一个分区），并将这些有序数据溢写到磁盘上。

而当数据处理完毕后，它会对磁盘上所有文件进行归并排序（因为在溢写的时候会生成多个溢出文件，需要把所有溢出的临时文件进行一次合并操作，以确保一个MapTask最终只产生一个中间数据文件，合并的过程中，也要调用 partitioner 进行分区和针对 key 进行排序。）。

对于 RecduceTak，它从每个 MapTask 上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask 统一对内存和磁盘上的所有数据进行一次归并排序。

（1）部分排序
MapReduce 根据输入记录的键对数据集排序，保证输出的每个文件内部有序。

（2）全排序
最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。
但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了 MapReduce 所提供的并行架构。
替代方案：首先创建一系列排好序的文件；其次，串联这些文件；最后，生成一个全局排序的文件。主要思路是使用一个分区来描述输出的全局排序。例如：可以为输出文件创建3个分区，在第一分区中，记录的单词首字母a-g，第二分区记录单词首字母h-n，第三分区记录单词首字母o-z。

我们知道在MapReduce中默认只有一个分区，且默认的分区规则是：根据key的hashcode%reducetask数来分发。

要想实现多个分区，我们需要在Driver端指定多个reducetask，另外还需要自定义分区类，改变分区规则。这个我们在下一节再讲述。

（3）辅助排序：（GroupingComparator分组）
在Reduce端对key进行分组（此过程在reduce()方法之前完成）。
应用于：在接收 key 为 bean 对象时，想让一个或几个字段相同（全部字段比较不相同）的key进入到同一个reduce()方法时，可以采用分组排序。

之后我们也会学习自定义分组，改变MapReduce的默认分组的规则。

（4）二次排序
在自定义排序过程中，如果 compareTo() 方法中的判断条件为两个即为二次排序（比如我们按照班级升序排序，同班级的学生再按照学号升序排序）。

WritableComparable 继承自 Writable和 java.lang.Comparable 接口，是一个Writable也是一个Comparable，也就是说，既可以序列化，也可以比较！

Comparable可以认为是一个内比较器，实现了Comparable接口的类有一个特点，就是这些类是可以和自己比较的，至于具体和另一个实现了Comparable接口的类如何比较，则依赖compareTo()方法的实现。compareTo()方法的返回值是int，有三种情况：
（1）比较者大于被比较者（也就是compareTo()方法里面的对象），那么返回正整数；
（2）比较者等于被比较者，那么返回0；
（3）比较者小于被比较者，那么返回负整数。

Writable和WritableComparable区别？
WritableComparable 就是比Writable多了一个compareTo()方法，用来判断key是否唯一或者说是不是相同。

定义一个FlowBeanSort 类（Java标准类），实现 WritableComparable 接口。

1. Java标准类要求是什么？

所有的成员变量都要使用private关键字修饰：示例中我们总共定义了3个私有的成员变量，分别是upFlow，downFlow 和 sumFlow。
为每一个成员变量编写一对Getter/Setter方法：示例中我们为3个成员变量都编写了一对Getter/Setter方法。
编写一个无参数的构造方法：示例中编写了FlowBean()无参数的构造方法。因为反序列化时，需要反射调用无参构造方法，所以必须定义无参构造方法。
编写一个有参构造方法：示例中编写了FlowBean(long upFlow, long downFlow) 有参构造方法，其中sumFlow=upFlow+downFlow。

这样标准的类也叫做Java Bean。

另外，我们还可以重写toString()方法，让成员变量用”\t”分开。

2. 重写序列化方法和反序列化方法
需要重写Writable接口中的序列化方法和反序列化方法。

序列化方法 write(DataOutput out)：将数据写入到二进制数据流中。
反序列化方法 readFields(DataInput in)：从二进制数据流中读取数据。

特别需要注意的是：字段的反序列化顺序与序列化时的顺序保持一致，并且参数类型和个数也一致。

3. 重写compareTo()方法
实现Comparable接口的类如何比较，依赖compareTo()方法。
我们的需求是：将统计结果按照总流量倒序排序。 所以我们使用的是sumFlow。

int compareTo(T o)的返回值是int类型，主要分为三大类：负整数、零或正整数，根据此对象是小于、等于还是大于指定对象。

也就是说当语句return this.sumFlow > o.getSumFlow() ? 1 : -1;的返回值为1时，也就是说 this的值大于o的值时，compareTo是按照升序（由小到大）排序的！
当返回值为-1时，也就是说 this 的值小于o的值时，compareTo()是按照降序（由大到小）排序的！
当返回值为0时，this的值等于o的值。

所以要是按照倒序（降序排列）可以写成是this.sumFlow > o.getSumFlow() ? -1 : 1;或者o.getSumFlow()>this.sumFlow?1:-1;

Ps：（快速记忆法）当前对象与后一个对象进行比较，如果比较结果为1进行交换，其他不进行交换。 this就是当前对象，另一个就是后一个对象。

当前对象比后一个对象大，返回值为1时，则前后交换，说明为升序。
当前对象比后一个对象小，返回值为1时，则前后交换，说明为降序。

要完成上述的需求，还需要完成三个程序分别是一个 Mapper 类、一个 Reducer 类和一个用于连接整个过程的驱动 Driver 主程序。首先我们来看一下Mapper类。

分析：以需求一的输出结果作为排序的输入数据，自定义FlowBeanSort，以 FlowBeanSort 为 Map 输出的 key，以手机号作为 Map 输出的 value，因为 MapReduce 程序会对 Map 阶段输出的key 进行排序。

已知 Mapper 中的业务逻辑写在 map() 方法中，在此 map()方法中我们需要实现的是将读取到的每一行文本拆分成4列，分别是手机号，总上行流量、总下行流量和总流量，对 {总上行流量，总下行流量，总流量} 进行封装，并实现按照总流量进行降序排序，将Map的输出端修改为<FlowBeanSort ,手机号>发送给 ReduceTask 。具体代码如下所示：

// 输入数据是上一个统计程序的输出结果，已经是各个手机号的总流量信息
public class FlowSumSortMapper extends Mapper<LongWritable, Text, FlowBeanSort, Text> {

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, FlowBeanSort, Text>.Context context)
			throws IOException, InterruptedException {
		// （1）获取一行文本的内容，并将其转换为String类型，之后按照分隔符“\t”进行切分
		String[] splits = value.toString().split("\t");
		// （2）取出手机号
		String telephone = splits[0];
		// （3）封装对象
		FlowBeanSort fbs = new FlowBeanSort();
		fbs.setUpFlow(Long.parseLong(splits[1]));
		fbs.setDownFlow(Long.parseLong(splits[2]));
		fbs.setSumFlow(Long.parseLong(splits[3]));
		// （4）将封装的fbs对象作为key，将手机号作为value，分发给Reduce端
		context.write(fbs, new Text(telephone));
	}
}

KEYIN：是指框架读取到的数据的key的类型，在默认的InputFormat下，读到的key是一行文本的起始偏移量，所以key的类型是Long，对应 Hadoop 中的 LongWritable
VALUEIN：是指框架读取到的数据的value的类型，在默认的InputFormat下，读到的value是一行文本的内容，所以value的类型是String，对应 Hadoop 中的 Text
KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此程序中，我们输出的key是封装并实现了自定义排序的流量信息类 FlowBeanSort
VALUEOUT：用户自定义逻辑方法返回数据中value的类型，由用户业务逻辑决定，在此程序中，我们输出的 value 是手机号，所以是String，对应 Hadoop 中的 Text

分析：这里的 reduce()方法只需要实现将 Map 端输出的 key-value 调换后输出即可，即修改为<手机号，FlowBeanSort>的形式。具体代码如下所示：

public class FlowSumSortReducer extends Reducer<FlowBeanSort, Text, Text, FlowBeanSort> {
	/*
	 * <FlowBeanSort,电话号> ===> <电话号,FlowBeanSort>
	 */
	@Override
	protected void reduce(FlowBeanSort key, Iterable<Text> values,
			Reducer<FlowBeanSort, Text, Text, FlowBeanSort>.Context context) throws IOException, InterruptedException {
		// 遍历集合
		for (Text tele : values) {
			// 将手机号作为key，将封装好的流量信息作为value，作为最终的输出结果
			context.write(new Text(tele), key);
		}
	}
}

KEYIN：对应 Mapper 端输出的 KEYOUT，即封装的流量信息类 FlowBeanSort
VALUEIN：对应 Mapper 端输出的 VALUEOUT，即手机号，所以是 String，对应 Hadoop 中的 Text
KEYOUT：用户自定义逻辑方法返回数据中key的类型，由用户业务逻辑决定，在此程序中，我们输出的key是手机号，所以是String，对应 Hadoop 中的 Text
VALUEOUT：用户自定义逻辑方法返回数据中value的类型，由用户业务逻辑决定，在此程序中，我们输出的value是封装好的实现了自定义排序的流量信息类，所以是 FlowBeanSort

Driver 端为该 FlowSumSortDemo 程序运行的入口，相当于 YARN 集群（分配运算资源）的客户端，需要创建一个 Job 类对象来管理 MapReduce 程序运行时需要的相关运行参数，最后将该 Job 类对象提交给 YARN。

Job对象指定作业执行规范，我们可以用它来控制整个作业的运行。接下来，我们来看一下作业从提交到执行的整个过程。

1. 创建 Job

// 创建配置文件对象，指定mapreduce程序所需的 HDFS 相关参数
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");  // 只配置了此参数，代表程序在本地运行，不提交到YARN集群，切操作目录为HDFS
// 新建一个 job 任务
Job job = Job.getInstance(conf);

// 将 job 所用到的那些类（class）文件，打成jar包 
job.setJarByClass(FlowSumSortDemo.class);

3. 设置各个环节的函数
指定我们自定义的 mapper 类和 reducer 类，通过 Job 对象进行设置，将自定义的函数和具体的作业联系起来。具体代码如下：

// 指定 mapper 类和 reducer 类
job.setMapperClass(FlowSumSortMapper.class);
job.setReducerClass(FlowSumSortReducer.class);

// 指定 MapTask 的输出key-value类型（可以省略）
job.setMapOutputKeyClass(FlowBeanSort.class);
job.setMapOutputValueClass(Text.class);

// 指定 ReduceTask 的输出key-value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBeanSort.class);

5. 设置输入输出文件目录
在设置输入输出文件目录时，可以选择使用绝对目录，就是直接在语句中写入目录。具体代码如下：

// 指定该 mapreduce 程序数据的输入和输出路径，此处输入、输出为固定文件目录
Path inPath=new Path("/flow/output_sum");
Path outpath=new Path("/flow/output_sort");
FileInputFormat.setInputPaths(job,inPath);
FileOutputFormat.setOutputPath(job, outpath);

6. 提交并运行作业
单个任务的提交可以直接使用如下语句：

job.waitForCompletion(true);

Job运行是通过job.waitForCompletion(true)，true表示将运行进度等信息及时输出给用户，false的话只是等待作业结束。

执行结果如下：

13502468823	101663100	1529437140	1631100240
13925057413	153263880	668647980	821911860
13726238888	34386660	342078660	376465320
13726230503	34386660	342078660	376465320
18320173382	132099660	33430320	165529980
13560439658	28191240	81663120	109854360
13660577991	96465600	9563400	106029000
15013685858	50713740	49036680	99750420
13922314466	41690880	51559200	93250080
15920133257	43742160	40692960	84435120
84138413	57047760	19847520	76895280
13602846565	26860680	40332600	67193280
18211575961	21164220	29189160	50353380
15989002119	26860680	2494800	29355480
13560436666	15467760	13222440	28690200
13926435656	1829520	20956320	22785840
13480253104	2494800	2494800	4989600
13826544101	3659040	0	3659040
13926251106	3326400	0	3326400
13760778710	1663200	1663200	3326400
13719199419	3326400	0	3326400

最后一列是总流量，从执行结果中可以看出，是按照总流量进行的降序排列。

以流量统计项目案例为例：

（1）数据样例

13726238888 2481 24681 
13560436666 1116 954 
13726230503 2481 24681 
13826544101 264 0 
13926435656 132 1512 
13926251106 240 0 
18211575961 1527 2106

完整数据文件为/root/info/data/8/flow.txt。

（2）字段释义

字段中文释义	字段英文释义	数据类型
手机号	phone	String
上行流量	upflow	Long
下行流量	downflow	Long

（3）项目需求

在需求二的基础上再添加一个需求：将流量汇总统计结果按照手机归属地的不同（手机号前3位）输出到不同文件中。
（需求二是将统计结果按照总流量倒序排序。我们可以在需求二代码的基础上添加一个自定义分区类，Mapper和Reducer的代码完全不用修改，直接复用即可。）

134		开头的手机号放到 0 号分区文件
135		开头的手机号放到 1 号分区文件
136		开头的手机号放到 2 号分区文件
137		开头的手机号放到 3 号分区文件
138		开头的手机号放到 4 号分区文件
139		开头的手机号放到 5 号分区文件
其它		开头的手机号放到 6 号分区文件
...

在 MapReduce 中默认的 ReduceTask 的个数为1，所以我们之前程序的只能生成1个结果文件。现在的话我们需要按照手机号的前3位进行分区，放到7个不同的结果文件。

（4）项目解析

从需求中可以看出，我们需要把最终的输出结果分发到7个不同的文件中。我们知道，最终的输出结果是来自于 ReduceTask，那么，如果要得到7个文件，意味着需要有同样数量的 ReduceTask 在运行。

ReduceTask 的数据来自于 MapTask，也就说 MapTask 要划分数据，对于不同的数据分配给不同的 ReduceTask 运行。MapTask 划分数据的过程就称作 Partition，负责实现划分数据的类称作 Partitioner。

所以，我们要想按照手机号的前3位进行分区，我们就需要创建一个类继承 Partitioner，然后重写自定义分区方法getPartition()。

在 Hadoop 的 MapReduce 过程中，每个 MapTask 处理完数据后，如果存在自定义的 Combiner 类，会先进行一次本地的 Reduce 操作，然后把数据发送到 Partitioner（负责实现划分数据的类），由 Partitioner 来决定每条记录应该送往哪个 Reduce 节点，默认使用的是 HashPartitioner，其核心代码如下：

public class HashPartitioner<K, V> extends Partitioner<K, V> {

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }
}

HashPartitioner 是专门用来处理Mapper任务输出的，getPartition()方法有三个形参，源码中key、value分别指的是Mapper任务的输出，numReduceTasks 指的是设置的 ReduceTask 数量，默认值是1。

key.hashCode()：是对 Map 输出的 key 取 hashCode 值
&：是Java中的位运算符，在数据的二进制层面上按位与的意思（只有对应的两个二进位均为1时，结果位才为1 ，否则为0。）
Integer.MAX_VALUE：int 类型的最大值，最高位为0，符号位，表示是正数，任何数和0进行运算，都得0，都是正数。
%：求余运算

综合而言，(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks 是要保证任何Map输出的key在与numReduceTasks取模后决定的分区为正整数。

当numReduceTasks为1的时候，任何整数与1相除的余数肯定是0，也就是getPartition()方法的返回值总是0。也就是 MapTask 的输出总是送给一个ReduceTask，最终只能输出到一个文件中。

getPartition()函数的作用：
（1）获取 key 的哈希值
（2）默认的分发规则为：根据 key 的 hashcode%reducetask 数来分发
（3）这样做的目的是可以把<key,value>对均匀地分发到各个对应编号的 ReduceTask 节点上，达到 ReduceTask 节点的负载均衡。

自定义分区

据此分析，如果想要最终输出到多个文件中，在 MapTask 中对数据应该划分到多个区中。那么，我们只需要按照一定的规则让getPartition()方法的返回值是0,1,2,3…即可。

大部分情况下，我们都会使用默认的分区函数，但有时我们又有一些特殊的需求，而需要定制Partition来完成我们的业务，就像是此流量统计项目案例中要求的：将流量汇总统计结果按照手机归属地的不同（手机号前3位）输出到不同文件中。

134		开头的手机号放到 0 号分区文件
135		开头的手机号放到 1 号分区文件
136		开头的手机号放到 2 号分区文件
137		开头的手机号放到 3 号分区文件
138		开头的手机号放到 4 号分区文件
139		开头的手机号放到 5 号分区文件
其它		开头的手机号放到 6 号分区文件

这时候，我们使用的默认分区函数就不行了，所以需要我们定制自己的Partition，首先分析下，我们需要7个分区输出，所以在设置Reduce的个数时，一定要设置为7，其次在自定义的 Partition 里，进行分区时，要根据手机号的前3位进行具体的分区，而不是根据 key 的hash码来分区。

自定义分区很简单，第一步是创建一个类，继承抽象类 Partitioner，KEY, VALUE 对应 Map 端的输出数据的 key-value 类型。

在我们上节课中，我们书写的FlowSumSortMapper.java，Map的输出key是封装的FlowBeanSort对象{总上行流量，总下行流量，总流量}，Map的输出value是手机号。所以此处的key-value类型分别是FlowBeanSort和Text。

getPartition()方法的三个参数：

key：对应 MapTask 的输出 key，所以是封装并实现了自定义排序的流量信息类 FlowBeanSort
value：对应 MapTask 的输出 value，所以是手机号
numPartitions：指的是设置的 ReduceTask 的数量，默认值是1。

因为value是代表的手机号，所以我们需要将value的前3位取出来，使用value.toString().substring(0, 3);

首先使用toString()方法将Hadoop的数据类型Text转换成Java的数据类型String；
之后使用substring(起始索引，结束索引)：截取字符串，包括起始索引，不包括结束索引，索引默认从0开始。

之后使用if...else条件判断语句，将手机号前3位和分区号进行对应。

最后将分区号使用return进行返回。

自定义分区

（2）另外，还要在 Driver 端给任务设置需要执行的分区类：

job.setPartitionerClass(TelePartitioner.class);

（3）自定义分区后，要根据自定义的分区逻辑设置相应数量的ReduceTask：

job.setNumReduceTasks(7);

需要注意的是：

如果 ReduceTask 的数量 > getPartition 的结果数，则会多产生几个空的输出文件 part-r-000xx；
如果1 < ReduceTask 的数量 < getPartition 的结果数，则有一部分分区数据无处安放，会 Exception；
如果 ReduceTask 的数量 = 1，则不管 MapTask 端输出多少个分区文件，最终结果都交给这一个 ReduceTask，最终也就只会产生一个结果文件 part-r-00000。

例如：假设自定义分区数为 7，则：
（1）job.setNumReduceTasks(1); # 会正常运行，只不过会产生一个输出文件part-r-00000
（2）job.setNumReduceTasks(2); # 会报错，因为2,3,4,5,6好分区的数据无处存放
（3）job.setNumReduceTasks(8); # 大于7，程序会正常运行，会产生1个空文件（part-r-00007）

自定义分区后，需要在 Driver 端给任务设置需要执行的分区类：

job.setPartitionerClass(TelePartitioner.class)

另外，根据自定义的分区逻辑设置相应数量的ReduceTask：

job.setNumReduceTasks(7);

注意：
最好是有几个分区就设置几个 ReduceTask。

在 Driver 端执行 MapReduce 程序后，会在/flow/output_partition目录下生成7个 part-r-00000(0-6) 文件，依次对应 134,135,136,137,138,139 以及其它的分区：

可以在 eclipse 中或者使用 cat命令在终端查看结果文件的内容，发现成功按照手机号的前3位进行了分区，且每个分区文件中按照总流量进行了降序排序：

任务一：流量统计项目案例

在自定义排序的基础上再添加一个需求：将流量汇总统计结果按照手机归属地的不同（手机号前3位）输出到不同文件中。（实现自定义分区）

从需求中可以看出，我们需要把最终的输出结果分发到不同的文件中。我们知道，最终的输出结果是来自于 ReduceTask，那么，如果要得到多个结果文件，意味着需要有同样数量的 ReduceTask 在运行。

ReduceTask 的数据来自于 MapTask，也就说 MapTask 要划分数据，对于不同的数据分配给不同的 ReduceTask 运行。MapTask 划分数据的过程就称作 Partition，负责实现划分数据的类称作 Partitioner。

所以，我们要想按照手机号的前3位进行分区，我们就需要创建一个类继承 Partitioner，然后重写自定义分区方法getPartition()。

任务二：Partitioner 分区

2.1 MapReduce 默认分区介绍

HashPartitioner：专门用来处理Mapper任务输出的，MapReduce默认的分区器。

getPartition()方法：有三个形参，源码中key、value分别指的是Mapper任务的输出，numReduceTasks 指的是设置的 ReduceTask 数量，默认值是1。

getPartition()函数的作用：
（1）获取 key 的哈希值
（2）默认的分发规则为：根据 key 的 hashcode%reducetask 数来分发
（3）这样做的目的是可以把<key,value>对均匀地分发到各个对应编号的 ReduceTask 节点上，达到 ReduceTask 节点的负载均衡。

2.2 自定义分区

（1）创建一个类继承抽象类 Partitioner，然后重写getPartition()方法。

（2）要在 Driver 端给任务设置需要执行的分区类，使用的命令是：
job.setPartitionerClass(xxx.class);
（3）在 Driver 端设置相应数量的ReduceTask，使用的命令是：
job.setNumReduceTasks(xxx);
注意：
最好是有几个分区就设置几个 ReduceTask。