Hadoop大数据技术教程（ wukong-1.0v）_创建crt工具和服务器hadoop01的快速连接

1 初识Hadoop

什么是大数据

随着近几年计算机技术和互联网的发展，“大数据”这个词被提及的越来越频繁。与此同时，大数据的快速发展也在无时无刻影响着我们的生活。例如，医疗方面，大数据能够帮助医生预测疾病；电商方面，大数据能够向顾客个性化推荐商品；交通方面，大数据会帮助人们选择最佳出行方案。

Hadoop作为一个能够对大量数据进行分布式处理的软件框架，用户可以利用Hadoop生态体系开发和处理海量数据。由于Hadoop有可靠及高效的处理性能，使得它逐渐成为分析大数据的领先平台。

高速发展的信息时代，新一轮科技革命和变革正在加速推进，技术创新日益成为重塑经济发展模式和促进经济增长的重要驱动力量，而“大数据”无疑是核心推动力。

那么，什么是“大数据”呢？如果从字面意思来看，大数据指的是巨量数据。那么可能有人会问，多大量级的数据才叫大数据？不同的机构或学者有不同的理解，难以有一个非常定量的定义，只能说，大数据的计量单位已经越过TB级别发展到PB、EB、ZB、YB甚至BB来衡量。

最早提出“大数据”这一概念的是全球知名咨询公司麦肯锡，他是这样定义大数据的：一种规模大到在获取、存储、管理、分析方面大大超出了传统数据库软件工具能力范围的数据集合，具有海量的数据规模、快速的数据流转、多样的数据类型以及价值密度四大特征。

研究机构Gartner是这样定义大数据的：“大数据”是需要新处理模式才能具有更强的决策力、洞察发现力和流转优化能力来适应海量、高增长率和多样化的信息资产。

大数据的特征

大数据主要具有以下四个方面的典型特征，即大量（Volume）、多样（Varity）、高速（Velocity）和价值（Value），即所谓的“4V”。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-D5B5aGHS-1600740996729)(hadoop.assets/image-20200901120118394.png)]

研究大数据的意义

现在的社会是一个高速发展的社会，科技发达，信息流通，人们之间的交流也越来越密切，生活也越来越便捷，然而大数据就是这个高科技时代的产物。阿里巴巴的创办人马云曾经说过，未来的时代将不是IT时代，而是DT的时代，DT就是Data Technology数据科技，这显示出大数据对于阿里巴巴集团来说是举足轻重的。

有人把数据比喻为蕴藏能量的煤矿。煤炭按照性质有焦煤、无烟煤、肥煤、贫煤等分类，而露天煤矿、深山煤矿的挖掘成本又不一样。与此类似，大数据并不在于“大”，而在于“有用”。数据的价值含量、挖掘成本比数量更为重要。对于很多行业而言，如何利用这些大规模数据，发掘其潜在价值，才是赢得核心竞争力的关键。

研究大数据，最重要的意义是预测。因为数据从根本上讲，是对过去和现在的归纳和总结，其本身不具备趋势和方向性的特征，但是我们可以应用大数据去了解事物发展的客观规律、了解人类行为，并且能够帮助我们改变过去的思维方式，建立新的数据思维模型，从而对未来进行预测和推测。知名互联网公司谷歌对其用户每天频繁搜索的词汇进行数据挖掘，从而进行相关的广告推广和商业研究。

大数据的应用场景

医疗行业的应用

研究大数据，最重要的意义是预测。因为数据从根本上讲，是对过去和现在的归纳和总结，其本身不具备趋势和方向性的特征，但是我们可以应用大数据去了解事物发展的客观规律、了解人类行为，并且能够帮助我们改变过去的思维方式，建立新的数据思维模型，从而对未来进行预测和推测。知名互联网公司谷歌对其用户每天频繁搜索的词汇进行数据挖掘，从而进行相关的广告推广和商业研究。

大数据让就医、看病更简单。随着大数据在医疗行业的深度融合，大数据平台积累了海量的病例、病例报告、治愈方案、药物报告等信息资源，所有常见的病例、既往病例等都记录在案，医生通过有效、连续的诊疗记录，给病人提供优质、合理的诊疗方案。

1. 优化医疗方案，提供最佳治疗方法

借助于大数据平台，可以搜集不同病人的疾病特征、病例和治疗方案，从而建立医疗行业的病人分类数据库。在医生诊断病人时可以参考病人的疾病特征、化验报告和检测报告，参考疾病数据库来快速帮助病人确诊。在制定治疗方案时，医生可以依据病人的基因特点，调取相似基因、年龄、人种、身体情况相同的有效治疗方案，制定出适合病人的治疗方案，帮助更多人及时进行治疗。

2. 有效预防预测疾病

解决患者的疾病，最为简单的方式就是防患于未然。通过大数据对于群众的人体数据监控，将各自的健康数据、生命体征指标都集合在数据库和健康档案中。群众需要定期去做检查，及时更新数据，以便于通过大数据来预防和预测疾病的发生，做到早治疗、早康复。

金融行业的应用

1. 精准营销

互联网时代的银行在互联网的冲击下，迫切的需要掌握更多用户信息，继而构建用户360度立体画像，即可对细分的客户进行精准营销、实时营销等个性化智慧营销。

2. 风险管控

应用大数据平台，可以统一管理金融企业内部多源异构数据和外部征信数据，更好的完善风控体系。内部可保证数据的完整性与安全性，外部可控制用户风险。

3. 决策支持

通过大数据分析方法改善经营决策，为管理层提供可靠的数据支撑，从而使经营决策更高效、敏捷、精准 。

4. 服务创新

通过对大数据的应用，改善与客户之间的交互、增加用户粘性，为个人与政府提供增值服务，不断增强金融企业业务核心竞争力。

5. 产品创新

通过高端数据分析和综合化数据分享，有效对接银行、保险、信托、基金等金融产品，使金融企业能够从其他领域借鉴并创造新的金融产品。

零售行业的应用

在美国零售业上有这样一个传奇故事，某家商店将纸尿裤和啤酒并排放在一起销售，结果纸尿裤和啤酒的销量双双增长！为什么看起来风马牛不相及的两种商品搭配在一起，能取到如此惊人的效果呢？后来经过分析发现这些购买者多数是已婚男士，这些男士在为小孩买纸尿裤的同时，也会为自己买一些啤酒。发现这个秘密后，沃尔玛超市就将啤酒摆放在尿不湿旁边，顾客购买会更方便，销量自然也会大幅上升。

1. 精准定位零售行业市场

企业进入或开拓某一区域零售行业市场，首先要进行项目评估和可行性分析，只有通过项目评估和可行性分析才能决定是否适合进入或者开拓这块市场。通常分析这个区域流动人口、消费水平、客户的消费习惯、市场对产品的认知度以及当前的市场供需情况等等，这些问题的背后包含的海量信息构成了零售行业市场调研的大数据，对这些大数据的分析就是零售行业市场精准定位的过程。

2. 支撑行业收益管理

大数据时代的来临，为企业收益管理工作的开展提供了广阔的空间。需求预测、细分市场和敏感度分析对数据需求量很大，而传统的数据分析大多是对企业自身的历史数据进行预测和分析，容易忽视整个零售行业信息，因此预测结果难免会存在偏差。企业在实施收益管理过程中，如果在自有的数据基础上，依靠自动化信息采集软件来收集更多的零售行业数据，以此来了解更多的市场信息，这将会对制订准确的收益策略，取得更高收益起到推进作用。

3. 挖掘零售行业新需求

作为零售行业企业，如果能收集网上零售行业的评论数据，建立网评大数据库，然后再利用分词、聚类、情感分析消费者的消费行为、价值取向、评论中体现的新消费需求和企业产品质量问题，以此来改进和创新产品，量化产品价值，制订合理的价格及提高服务质量，从中获取更大的收益。

Hadoop的发展历史

Hadoop的优势

Hadoop的生态系统

随着Hadoop的不断发展，Hadoop生态体系越来越完善，现如今已经发展成一个庞大的生态体系。

HDFS分布式文件系统

HDFS是Hadoop的分布式文件系统，它是Hadoop生态系统中的核心项目之一，是分布式计算中数据存储管理基础。

MapReduce分布式计算框架

MapReduce是一种计算模型，用于大规模数据集（大于1TB）的并行运算。

Yarn资源管理框架

Yarn（Yet Another Resource Negotiator）是Hadoop 2.0中的资源管理器，它可为上层应用提供统一的资源管理和调度。

Sqoop数据迁移工具

Sqoop是一款开源的数据导入导出工具，主要用于在Hadoop与传统的数据库间进行数据的转换。

Mahout数据挖掘算法库

Mahout是Apache旗下的一个开源项目，它提供了一些可扩展的机器学习领域经典算法的实现，旨在帮助开发人员方便快捷地创建智能应用程序。

HBase分布式存储系统

HBase是Google Bigtable克隆版，它是一个针对结构化数据的可伸缩、高可靠、高性能、分布式和面向列的动态模式数据库。

Zookeeper分布式协作服务

Zookeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和HBase的重要组件。

Hive基于Hadoop的数据仓库

Hive是基于Hadoop的一个分布式数据仓库工具，可以将结构化的数据文件映射为一张数据库表，将SQL语句转换为MapReduce任务进行运行。

Flume日志收集工具

Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，Flume支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。

Hadoop的版本

•Hadoop发行版本分为开源社区版和商业版。

•社区版是指由Apache软件基金会维护的版本，是官方维护的版本体系。

•商业版Hadoop是指由第三方商业公司在社区版Hadoop基础上进行了一些修改、整合以及各个服务组件兼容性测试而发行的版本。

•Hadoop自诞生以来，主要分为Hadoop1、Hadoop2、Hadoop3三个系列的多个版本，

•目前市场上最主流的是Hadoop2.x版本。

•Hadoop2.x版本指的是第2代Hadoop，它是从Hadoop1.x发展而来的，并且相对于Hadoop1.x来说，有很多改进。

Hadoop1.0内核主要由分布式存储系统HDFS和分布式计算框架MapReduce两个系统组成，而Hadoop2.x版本主要新增了资源管理框架Yarn以及其他工作机制的改变。

2 Hadoop集群构建

虚拟机的安装

https://www.bilibili.com/read/cv7450678

Centos系统安装

https://www.bilibili.com/read/cv7456817

安装SecureCRT工具

https://www.bilibili.com/read/cv7457324

配置虚拟网络IP

https://www.bilibili.com/read/cv7463717

使用SecureCRT远程访问Centos系统

https://www.bilibili.com/read/cv7457324中的后半部有讲解

虚拟机的克隆

略（比较简单）

Linux系统网络配置

创建3台虚拟机hadoop01,hadoop02,hadoop03，每台虚拟机进行如下操作

1.静态Ip配置

参考ip可选范围，配置静态IP地址，成功后可以使用SecureCRT工具

$ vi /etc/sysconfig/network-scripts/ifcfg-ens33

#举例 ：hodoop01 
TYPE=Ethernet
PROXY_METHOD=none
BROWSER_ONLY=no
BOOTPROTO=static
DEFROUTE=yes
IPV4_FAILURE_FATAL=no
IPV6INIT=yes
IPV6_AUTOCONF=yes
IPV6_DEFROUTE=yes
IPV6_FAILURE_FATAL=no
IPV6_ADDR_GEN_MODE=stable-privacy
NAME=ens33
UUID=ce1e1643-4332-4151-a2a8-86dc56e77b93
DEVICE=ens33
ONBOOT=yes
IPADDR=192.168.206.200
NETMASK=255.255.255.0
GATEWAY=192.168.206.2
DNS1=8.8.8.8
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

2.主机名

（1）配置主机名，具体指令如下。

针对centos 6.x

$ vi /etc/sysconfig/network

NETWORKING=yes
HOSTNAME=hadoop01

1
2
3
4
5

针对 centos 7.x

$ vi /etc/hostname

	hadoop01
1
2
3

3.配置ip的映射

$ vi /etc/hosts

192.168.206.200 hadoop01
192.168.206.201 hadoop02
192.168.206.202 hadoop03
1
2
3
4
5

4.配置网卡设备的MAC地址

​ centos7 .x 没有该文件,不需要配置

​ centos 6.x版本需要，自行百度

$ vi /etc/udev/rules.d/70-persistent-net.rules

1
2

5.配置效果验证

ping hadoop01
ping hadoop02
ping hadoop03
ping www.baidu.com
1
2
3
4

SSH服务配置

实际工作中，服务器被放置在机房中，同时受到地域和管理的限制，开发人员通常不会进入机房操作直接上机操作，而是通过远程连接服务器，进行相关操作。

在集群开发中，主节点通常会对集群中各个节点频繁的访问，就需要不断输入目标服务器的用户名和密码，这种操作方式非常麻烦并且还会影响集群服务的连续运行。

为了解决上述问题，我们可以通过配置SSH服务来实现远程登录和SSH的免密登录功能。

1. SSH远程登录功能配置

（1）安装并开启SSH服务

#查询是否已经安装ssh服务，centos7 默认安装
$ rpm -qa | grep ssh
1
2

# 查询ssh服务是否启动,centos7默认启动
$ ps -e | grep sshd
1
2

# 如果发现系统中没有安装ssh，怎么办呢
$ yum install openssh-server
1
2

（2）创建CRT工具和服务器hadoop01的快速连接

​ 略

见 安装SecureCRT工具
https://www.bilibili.com/read/cv7457324

2. SSH免密登录功能配置

（1）生成SSH文件（公钥和私钥）

​ 公钥加密，私钥解密

​ hadoop01

$ ssh-keygen -t rsa
1

和hadoop03同理，不演示了

（2）拷贝3台机器的公钥到一台机器的某文件，将该文件发送到每台机器上

#分别在3台机器上执行该命令，生成某文件为（/root/.ssh/authorized_keys）
$ ssh-copy-id hadoop01
1
2

（3）将某文件，发送到hadoop02和hadoop03 机器内

hadoop01上

$ scp /root/.ssh/authorized_keys hadoop02:/root/.ssh

$ scp /root/.ssh/authorized_keys hadoop03:/root/.ssh

1
2
3
4

（4）验证免密登录效果

$ ssh hadoop02
$  exit
$ ssh hadoop03
1
2
3

Hadoop集群部署模式

独立模式

在独立模式下，所有程序都在单个JVM上执行，调试Hadoop集群的MapReduce程序也非常方便。一般情况下，该模式常用于学习或开发阶段进行调试程序。

伪分布模式

在伪分布式模式下， Hadoop程序的守护进程都运行在一台节点上，该模式主要用于调试Hadoop分布式程序的代码，以及程序执行是否正确。伪分布式模式是完全分布式模式的一个特例。

完全分布式模式

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

JDK安装

参考文档

https://www.bilibili.com/read/cv7532179

官网下载地址

https://www.oracle.com/java/technologies/oracle-java-archive-downloads.html

1-新建目录专门存放安装包

mkdir -p /export/software/
cd /export/software/
1
2

2-使用crt，安装lrzsz命令，实现将win上文件传输到linux上

win和linux都需要联网

yum install lrzsz -y
1

3-创建目录，存放解压后的文件

mkdir -p /export/servers/
1

4-将压缩包解压到指定目录

tar -zxvf jdk-8u161-linux-x64.tar.gz -C /export/servers/
1

5-将解压后包目录，重新命名，方便使用

mv mv jdk1.8.0_161/ jdk/
1

6-配置jdk的系统环境变量

vi /etc/profile

#添加如下内容
 export JAVA_HOME=/export/servers/jdk
 export PATH=$PATH:$JAVA_HOME/bin
 export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/bin/tools.jar
1
2
3
4
5
6

7- 使修改生效

source /etc/profile
1

8-检验是否生效

java -version
1

Hadoop安装

帮助文档

https://www.bilibili.com/read/cv7591643

官网下载地址(贼慢)

https://archive.apache.org/dist/hadoop/common/

hadoop-2.7.7.tar.gz

1-解压并安装

# hadoop-2.7.7.tar.gz 上传到 /export/software/
cd /export/software/
tar -zxvf hadoop-2.7.7.tar.gz -C /export/servers/
1
2
3

2- /etc/profile 中配置hadoop系统环境变量

export HADOOP_HOME=/export/servers/hadoop-2.7.7
export PATH=:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH

1
2
3

3-使修改生效

source /etc/profile
1

4-验证

hadoop version
1

Hadoop集群配置

1-配置Hadoop集群主节点

1.1 修改hadoop-env.sh文件

export JAVA_HOME=/export/servers/jdk
1

1.2 修改core-site.xml文件

<configuration>
	<property>
    	<name>fs.defaultFS</name>
        <value>hdfs://hadoop01:9000</value>
    </property>
    <property>
    	<name>hadoop.tmp.dir</name>
        <value>/export/servers/hadoop-2.7.7/tmp</value>
    </property>
</configuration>
1
2
3
4
5
6
7
8
9
10

1.3 修改hdfs-site.xml文件

<configuration>
         <property>
                <name>dfs.replication</name>
                <value>3</value>
        </property>
        <property>
                <name>dfs.namenode.secondary.http-address</name>
                <value>hadoop02:50090</value>
        </property>
</configuration>

1
2
3
4
5
6
7
8
9
10
11

1.4 修改mapred-site.xml 文件（复制mapred-site.xml.template获得）

<configuration>
	<property>
    	<name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>
</configuration>
1
2
3
4
5
6

1.5 修改 yarn-site.xml文件

<configuration>
	<property>
    	<name>yarn.resourcemanager.hostname</name>
        <value>hadoop01</value>
    </property>
    <property>
    	<name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>
</configuration>
1
2
3
4
5
6
7
8
9
10

1.6 修改slaves文件(覆盖localhost)

hadoop01
hadoop02
hadoop03
1
2
3

2 将集群主节点的配置文件分发到其他子节点

完成Hadoop集群主节点hadoop01的配置后，还需要将系统环境配置文件、JDK安装目录和Hadoop安装目录分发到其他子节点hadoop02和hadoop03上，具体指令：

$ scp /etc/profile hadoop02:/etc/profile
$ scp /etc/profile hadoop03:/etc/profile
$ scp -r /export/  hadoop02:/
$ scp -r /export/  hadoop03:/

1
2
3
4
5

3 在hadoop02和hadoop03上执行,使环境变量生效

source /etc/profile
1

格式化文件系统

初次启动HDFS集群时，必须对主节点进行格式化处理。

格式化文件系统指令如下：

# 任意目录下输入如下
$ hdfs namenode -format
或者
$ hadoop namenode -format
1
2
3
4

启动和关闭Hadoop集群

针对Hadoop集群的启动，需要启动内部包含的HDFS集群和YARN集群两个集群框架。启动方式有两种：一种是单节点逐个启动；另一种是使用脚本一键启动。

1 单节点逐个启动和关闭

    
在主节点上执行指令启动/关闭HDFS NameNode进程；
hadoop-daemon.sh start namenode
在每个从节点上执行指令启动/关闭HDFS DataNode进程；
hadoop-daemon.sh start datanode

在主节点上执行指令启动/关闭YARN ResourceManager进程；
yarn-daemon.sh start resourcemanager
在每个从节点上执行指令启动/关闭YARN nodemanager进程；
yarn-daemon.sh start nodemanager

在节点hadoop02执行指令启动/关闭SecondaryNameNode进程
 hadoop-daemon.sh start secondarynamenode
1
2
3
4
5
6
7
8
9
10
11
12
13

注意

hadoop01 有1个主节点 和1个从节点

hadoop-daemon.sh start namenode
hadoop-daemon.sh start datanode
1
2

yarn-daemon.sh start resourcemanager
yarn-daemon.sh start nodemanager
1
2

hadoop02 有1个从节点

hadoop-daemon.sh start datanode
1

yarn-daemon.sh start nodemanager
1

hadoop-daemon.sh start secondarynamenode
1

hadoop03 有1个从节点

hadoop-daemon.sh start datanode
1

yarn-daemon.sh start nodemanager
1

2 脚本一键启动和关闭

在主节点hadoop01上执行指令“start-dfs.sh”或“stop-dfs.sh”启动/关闭所有HDFS服务进程；(注意，不包含 secondarynamenode 进程)

在主节点hadoop01上执行指令“start-yarn.sh”或“stop-yarn.sh”启动/关闭所有YARN服务进程；

在主节点hadoop01上执行“start-all.sh”或“stop-all.sh”指令，直接启动/关闭整个Hadoop集群服务。

1
2
3
4
5
6

3. Hadoop测试效果

Hadoop集群服务启动后，在各个机器上执行“jps”指令查看各节点的服务进程的启动情况，效果如下所示。

jps
1

通过UI界面查看Hadoop运行状态

Hadoop集群正常启动后，它默认开放了两个端口50070和8088，分别用于监控HDFS集群和YARN集群。通过UI界面可以方便地进行集群的管理和查看，只需要在本地操作系统的浏览器输入集群服务的IP和对应的端口号即可访问。

在Windows系统下，访问http://hadoop01:50070，查看HDFS集群状态，且从图中可以看出HDFS集群状态显示正常。

在Windows系统下，访问http://hadoop01:8088，查看Yarn集群状态，且从图中可以看出Yarn集群状态显示正常。

#关闭hadoop01/02/03的防火墙

#关闭
systemctl stop firewalld.service
#查看状态
systemctl status firewalld.service
1
2
3
4

#在window系统的目录：C:\Windows\System32\drivers\etc\hosts文件中添加

192.168.206.200 hadoop01
192.168.206.201 hadoop02
192.168.206.202 hadoop03
1
2
3

#访问 http://hadoop01:50070

#访问 http://hadoop01:8088

Hadoop集群初体验

参考文档

https://www.bilibili.com/read/cv7598831

Hadoop经典案例单词统计

1 打开HDFS的UI界面，查看HDFS中是否有数据文件，默认是没有数据文件。

2 准备文本文件，在Linux系统上编辑一个文本文件，然后上传至HDFS上。

#在linux系统中创建一个目录，创建一个文件，写点内容
mkdir -p /export/data
cd /export/data
vi word.txt (写点内容)
#将该文件上传到hdfs中目录下： /wordcount/input/
hadoop fs -mkdir -p /wordcount/input
hadoop fs -put /export/data/word.txt /wordcount/input/

1
2
3
4
5
6
7
8

3 运行hadoop-mapreduce-examples-2.7.4.jar包，实现词频统计。

重要

务必要关闭所有机器h01,h02,h03的防护墙，不然执行会报错

hadoop jar hadoop-mapreduce-examples-2.7.7.jar  wordcount /wordcount/input /wordcount/output
1

4 查看UI界面，Yarn集群UI界面出现程序运行成功的信息。HDFS集群UI界面出现了结果文件。

3 HDFS分布式文件系统

Hadoop的核心是HDFS和MapReduce。其中，HDFS是解决海量大数据文件存储的问题，是目前应用最广泛的分布式文件系统。

HDFS的演变

HDFS 源于 Google 在2003年10月份发表的GFS（Google File System）论文，接下来，我们从传统的文件系统入手，开始学习分布式文件系统，以及分布式文件系统是如何演变而来？

HDFS的基本概念

HDFS（Hadoop Distributed Filesystem）是一个易于扩展的分布式文件系统，运行在成百上千台低成本的机器上。它与现有的分布式文件系统有许多相似之处，都是用来存储数据的系统工具，而区别于HDFS具有高度容错能力，旨在部署在低成本机器上。HDFS主要用于对海量文件信息进行存储和管理，也就是解决大数据文件（如TB乃至PB级）的存储问题。

1. NameNode(名称节点)

NameNode是HDFS集群的主服务器，通常称为名称节点或者主节点。一旦NameNode关闭，就无法访问Hadoop集群。NameNode主要以元数据的形式进行管理和存储，用于维护文件系统名称并管理客户端对文件的访问；NameNode记录对文件系统名称空间或其属性的任何更改操作；HDFS负责整个数据集群的管理，并且在配置文件中可以设置备份数量，这些信息都由NameNode存储。

2. DataNode(数据节点)

DataNode是HDFS集群中的从服务器，通常称为数据节点。文件系统存储文件的方式是将文件切分成多个数据块，这些数据块实际上是存储在DataNode节点中的，因此DataNode机器需要配置大量磁盘空间。它与NameNode保持不断的通信，DataNode在客户端或者NameNode的调度下，存储并检索数据块，对数据块进行创建、删除等操作，并且定期向NameNode发送所存储的数据块列表。

3. Block（数据块）

每个磁盘都有默认的数据块大小，这是磁盘进行数据读/写的最小单位，HDFS同样也有块（block）的概念，它是抽象的块，而非整个文件作为存储单元，在Hadoop2.x版本下，默认大小是128M，且备份3份，每个块尽可能地存储于不同的DataNode中。按块存储的好处主要是屏蔽了文件的大小，提供数据的容错性和可用性。

4. Rack（机架）

Rack是用来存放部署Hadoop集群服务器的机架，不同机架之间的节点通过交换机通信，HDFS通过机架感知策略，使NameNode能够确定每个DataNode所属的机架ID，使用副本存放策略，来改进数据的可靠性、可用性和网络带宽的利用率。

5. Metadata（元数据）

元数据从类型上分可分三种信息形式，一是维护HDFS文件系统中文件和目录的信息，例如文件名、目录名、父目录信息、文件大小、创建时间、修改时间等；二是记录文件内容存储相关信息，例如文件分块情况、副本个数、每个副本所在的DataNode信息等；三是用来记录HDFS中所有DataNode的信息，用于DataNode管理。

HDFS的特点

随着互联网数据规模的不断增大，对文件存储系统提出了更高的要求，需要更大的容量、好更的性能以及安全性更高的文件存储系统，与传统分布式文件系统一样，HDFS分布式文件系统也是通过计算机网络与节点相连，也有传统分布式文件系统的优点和缺点。

优点

高容错

流式数据访问

支持超大文件

高数据吞吐量

缺点

高延迟

不适合小文件存取

不适合并发写入

HDFS架构和原理

HDFS存储架构

HDFS是一个分布式的文件系统，相比普通的文件系统来说更加复杂，因此在学习HDFS的操作之前有必要先来学习一下HDFS的存储架构。

•HDFS采用主从架构（Master/Slave架构）。

•HDFS集群是由一个NameNode和多个的 DataNode组成。

HDFS写数据原理

Client从HDFS中存储数据，即为Write（写）数据。

步骤

1 客户端发起文件上传请求，通过RPC（远程过程调用）与NameNode建立通讯

2 NameNode检查元数据文件的系统目录树

3 若系统目录树的父目录不存在该文件相关信息，返回客户端可以上传文件

4 客户端请求上传第一个Block数据块以及数据块副本的数量

5 NameNode检测元数据文件中DataNode信息池，找到可用的数据节点

6 NameNode检查元数据文件的系统目录树

7 若系统目录树的父目录不存在该文件相关信息，返回客户端可以上传文件

8 DataNode之间建立Pipeline后，逐个返回建立完毕信息

9 客户端与DataNode建立数据传输流，开始发送数据包

10 客户端向DataNode_01上传第一个Block数据块，当DataNode_01收到一个Packet就会传给DataNode_02，DataNode_02传给DataNode_03，DataNode_01每传送一个Packet都会放入一个应答队列等待应答。

11 数据被分割成一个个Packet数据包在Pipeline上依次传输，而在Pipeline反方向上，将逐个发送Ack，最终由Pipeline中第一个DataNode节点DataNode_01将Pipeline的 Ack信息发送给客户端。

12 DataNode返回给客户端，第一个Block块传输完成。客户端则会再次请求NameNode上传第二个Block块和第三块到服务器上，重复上面的步骤，直到3个Block都上传完毕。

HDFS读数据原理

从HDFS中查找数据，即为Read（读）数据。

步骤

1 客户端向NameNode发起RPC请求，来获取请求文件Block数据块所在的位置。

2 NameNode检测元数据文件，会视情况返回Block块信息或者全部Block块信息，对于每个Block块，NameNode都会返回含有该Block副本的DataNode地址。

3 客户端会选取排序靠前的DataNode来依次读取Block块，每一个Block都会进行CheckSum若文件不完整，则客户端会继续向NameNode获取下一批的Block列表，直到验证读取出来文件是完整的，则Block读取完毕。

4 客户端会把最终读取出来所有的Block块合并成一个完整的最终文件（例如：1.txt）。

HDFS的shell操作

Shell在计算机科学中俗称“壳”，是提供给使用者使用界面的,进行与系统交互的软件，通过接收用户输入的命令执行相应的操作，Shell分为图形界面Shell和命令行式Shell。

文件系统（FS）Shell包含了各种的类Shell的命令，可以直接与Hadoop分布式文件系统以及其他文件系统进行交互。

#基本语法结构
hadoop fs <args>
hdfs   dfs  <args>
1
2
3

# <args>
# 上传文件
-put
# 删除文件/空白文件夹
-help
# 统计目录下所有文件大小
-du
# 等等
1
2
3
4
5
6
7
8

应用场景

Shell定时采集数据到HDFS

服务器每天会产生大量日志数据，并且日志文件可能存在于每个应用程序指定的data目录中，在不使用其它工具的情况下，将服务器中的日志文件规范的存放在HDFS中。通过编写简单的Shell脚本，用于每天自动采集服务器上的日志文件，并将海量的日志上传至HDFS中。

1. 配置环境变量

首先在/export/data/logs目录下（如果目录不存在，则需要提前创建）使用vi命令创建upload2HDFS.sh脚本文件，在编写Shell脚本时，需要设置Java环境变量和Hadoop环境变量，这样做是用来提高系统的可靠性，保障运行程序的机器在没有配置环境变量的情况下依然能够运行脚本。

upload2HDFS.sh

#!/bin/bash

#配置java环境变量
export JAVA_HOME=/export/servers/jdk
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib 
export PATH=${JAVA_HOME}/bin:$PATH

#配置hadoop环境变量
export HADOOP_HOME=/export/servers/hadoop-2.7.7/
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

1
2
3
4
5
6
7
8
9
10
11
12

2. 准备日志存放目录和待上传文件

为了让开发者便于控制上传文件的流程，可以在脚本中设置一个日志存放目录和待上传文件目录，若上传过程中发生错误只需要查看该目录就能知道文件的上传进度。

upload2HDFS.sh

#日志文件存放的目录
log_src_dir=/export/data/logs/log/
#待上传文件存放的目录
log_toupload_dir=/export/data/logs/toupload/
1
2
3
4

3. 设置日志文件上传的路径

设置上传的HDFS目标路径，命名格式以时间结尾，并且输出打印信息

upload2HDFS.sh

date1='date -d last-day +%Y_%m_%d'
#日志文件上传到hdfs的根路径
hdfs_root_dir=/data/clickLog/$date1/
#打印环境变量信息
echo "envs: hadoop_home: SHADOOP HOME"
#读取日志文件的目录,判断是否有需要上传的文件
echo "log_src_dir:"$log_src_dir
1
2
3
4
5
6
7

4. 实现文件上传

上传文件的过程就是遍历文件目录的过程，将文件首先移动到待上传目录，再从待上传目录中上传到HDFS中。

ls $log_src_dir | while read fileName
do 
	if [[ "$fileName" == access.log.* ]]; then
		date =`date+%Y_%m_%d_%H_%M_%S`
		#将文件移动到待上传目录并重命名
		echo "moving $log_src_dir$flieName to $log_toupload_dir"xxxxx_click_log_$fileName"$date"
		mv $log_src_dir$fileName $log_toupload_dir“xxxxx_click_log_$fileName”$date
		#将待上传的文件path写入一个列表文件willDoing
		echo $log_toupload_dir"xxxxx_click_log_$fileName"$date>>$log_toupload_dir"willDoing."$date
	fi
done
1
2
3
4
5
6
7
8
9
10
11

将文件从待上传目录传到HDFS中，具体如下

若是在每天12点凌晨执行一次，我们可以使用Linux Crontab表达式执行定时任务

5. 执行程序展示运行结果。

为了模拟生产环境，在日志存放目录/export/data/logs/log/中，手动创建日志文件，access.log表示正在源源不断的产生日志的文件，access.log.1、access.log.2等表示已经滚动完毕的日志文件，即为待上传日志文件。

在upload2HDFS.sh文件路径下运行脚本 sh upload2HDFS.sh，先将日志存放目录log中的日志文件移到待上传toupload目录下，并根据业务需求重命名；然后脚本执行“hadoop put”上传命令，将待上传目录下的所有日志文件上传至HDFS；最后通过HDFS Web界面可看到需要采集的日志文件已按照日期分类，上传至HDFS中。

HDFS的Java API操作

参考文档

https://www.bilibili.com/read/cv7604251

由于Hadoop是使用Java语言编写的，因此可以使用Java API操作Hadoop文件系统。HDFS Shell本质上就是对Java API的应用，通过编程的形式操作HDFS，其核心是使用HDFS提供的Java API构造一个访问客户端对象，然后通过客户端对象对HDFS上的文件进行操作（增、删、改、查）。

HDFS Java API介绍

Hadoop整合了众多文件系统，HDFS只是这个文件系统的一个实例。

在Java中操作HDFS，创建一个客户端实例主要涉及以下两个类：

•Configuration：该类的对象封装了客户端或者服务器的配置，Configuration实例会自动加载HDFS的配置文件core-site.xml，从中获取Hadoop集群的配置信息。

•FileSystem：该类的对象是一个文件系统对象。

FileSystem对象的一些方法可以对文件进行操作，常用方法如下：

案例——使用Java API操作HDFS

0.在Window系统下，配置hadoop环境，使用Java开发工具编写代码，相当于Hadoop客户端

在win系统下不配置hadoop环境，直接运行代码会报错，显示缺少

winutils.exe 和 hadoop.dll 两个文件

步骤

1-将hadoop-win-2.7.7文件夹拷贝到一个没有中文没有空格的路径中；

比如

D:\Program Files\hadoop-win-2.7.7\bin
1

2-在windows系统，配置hadoop的环境变量: HADOOP_HOME,并将%HADOOP_HOME%\bin添加到path中

3-:把hadoop2.7.7文件夹中bin目录下的hadoop.dll 放到系统盘: C:Windows System32目录

4-关闭windows重启

1. 搭建项目环境

创建一个项目名为“HadoopDemo”，包名为“com.wukong”的Maven项目，并在项目的pom.xml文件中引入hadoop-common、hadoop-hdfs、hadoop-client以及单元测试junit的依赖。

补充

使用Idea中自带的maven插件，构建maven项目，比较方便。

创建maven项目时候，选择maven-quickstart 原型。

<dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.11</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>2.7.7</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>2.7.7</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>2.7.7</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-core</artifactId>
      <version>2.7.7</version>
    </dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

2. 初始化客户端对象

首先在项目src文件夹下创建com.wukong.hdfsdemo包，并在该包下创建HDFS_CRUD.java文件，编写Java测试类，构建Configuration和FileSystem对象，初始化一个客户端实例进行相应的操作。

3. 上传文件到HDFS

由于采用Java测试类来实现JavaApi对HDFS的操作，因此可以在HDFS_CRUD.java文件中添加一个testAddFileToHdfs()方法来演示本地文件上传到HDFS的示例。

4. 从HDFS下载文件到本地

在HDFS_CRUD.java文件中添加一个testDownloadFileToLocal()方法，来实现从HDFS中下载文件到本地系统的功能。

5. 目录操作

在HDFS_CRUD.java文件添加一个testMkdirAndDeleteAndRename()方法，实现目录的创建、删除、重命名的功能。

6. 查看目录中的文件信息

在HDFS_CRUD.java文件中添加一个testListFiles()方法，实现查看目录中所有文件的详细信息的功能。

package com.wukong.hdfsdemo;

import java.io.FileNotFoundException;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.Before;
import org.junit.Test;

public class HDFS_CRUD {
	FileSystem fs = null;

	@Before
	public void init() throws Exception {
		// 构造一个配置参数对象,设置一个参数：我们要访问的hdfs的URI
		Configuration conf = new Configuration();
		// 这里指定使用的是HDFS文件系统
		conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
		// 通过如下的方式进行客户端身份的设置
		System.setProperty("HADOOP_USER_NAME", "root");
		// 通过FileSystem的静态方法获取文件系统客户端对象
		fs = FileSystem.get(conf);
	}

	@Test
	public void testAddFileToHdfs() throws IOException {
		// 要上传的文件所在本地路径
		Path src = new Path("D:/test.txt");
		// 要上传到hdfs的目标路径
		Path dst = new Path("/testFile");
		// 上传文件方法
		fs.copyFromLocalFile(src, dst);
		// 关闭资源
		fs.close();
	}

	// 从hdfs中复制文件到本地文件系统
	@Test
	public void testDownloadFileToLocal() throws IllegalArgumentException, IOException {
		// 下载文件
		fs.copyToLocalFile(new Path("/testFile"), new Path("D:/"));
	}

	// 创建，删除，重命名文件
	@Test
	public void testMkdirAndDeleteAndRename() throws Exception {
		// 创建目录
		fs.mkdirs(new Path("/a/b/c"));
		fs.mkdirs(new Path("/a2/b2/c2"));
		// 重命名文件或文件夹
		fs.rename(new Path("/a"), new Path("/a3"));
		// 删除文件夹，如果是非空文件夹，参数2必须给值true
		fs.delete(new Path("/a2"), true);
	}

	// 查看目录信息，只显示文件
	@Test
	public void testListFiles() throws FileNotFoundException, IllegalArgumentException, IOException {
		// 获取迭代器对象
		RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
		while (listFiles.hasNext()) {
			LocatedFileStatus fileStatus = listFiles.next();
			// 打印当前文件名
			System.out.println(fileStatus.getPath().getName());
			// 打印当前文件块大小
			System.out.println(fileStatus.getBlockSize());
			// 打印当前文件权限
			System.out.println(fileStatus.getPermission());
			// 打印当前文件内容长度
			System.out.println(fileStatus.getLen());
			// 获取该文件块信息（包含长度，数据块，datanode的信息）
			BlockLocation[] blockLocations = fileStatus.getBlockLocations();
			for (BlockLocation bl : blockLocations) {
				System.out.println("block-length:" + bl.getLength() + "--" + "block-offset:" + bl.getOffset());
				String[] hosts = bl.getHosts();
				for (String host : hosts) {
					System.out.println(host);
				}
			}
			System.out.println("----------------------------");
		}
	}
    
// 查看文件及文件夹信息
    @Test
    public void testListAll() throws FileNotFoundException, IllegalArgumentException, IOException {
        // 获取HDFS系统中文件和目录的元数据等信息
        FileStatus[] listStatus = fs.listStatus(new Path("/"));

        for (FileStatus fstatus : listStatus) {
            String flag = "d--             ";
            // 判断是文件还是文件夹
            if (fstatus.isFile()) {
                flag = "f--         ";
            }
            System.out.println(flag + fstatus.getPath().getName());
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

4 MapReduce分布式计算框架

MapReduce是Hadoop系统核心组件之一，它是一种可用于大数据并行处理的计算模型、框架和平台，主要解决海量数据的计算，是目前分布式计算模型中应用较为广泛的一种。

MapReduce核心思想

MapReduce的核心思想是“分而治之”。所谓“分而治之”就是把一个复杂的问题，按照一定的“分解”方法分为等价的规模较小的若干部分，然后逐个解决，分别找出各部分的结果，把各部分的结果组成整个问题的结果，这种思想来源于日常生活与工作时的经验，同样也完全适合技术领域。

MapReduce作为一种分布式计算模型，它主要用于解决海量数据的计算问题。使用MapReduce操作海量数据时，每个MapReduce程序被初始化为一个工作任务，每个工作任务可以分为Map和Reduce两个阶段。

Map阶段负责将任务分解，即把复杂的任务分解成若干个“简单的任务”来并行处理，但前提是这些任务没有必然的依赖关系，可以单独执行任务。

Reduce阶段负责将任务合并，即把Map阶段的结果进行全局汇总点击此处输入文字。

MapReduce就是“任务的分解与结果的汇总”。即使用户不懂分布式计算框架的内部运行机制，但是只要能用Map和Reduce思想描述清楚要处理的问题，就能轻松地在Hadoop集群上实现分布式计算功能。


MapReduce编程模型

MapReduce是一种编程模型，用于处理大规模数据集的并行运算。使用MapReduce执行计算任务的时候，每个任务的执行过程都会被分为两个阶段，分别是Map和Reduce，其中Map阶段用于对原始数据进行处理，Reduce阶段用于对Map阶段的结果进行汇总，得到最终结果。

MapReduce编程实例-词频统计

首先，MapReduce通过默认组件TextInputFormat将待处理的数据文件（如text1.txt和text2.txt），把每一行的数据都转变为<key，value>键值对。

其次，调用Map()方法，将单词进行切割并进行计数，输出键值对作为Reduce阶段的输入键值对。

最后，调用Reduce()方法将单词汇总、排序后，通过TextOutputFormat组件输出到结果文件中。

MapReduce工作过程

MapTask工作原理

MapTask作为MapReduce工作流程前半部分，它主要经历5个阶段，分别是Read阶段、Map阶段、Collect阶段、Spill阶段和Combiner阶段。

ReduceTask工作原理

ReduceTask的工作过程主要经历了5个阶段，分别是Copy阶段、Merge阶段、Sort阶段、Reduce阶段和Write阶段。

Shuffle工作原理

Shuffle是MapReduce的核心，它用来确保每个reducer的输入都是按键排序的。它的性能高低直接决定了整个MapReduce程序的性能高低，map和reduce阶段都涉及到了shuffle机制。

MapReduce 编程组件

InputFormat组件

主要用于描述输入数据的格式，它提供两个功能，分别是数据拆分和为Mapper提供输入数据。

Mapper组件

Hadoop提供的Mapper类是实现Map任务的一个抽象基类，该基类提供了一个map()方法。

Reducer组件

Map过程输出的键值对，将由Reducer组件进行合并处理，最终的某种形式的结果输出

Partitioner组件

Partitioner组件可以让Map对Key进行分区，从而可以根据不同的key分发到不同的Reduce中去处理，其目的就是将key均匀分布在ReduceTask上

Combiner组件

Combiner组件的作用就是对Map阶段的输出的重复数据先做一次合并计算，然后把新的（key，value）作为Reduce阶段的输入。

OutputFormat组件

OutputFormat是一个用于描述MapReduce程序输出格式和规范的抽象类。

MapReduce 运行模式

本地运行模式

在当前的开发环境模拟MapReduce执行环境，处理的数据及输出结果在本地操作系统。

WordCountCombiner.java

package cn.wukong.hadoop.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

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

	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Context context) throws IOException, InterruptedException {
		// 1.局部汇总
		int count = 0;
		for (IntWritable v : values) {
			count += v.get();
		}
		context.write(key, new IntWritable(count));
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

WordCountDriver.java

package cn.wukong.hadoop.mr;

import org.apache.hadoop.conf.Configuration;
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;

/**
 * Driver类就是MR程序运行的主类，本类中组装了一些程序运行时所需要的信息
 * 比如：使用的Mapper类是什么，Reducer类，数据在什么地方，输出在哪里
 * 
 * @author wukong
 *
 */
public class WordCountDriver {

	public static void main(String[] args) throws Exception {
		// 通过Job来封装本次MR的相关信息
		Configuration conf = new Configuration();

		Job wcjob = Job.getInstance(conf);

		// 指定MR Job jar包运行主类
		wcjob.setJarByClass(WordCountDriver.class);
		// 指定本次MR所有的Mapper Reducer类
		wcjob.setMapperClass(WordCountMapper.class);
		wcjob.setReducerClass(WordCountReducer.class);

		// 设置我们的业务逻辑 Mapper类的输出 key和 value的数据类型
		wcjob.setMapOutputKeyClass(Text.class);
		wcjob.setMapOutputValueClass(IntWritable.class);

		// 设置我们的业务逻辑 Reducer类的输出 key和 value的数据类型
		wcjob.setOutputKeyClass(Text.class);
		wcjob.setOutputValueClass(IntWritable.class);
		
		//设置Combiner组件
		wcjob.setCombinerClass(WordCountCombiner.class);
		

		// 指定要处理的数据所在的位置
		FileInputFormat.setInputPaths(wcjob, "D:/mr/input"); 
		// 指定处理完成之后的结果所保存的位置
		FileOutputFormat.setOutputPath(wcjob, new Path("D:/mr/output"));

		// 提交程序并且监控打印程序执行情况
		boolean res = wcjob.waitForCompletion(true);
		System.exit(res ? 0 : 1);
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

WordCountMapper.java

package cn.wukong.hadoop.mr;

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;

/**
 * 
 * 这里就是MapReduce程序 Map阶段业务逻辑实现的类 Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 * 
 * KEYIN:表示mapper数据输入时key的数据类型，在默认读取数据组件下，叫作ImportFormat,它的行为是每行读取待处理的数据
 * 读取一行，就返回一行给MR程序，这种情况下 KEYIN就表示每一行的起始偏移，因此数据类型是Long
 * 
 * VALUEIN:表示mapper数据输入的时候Value的数据类型，在默认读取数据组件下，valueIN就表示读取的一行内容 因此数据类型是String
 * 
 * KEYOUT:表示mapper阶段数据输出的时候key的数据类型，在本案例中输出的key是单词，因此数据类型是String
 * ValueOUT:表示mapper阶段数据输出的时候value的数据类型，在本案例中输出的value是单次的此书，因此数据类型是Integer
 * 
 * 这里所说的数据类型String,Long都是JDK的自带的类型，数据在分布式系统中跨网络传输就需要将数据序列化，默认JDK序列化时效率低下，因此
 * 使用Hadoop封装的序列化类型。 
 long--LongWritable 
 String --Text 
 Integer intWritable ....
 *
 * @author wukong
 *
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
	/**
	 * 这里就是mapper阶段具体业务逻辑实现的方法 该方法的调用取决于读取数据的组件有没有给MR传入数据
	 * 如果有数据传入，每一个<k,v>对，map就会被调用一次
	 */
	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		// 拿到传入进来的一行内容，把数据类型转换为String
		String line = value.toString();
		// 将这行内容按照分隔符切割
		String[] words = line.split(" ");
		// 遍历数组，每出现一个单词就标记一个数组1 例如：<单词,1>
		for (String word : words) {
			// 使用MR上下文context，把Map阶段处理的数据发送给Reduce阶段作为输入数据
			context.write(new Text(word), new IntWritable(1));
			//第一行 hadoop hadoop spark  发送出去的是<hadoop,1><hadoop,1><spark,1> 
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

WordCountReducer.java

package cn.wukong.hadoop.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

//都要继承Reducer 这就是我们所说的变成模型，只需要套模板就行了
/**
 * 这里是MR程序 reducer阶段处理的类
 * 
 * KEYIN：就是Reducer阶段输入的数据key类型，对应Mapper阶段输出KEY类型 ，在本案例中就是单词
 * 
 * VALUEIN：就是Reducer阶段输入的数据value类型，对应Mapper阶段输出VALUE类型 ，在本案例中就是个数
 * 
 * KEYOUT:就是Reducer阶段输出的数据key类型，在本案例中，就是单词 Text
 * 
 * VALUEOUT:reducer阶段输出的数据value类型，在本案例中，就是单词的总次数
 * 
 * @author wukong
 *
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

	/**
	 * 这里是REDUCE阶段具体业务类的实现方法
	 * 第一行 hadoop hadoop spark  发送出去的是<hadoop,1><hadoop,1><spark,1> 
	 * reduce接受所有来自Map阶段处理的数据之后，按照Key的字典序进行排序
	 * 按照key是否相同作一组去调用reduce方法
	 * 本方法的key就是这一组相同的kv对 共同的Key
	 * 把这一组的所有v作为一个迭代器传入我们的reduce方法
	 * 
	 * 迭代器：<hadoop,[1,1]>
	 * 
	 */
	@Override
	protected void reduce(Text key, Iterable<IntWritable> value,
			Context context) throws IOException, InterruptedException {
		//定义一个计数器
		int count = 0;
		//遍历一组迭代器，把每一个数量1累加起来就构成了单词的总次数
		
		//
		for (IntWritable iw : value) {
			count += iw.get();
		}
        //<hello,3>
		context.write(key, new IntWritable(count));
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

集群运行模式

把MapReduce程序打成一个Jar包，提交至Yarn集群上去运行任务。由于Yarn集群负责资源管理和任务调度，程序会被框架分发到集群中的节点上并发的执行，因此处理的数据和输出结果都在HDFS文件系统中。

MapReduce性能优化策略

使用Hadoop进行大数据运算，当数据量极其大时，那么对MapReduce性能的调优重要性不言而喻，尤其是Shuffle过程中的参数配置对作业的总执行时间影响特别大，我们可以从五个方面对MapReduce程序进行性能调优，分别是数据输入、Map阶段、Reduce阶段、Shuffle阶段和其他调优属性方面。

数据输入

在执行MapReduce任务前，将小文件进行合并，大量小文件会产生大量的map任务，增大map任务装载次数，而任务装载较耗时，从而导致MapReduce运行速度较慢。因此采用****CombineTextInputFormat来作为输入，解决输入端大量的小文件场景。**

Map阶段

•减少溢写（spill）次数

•减少合并（merge）次数

•在map之后，不影响业务逻辑前提下，先进行combine处理，减少 I/O

Reduce阶段

•合理设置map和reduce数

•设置map、reduce共存

•规避使用reduce

•合理设置reduce端的buffer

Shuffle阶段

Shuffle阶段的调优就是给Shuffle过程尽量多地提供内存空间，以防止出现内存溢出现象，可以由参数mapred.child.java.opts来设置，任务节点上的内存大小应尽量大。

其他调优属性

MapReduce还有一些基本的资源属性的配置，这些配置的相关参数都位于mapred-default.xml文件中，我们可以合理配置这些属性提高MapReduce性能，例如合理设置MapTask、ReduceTask等参数

MapReduce 经典案例：倒排索引

1. 倒排索引介绍

倒排索引是文档检索系统中最常用的数据结构，被广泛应用于全文搜索引擎。倒排索引主要用来存储某个单词（或词组）在一组文档中的存储位置的映射，提供了可以根据内容来查找文档的方式，而不是根据文档来确定内容，因此称为倒排索引（Inverted Index）。带有倒排索引的文件我们称为倒排索引文件，简称倒排文件(Inverted File)。

2. 案例需求及分析

现假设有三个源文件file1.txt、file2.txt和file3.txt，需要使用倒排索引的方式对这三个源文件内容实现倒排索引，并将最后的倒排索引文件输出。

首先，使用默认的TextInputFormat类对每个输入文件进行处理，得到文本中每行的偏移量及其内容。Map过程首先分析输入的<key，value>键值对，经过处理可以得到倒排索引中需要的三个信息：单词、文档名称和词频。

经过Map阶段数据转换后，同一个文档中相同的单词会出现多个的情况，而单纯依靠后续Reduce阶段无法同时完成词频统计和生成文档列表，所以必须增加一个Combine阶段，先完成每一个文档的词频统计。

经过上述两个阶段的处理后，Reduce阶段只需将所有文件中相同key值的value值进行统计，并组合成倒排索引文件所需的格式即可。

1. Map阶段实现

首先，使用开发工具打开之前创建的Maven项目HadoopDemo，并且新创建cn.wukong.mr.invertedIndex包，在该路径下编写自定义Mapper类InvertedIndexMapper，主要用于将文本中的单词按照空格进行切割，并以冒号拼接，“单词：文档名称”作为key，单词次数作为value，都以文本方式输出至Combine阶段。

InvertedIndexMapper.java

package cn.wukong.mr.InvertedIndex;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class InvertedIndexMapper extends Mapper<LongWritable, Text, Text, Text> {

	private static Text keyInfo = new Text();// 存储单词和 URL 组合
	private static final Text valueInfo = new Text("1");// 存储词频,初始化为1

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String line = value.toString();
		String[] fields = StringUtils.split(line, " ");// 得到字段数组
		FileSplit fileSplit = (FileSplit) context.getInputSplit();// 得到这行数据所在的文件切片
		String fileName = fileSplit.getPath().getName();// 根据文件切片得到文件名
		for (String field : fields) {
			// key值由单词和URL组成，如“MapReduce:file1   ，1”
			keyInfo.set(field + ":" + fileName);
			context.write(keyInfo, valueInfo);
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

2. Combine阶段实现

根据Map阶段的输出结果形式，在cn.wukong.mr.InvertedIndex包下，自定义实现Combine阶段的类InvertedIndexCombiner，对每个文档的单词进行词频统计。

InvertedIndexCombiner.java

package cn.wukong.mr.InvertedIndex;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
 *  @author 悟空非空也
 *   B站/CSDN/公众号：悟空非空也
 *
 */
public class InvertedIndexCombiner extends Reducer<Text, Text, Text, Text> {

	private static Text info = new Text();

	// 输入： <MapReduce:file3 {1,1,...}>
	// 输出：<MapReduce file3:2>
	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		int sum = 0;// 统计词频
		for (Text value : values) {
			sum += Integer.parseInt(value.toString());
		}
		int splitIndex = key.toString().indexOf(":");
		// 重新设置 value 值由 URL 和词频组成
		info.set(key.toString().substring(splitIndex + 1) + ":" + sum);
		// 重新设置 key 值为单词
		key.set(key.toString().substring(0, splitIndex));
		context.write(key, info);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

3. Reduce阶段实现

根据Combine阶段的输出结果形式，同样在cn.wukong.mr.InvertedIndex包下，自定义Reducer类InvertedIndexMapper，主要用于接收Combine阶段输出的数据，并最终案例倒排索引文件需求的样式，将单词作为key，多个文档名称和词频连接作为value，输出到目标目录。

InvertedIndexReducer.java

package cn.wukong.mr.InvertedIndex;

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class InvertedIndexReducer extends Reducer<Text, Text, Text, Text> {  
	  
    private static Text result = new Text();  
  
    // 输入：<MapReduce file3:2>  
    // 输出：<MapReduce file1:1;file2:1;file3:2;>  
    @Override  
    protected void reduce(Text key, Iterable<Text> values, Context context)  
            throws IOException, InterruptedException {  
        // 生成文档列表  
        String fileList = new String();  
        for (Text value : values) {  
            fileList += value.toString() + ";";  
        }  
  
        result.set(fileList);  
        context.write(key, result);  
    }  
}  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

4. Driver程序主类实现

编写MapReduce程序运行主类InvertedIndexDriver，主要用于设置MapReduce工作任务的相关参数，由于本次演示的数据量较小，为了方便、快速进行案例演示，本案例采用了本地运行模式，指定的本地D:\InvertedIndex\input目录下的源文件（需要提前准备）实现倒排索引，并将结果输入到本地D:\InvertedIndex\output目录下。

InvertedIndexRunner.java

package cn.wukong.mr.InvertedIndex;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
/**
 *
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class InvertedIndexRunner {  
    public static void main(String[] args) throws IOException,  
            ClassNotFoundException, InterruptedException {  
        Configuration conf = new Configuration();  
        Job job = Job.getInstance(conf);  
  
        job.setJarByClass(InvertedIndexRunner.class);  
  
        job.setMapperClass(InvertedIndexMapper.class);  
        job.setCombinerClass(InvertedIndexCombiner.class);  
        job.setReducerClass(InvertedIndexReducer.class);  
  
        job.setOutputKeyClass(Text.class);  
        job.setOutputValueClass(Text.class);  
  
        FileInputFormat.setInputPaths(job, new Path("D:\\InvertedIndex\\input"));
        // 指定处理完成之后的结果所保存的位置
        FileOutputFormat.setOutputPath(job, new Path("D:\\InvertedIndex\\output")); 
        
        // 向 yarn 集群提交这个 job
        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }  
}  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

5. 效果测试

为了保证MapReduce程序正常执行，需要先在本地D:\InvertedIndex\input目录下创建file1.txt、file2.txt和file3.txt；然后执行MapReduce程序的程序入口InvertedIndexDriver类，正常执行完成后，会在指定的D:\InvertedIndex\output下生成结果文件。

MapReduce 经典案例：数据去重

1. 数据去重介绍

数据去重主要是为了掌握利用并行化思想来对数据进行有意义的筛选，数据去重指去除重复数据的操作。

2. 案例需求及分析

文件file1.txt本身包含重复数据，并且与file2.txt同样出现重复数据，现要求使用Hadoop大数据相关技术对以上2个文件进行去重操作，并最终将结果汇总到一个文件中。

(1) 编写MapReduce程序，在Map阶段采用Hadoop默认作业输入方式后，将key设置为需要去重的数据，而输出的value可以任意设置为空。

(2) 在Reduce阶段，不需要考虑每一个key有多少个value，可以直接将输入的key复制为输出的key，而输出的value可以任意设置为空，这样就会使用MapReduce默认机制对key（也就是文件中的每行内容）自动去重。

1. Map阶段实现

使用开发工具打开之前创建的Maven项目HadoopDemo，并且新创建cn.wukong.mr.dedup包，在该路径下编写自定义Mapper类DedupMapper，主要用于读取数据集文件将TextInputFormat默认组件解析的类似<0，2018-3-1 a >键值对修改为<2018-3-1 a，null>。

DedupMapper.java

package cn.wukong.mr.dedup;

import java.io.IOException;

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

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class DedupMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

	private static Text field = new Text();
	// <0,2018-3-3 c><11,2018-3-4 d>
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException,
			InterruptedException {
		field = value;
		context.write(field, NullWritable.get());
	}
	// <2018-3-3 c,null> <2018-3-4 d,null>
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

2. Reduce阶段实现

根据Map阶段的输出结果形式，同样在cn.wukong.mr.dedup包下，自定义Reducer类DedupReducer，主要用于接受Map阶段传递来的数据，根据Shuffle工作原理，键值key相同的数据就会被合并，因此输出数据就不会出现重复数据了。

DedupReducer.java

package cn.wukong.mr.dedup;

import java.io.IOException;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class DedupReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
	//  <2018-3-3 c,null><2018-3-4 d,null><2018-3-4 d,null>
	@Override
	protected void reduce(Text key, Iterable<NullWritable> values, Context context)
			throws IOException, InterruptedException {
		context.write(key, NullWritable.get());
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

3. Driver程序主类实现

编写MapReduce程序运行主类DedupDriver，主要用于设置MapReduce工作任务的相关参数。由于本次演示的数据量较小，为了方便、快速地进行案例演示，本案例采用了本地运行模式，对指定的本地D:\Dedup\input目录下的源文件（需要提前准备）实现数据去重，并将结果输入到本地D:\Dedup\output目录下。

DedupRunner.java

package cn.wukong.mr.dedup;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class DedupRunner {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(DedupRunner.class);

		job.setMapperClass(DedupMapper.class);
		job.setReducerClass(DedupReducer.class);

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

		FileInputFormat.setInputPaths(job, new Path("D:\\Dedup\\input"));
		// 指定处理完成之后的结果所保存的位置
		FileOutputFormat.setOutputPath(job, new Path("D:\\Dedup\\output"));

		job.waitForCompletion(true);

	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

4. 效果测试

为了保证MapReduce程序正常执行，需要先在本地D:\Dedup\input目录下创建文件file1.txt和file2.txt；然后，执行MapReduce程序的程序入口DedupDriver类，正常执行完成后，在指定的D:\Dedup\output目录下生成结果文件。

MapReduce 经典案例：TopN

1. TopN分析法介绍

TopN分析法是指从研究对象中按照某一个指标进行倒序或正序排列，取其中所需的N个数据，并对这N个数据进行重点分析的方法。

2. 案例需求及分析

现假设有数据文件num.txt，现要求使用MapReduce技术提取上述文本中最大的5个数据，并最终将结果汇总到一个文件中。

(1) 先设置MapReduce分区为1，即ReduceTask个数一定只有一个。我们需要提取TopN，即全局的前N条数据，不管中间有几个Map、Reduce，最终只能有一个用来汇总数据。

(2) 在Map阶段，使用TreeMap数据结构保存TopN的数据，TreeMap默认会根据其键的自然顺序进行排序，也可根据创建映射时提供的 Comparator进行排序，其firstKey()方法用于返回当前集合最小值的键。

(3) 在Reduce阶段，将Map阶段输出数据进行汇总，选出其中的TopN数据，即可满足需求。这里需要注意的是，TreeMap默认采取正序排列，需求是提取5个最大的数据，因此要重写Comparator类的排序方法进行倒序排序。

1. Map阶段实现

使用开发工具打开之前创建的Maven项目HadoopDemo，并且新建cn.wukong.mr.topN包，在该路径下编写自定义Mapper类TopNMapper，主要用于将文件中的每行数据进行切割提取，并把数据保存到TreeMap中，判断TreeMap是否大于5，如果大于5就需要移除最小的数据。TreeMap保存了当前文件最大5条数据后，再输出到Reduce阶段。

TopNMapper.java

package cn.wukong.mr.topn;

import java.util.TreeMap;

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

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class TopNMapper extends Mapper<LongWritable, Text, NullWritable, IntWritable> {

	private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>();

	// <0,10 3 8 7 6 5 1 2 9 4>
	// <xx,11 12 17 14 15 20>
	@Override
	public void map(LongWritable key, Text value, Context context) {
		String line = value.toString();
		String[] nums = line.split(" ");
		for (String num : nums) {
			repToRecordMap.put(Integer.parseInt(num), " ");
			if (repToRecordMap.size() > 5) {
				repToRecordMap.remove(repToRecordMap.firstKey());
			}
		}
	}
	@Override
	protected void cleanup(Context context) {
		for (Integer i : repToRecordMap.keySet()) {
			try {
				context.write(NullWritable.get(), new IntWritable(i));
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

2. Reduce阶段实现

根据Map阶段的输出结果形式，同样在cn.wukong.mr.topN包下，自定义Reducer类TopNReducer，主要用于编写TreeMap自定义排序规则，当需求取最大值时，只需要在compare()方法中返回正数即可满足倒序排列，reduce()方法依然是要满足时刻判断TreeMap中存放数据是前五个数，并最终遍历输出最大的5个数。

TopNReducer.java

package cn.wukong.mr.topn;

import java.io.IOException;
import java.util.Comparator;
import java.util.TreeMap;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class TopNReducer extends Reducer<NullWritable, IntWritable, NullWritable, IntWritable> {
	
	private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>(new Comparator<Integer>() {
		
		//返回一个基本类型的整型,谁大谁排后面.
		//返回负数表示：o1 小于o2
		//返回0表示:表示：o1和o2相等
		//返回正数表示：o1大于o2。
		public int compare(Integer a, Integer b) {
			return b - a;
		}
	});
	public void reduce(NullWritable key, Iterable<IntWritable> values, Context context)
			throws IOException, InterruptedException {
		for (IntWritable value : values) {
			repToRecordMap.put(value.get(), " ");
			if (repToRecordMap.size() > 5) {
				repToRecordMap.remove(repToRecordMap.firstKey());
			}
		}
		for (Integer i : repToRecordMap.keySet()) {
			context.write(NullWritable.get(), new IntWritable(i));
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

3. Driver程序主类实现

编写MapReduce程序运行主类TopNDriver，主要用于对指定的本地D:\topN\input目录下的源文件（需要提前准备）实现TopN分析，得到文件中最大的5个数，并将结果输入到本地D:\topN\output目录下。

TopNRunner.java

package cn.wukong.mr.topn;

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

/**
 * @author 悟空非空也
 * B站/CSDN/公众号：悟空非空也
 */
public class TopNRunner {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		Job job = Job.getInstance(conf);

		job.setJarByClass(TopNRunner.class);
		job.setMapperClass(TopNMapper.class);
		job.setReducerClass(TopNReducer.class);

		job.setNumReduceTasks(1);

		job.setMapOutputKeyClass(NullWritable.class);// map阶段的输出的key
		job.setMapOutputValueClass(IntWritable.class);// map阶段的输出的value

		job.setOutputKeyClass(NullWritable.class);// reduce阶段的输出的key
		job.setOutputValueClass(IntWritable.class);// reduce阶段的输出的value

		FileInputFormat.setInputPaths(job, new Path("D:\\topN\\input"));
		FileOutputFormat.setOutputPath(job, new Path("D:\\topN\\output"));

		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

4. 效果测试

为了保证MapReduce程序正常执行，需要先在本地D:\topN\input目录下创建文件num.txt；然后，执行MapReduce程序的程序入口TopNDriver类，正常执行完成后，在指定的D:\topN\output目录下生成结果文件。