Hadoop相关笔记Hadoop入门一、大数据概述二、Hadoop框架&大数据生态三、Hadoop运行环境搭建四、Hadoop运行模式五、Hadoop编译源码（面试重点）Hadoop（HDFS）（存储）一、概述*二、Shell命令（开发重点）三、HDFS客户端操作（开发重点）四、HDFS的数据流（面试重点）五、NameNode和SecondaryNameNode（面试开发重点）六、DataNode（面试开发重点）七、HDFS 2.X新特性Hadoop(MapReduce)（计算）一、MapRed

204 阅读 0 评论 135 点赞

我是靠谱客的博主稳重学姐，这篇文章主要介绍Hadoop相关笔记Hadoop入门一、大数据概述二、Hadoop框架&大数据生态三、Hadoop运行环境搭建四、Hadoop运行模式五、Hadoop编译源码（面试重点）Hadoop（HDFS）（存储）一、概述*二、Shell命令（开发重点）三、HDFS客户端操作（开发重点）四、HDFS的数据流（面试重点）五、NameNode和SecondaryNameNode（面试开发重点）六、DataNode（面试开发重点）七、HDFS 2.X新特性Hadoop(MapReduce)（计算）一、MapRed，现在分享给大家，希望可以做个参考。

Hadoop入门

一、大数据概述

1.大数据特点（4V）

Volume（大量）
Velocity（高速）
Variety（多样）
Value（低价值密度）

2.大数据部门组织结构

在这里插入图片描述

二、Hadoop框架&大数据生态

1.Hadoop是什么

分布式系统基础框架
主要解决，海量数据的存储和海量数据的分析计算

2.Hadoop三大发行版本

Apache版本最原始（最基础）的版本，对于入门学习最好。
Cloudera在大型互联网企业中用的较多。
Hortonworks文档较好。

3.Hadoop的优势（4高）

高可靠性
高扩展性
高效性
高容错性

4.Hadoop组成（面试重点）

（1）Hadoop1.x和Hadoop2.x区别

在这里插入图片描述

（2）HDFS架构概述

1）NameNode（nn）：存储文件的元数据（目录）

2）DataNode（dn）：在本地文件系统存储文件块数据，以及块数据的校验和（内容）

3）Secoundary NameNode（2nn）：用来监控HDFS状态的辅助后台程序，每隔一段时间获取HDFS元数据的快照

（3）YARN架构概述

1）ResourceManager（RM）主要作用：（整个集群资源的老大）

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

2）NodeManager（NM）主要作用：（某一个节点资源的老大）

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

3）ApplicationMaster（AM）作用：

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

4）Container

是YARN中的资源抽象，封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等
在这里插入图片描述

（4）MapReduce架构概述

MapReduce将计算过程分为两个阶段：Map和Reduce

1）Map阶段并行处理输入数据（分）

2）Reduce阶段对Map结果进行汇总（合）

5.大数据技术生态体系

在这里插入图片描述

三、Hadoop运行环境搭建

1.虚拟机环境准备

（1）克隆虚拟机

在这里插入图片描述

（2）修改静态IP及主机名

在这里插入图片描述
1）查看版本

 cat /etc/redhat-release

2）查看主机名

uname -a

3）修改主机名

vi /etc/hostname

在这里插入图片描述

保存退出

修改/etc/hosts文件

[admin@hadoop02 ~]$ sudo vi /etc/hosts

4）修改网络配置文件

/etc/sysconfig/network-scripts/ifcfg-*

在这里插入图片描述

保存退出

5）关闭防护墙

查看防火墙状态

service firewalld status

关闭防火墙

service firewalld stop

6）重启 reboot

7）重启后查看网络配置

命令：

ip addr

在这里插入图片描述

（3）创建admin用户

命令：

#添加用户	
useradd admin
#给用户指定/修改密码
passwd admin

（4）配置admin用户具有root权限

1）修改root用户对/etc/sudoers文件的读写权限（原来只读）

chmod 760 /etc/sudoers

2）修改文件/etc/sudoers的内容

root    ALL=(ALL)      ALL
admin   ALL=(ALL)       ALL
#保存退出

3）切换到admin用户

su admin

4）测试admin查看sudoers文件

[admin@hadoop02 root]$ ：

在命令前加 sudo 则是使用root权限了

（5）在/opt目录下创建文件夹

1）在/opt目录下创建module、software文件夹

software存储所有的jar包
module是jar包解压后存放位置

[admin@hadoop02 opt]$ sudo mkdir software
[admin@hadoop02 opt]$ sudo mkdir module

2）修改这两个文件的所有者和所有者的组

[admin@hadoop02 opt]$ sudo chown admin:admin module/ software/

2.安装JDK

1）将两个jar包拖入software文件夹中
在这里插入图片描述

2）卸载现有JDK

查询是否安装Java软件：

[admin@hadoop01 opt]$ rpm -qa | grepjava

如果安装的版本低于1.7，卸载该JDK：
```
[admin@hadoop01 opt]$ sudo rpm -e 软件包
```

3）解压JDK到/opt/module目录下

[admin@hadoop01 software]$ tar -zxvf jdk-8u144-linux-x64.tar.gz -C /opt/module/

4）配置JDK环境变量

先获取JDK路径

[admin@hadoop01 jdk1.8.0_144]$ pwd
/opt/module/jdk1.8.0_144

打开/etc/profile文件

[admin@hadoop01 jdk1.8.0_144]$ sudo vi /etc/profile

在profile文件末尾添加JDK路径

  #JAVA_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_144
  export PATH=$PATH:$JAVA_HOME/bin

5）查看java版本

[admin@hadoop01 jdk1.8.0_144] source /etc/profile
[admin@hadoop01 jdk1.8.0_144] java -version

3.安装Hadoop

1）解压

[admin@hadoop01 software]$ tar -zxvf hadoop-2.7.2.tar.gz -C /opt/module/

2）配置环境变量

获取Hadoop安装路径

[admin@hadoop01 hadoop-2.7.2]$ pwd
/opt/module/hadoop-2.7.2

打开/etc/profile文件

[admin@hadoop01 hadoop-2.7.2]$ sudo vi /etc/profile

在配置文件最后加上

#HADOOP_HOME
export HADOOP_HOME=/opt/module/hadoop-2.7.2
export PATH= $P A T H :$ HADOOP_HOME/bin
export PATH= $P A T H :$ HADOOP_HOME/sbin


3）测试hadoop是否装上

```shell
[admin@hadoop01 hadoop-2.7.2] source /etc/profile
[admin@hadoop01 hadoop-2.7.2] hadoop

4.Hadoop重要目录

bin目录：存放对Hadoop相关服务（HDFS,YARN）进行操作的脚本
etc目录：Hadoop的配置文件目录，存放Hadoop的配置文件
lib目录：存放Hadoop的本地库（对数据进行压缩解压缩功能）
sbin目录：存放启动或停止Hadoop相关服务的脚本
share目录：存放Hadoop的依赖jar包、文档、和官方案例

四、Hadoop运行模式

1.本地模式

（1）官方Grep案例

【grep(过滤)案例就是执行一个mapreduce程序,从一堆文件里面找出符合那个正则的单词,输出到一个文件夹里】

1）创建在hadoop-2.7.2文件下面创建一个input文件夹

[admin@hadoop01 hadoop-2.7.2]$ mkdir input

2）将Hadoop的xml配置文件复制到input文件夹

[admin@hadoop01 hadoop-2.7.2]$ cp etc/hadoop/*.xml input/

3）执行share目录下的MapReduce程序

[admin@hadoop01 hadoop-2.7.2]$ hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar grep input/ output 'dfs[a-z.]+'

4）查看输出结果

[admin@hadoop01 hadoop-2.7.2]$ cd output/
[admin@hadoop01 output]$ cat part-r-00000

（2）官方WordCount案例

【统计单词个数案例】

1）创建在hadoop-2.7.2文件下面创建一个wcinput文件夹

[admin@hadoop01 hadoop-2.7.2]$ mkdir wcinput

2）在wcinput文件下创建一个wc.input文件

[admin@hadoop01 hadoop-2.7.2] cd wcinput/
[admin@hadoop01 wcinput] touch wc.input

3）编辑wc.input文件

[admin@hadoop01 wcinput]$ vi wc.input

输入需要统计的单词

tianyi huichao lihua
zhangchen xiaoheng
xinbo xinbo
gaoyang gaoyang
yanjing yanjing

4）回到Hadoop目录/opt/module/hadoop-2.7.2，执行程序

[admin@hadoop01 wcinput] cd ..
[admin@hadoop01 hadoop-2.7.2] hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar wordcount wcinput/ wcoutput

5）查看结果

[admin@hadoop01 hadoop-2.7.2]$ cd wcoutput/
[admin@hadoop01 wcoutput]$ ll
[admin@hadoop01 wcoutput]$ cat part-r-00000

2.伪分布式模式

（1）启动HDFS&运行MapReduce程序

1）配置集群

（a）配置：core-site.xml（etc/hadoop文件下）

在etc/hadoop/core-site.xml中加入

<configuration>这个标签里面，已存在
	<!-- 指定HDFS中NameNode的地址 -->
    <property>
            <name>fs.defaultFS</name>
            <value>hdfs://hadoop01:9000</value>
    </property>

    <!-- 指定Hadoop运行时产生文件的存储目录 -->
    <property>
            <name>hadoop.tmp.dir</name>
            <value>/opt/module/hadoop-2.7.2/data/tmp</value>
    </property>
</configuration>

（b）配置：hadoop-env.sh（etc/hadoop文件下）

Linux系统中获取JDK的安装路径：

[atguigu@ hadoop101 ~]# echo $JAVA_HOME
/opt/module/jdk1.8.0_144

修改JAVA_HOME 路径：

export JAVA_HOME=/opt/module/jdk1.8.0_144

（c）配置：hdfs-site.xml（etc/hadoop文件下）

 <!-- [指定HDFS副本的数量 -->
  <property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>

2）启动集群

（a）格式化NameNode（第一次启动时格式化，以后就不要总格式化）

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs namenode -format

（b）启动NameNode

[admin@hadoop01 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start namenode

（c）启动DataNode

[admin@hadoop01 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start datanode

3）查看集群

（a）查看启动是否成功

[admin@hadoop01 hadoop-2.7.2]$ jps
2322 Jps
2277 DataNode
2186 NameNode

（b）web端查看HDFS文件系统

http://192.168.255.141:50070/
在这里插入图片描述

4）操作集群

（a）在HDFS文件系统上创建一个input文件夹

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs dfs -mkdir -p /user/admin/input

（b）将测试文件内容上传到文件系统上

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs dfs -put wcinput/wc.input /user/admin/input

（c）运行MapReduce程序

[admin@hadoop01 hadoop-2.7.2]$ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar wordcount /user/admin/input /user/admin/output

（d）查看输出结果

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs dfs -cat /user/admin/output/p*
gaoyang	2
huichao	1
lihua	1
tianyi	1
xiaoheng	1
xinbo	2
yanjing	2
zhangchen	1

5）注意：格式化NameNode之前

先查看进程是否退出

[admin@hadoop01 hadoop-2.7.2]$ jps
2277 DataNode
2186 NameNode
2619 Jps

再删掉产生的数据文件data和logs

（2）启动YARN并运行MapReduce程序

1）配置集群

（a）配置yarn-env.sh（etc/hadoop文件下）

export JAVA_HOME=/opt/module/jdk1.8.0_144/

（b）配置yarn-site.xml（etc/hadoop文件下）

<configuration>添加到这里面
<!-- Reducer获取数据的方式 -->
	<property>
		<name>yarn.nodemanager.aux-services</name>
    	<value>mapreduce_shuffle</value>
 	</property>
<!-- 指定YARN的ResourceManager的地址 -->
	<property>
        <name>yarn.resourcemanager.hostname</name>
        <value>hadoop01</value>
	</property>
    <property>
    	<name>yarn.resourcemanager.webapp.address</name>
    	<value>hadoop01:8088</value>
  	</property>
</configuration>

（c）配置mapred-env.sh（etc/hadoop文件下）

export JAVA_HOME=/opt/module/jdk1.8.0_144/

（d）配置： mapred-site.xml（etc/hadoop文件下）

对mapred-site.xml.template重新命名为 mapred-site.xml

[admin@hadoop01 hadoop]$ mv mapred-site.xml.template mapred-site.xml

指定MR运行在YARN上

<configuration>添加到这里面
<!-- 指定MR运行在YARN上 -->
	<property>
    	<name>mapreduce.framework.name</name>
    	<value>yarn</value>
	</property>
</configuration>

2）启动集群

（a）启动前必须保证NameNode和DataNode已经启动

[admin@hadoop01 hadoop]$ jps
2723 Jps
2277 DataNode
2186 NameNode

（b）启动ResourceManager

[admin@hadoop01 hadoop-2.7.2]$ sbin/yarn-daemon.sh start resourcemanager

（c）启动NodeManager

[admin@hadoop01 hadoop-2.7.2]$ sbin/yarn-daemon.sh start nodemanager

3）集群操作

（a）YARN的浏览器页面查看

http://192.168.255.141:8088/
在这里插入图片描述

（b）删除文件系统上的output文件

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs dfs -rm -r /user/admin/output

（c）执行MapReduce程序

[admin@hadoop01 hadoop-2.7.2]$ hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar wordcount /user/admin/input /user/admin/output

（d）查看运行结果

刷新页面http://192.168.255.141:8088/

（3）配置历史服务器

1）配置mapred-site.xml（etc/hadoop文件下）

[admin@hadoop01 hadoop-2.7.2]$ vi etc/hadoop/mapred-site.xml

<!-- 历史服务器端地址 -->
<property>
	<name>mapreduce.jobhistory.address</name>
    <value>hadoop01:10020</value>
    </property>

<!-- 历史服务器web端地址 -->
<property>
	<name>mapreduce.jobhistory.webapp.address</name>
	<value>hadoop01:19888</value>
</property>

2）启动历史服务器

[admin@hadoop01 hadoop-2.7.2]$ sbin/mr-jobhistory-daemon.sh start historyserver

3）查看历史服务器是否启动

[admin@hadoop01 hadoop-2.7.2]$ jps

4）查看JobHistory
在这里插入图片描述

（4）配置日志的聚集

注意：开启日志聚集功能，需要重新启动NodeManager 、ResourceManager和HistoryManager

1）关闭NodeManager 、ResourceManager和HistoryManager

2）配置yarn-site.xml

<!-- 日志聚集功能使能 -->
<property>
	<name>yarn.log-aggregation-enable</name>
    <value>true</value>
</property>
<!-- 日志保留时间设置7天 -->
<property>
	<name>yarn.log-aggregation.retain-seconds</name>
    <value>604800</value>
</property>

3）开启NodeManager 、ResourceManager和HistoryManager

4）删除HDFS上已经存在的输出文件

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs dfs -rm -r /user/admin/output

5）执行WordCount程序

[admin@hadoop01 hadoop-2.7.2]$ hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar wordcount /user/admin/input /user/admin/output

6）查看日志

http://hadoop01:19888/jobhistory/logs/hadoop01:32846/container_1585355177434_0001_01_000001/job_1585355177434_0001/admin

（5）配置文件说明

1）默认配置文件：

要获取的默认文件	文件存放在Hadoop的jar包中的位置
[core-default.xml]	hadoop-common-2.7.2.jar/ core-default.xml
[hdfs-default.xml]	hadoop-hdfs-2.7.2.jar/ hdfs-default.xml
[yarn-default.xml]	hadoop-yarn-common-2.7.2.jar/ yarn-default.xml
[mapred-default.xml]	hadoop-mapreduce-client-core-2.7.2.jar/ mapred-default.xml

2）自定义配置文件：

core-site.xml、hdfs-site.xml、yarn-site.xml、mapred-site.xml 四个配置文件存放在$HADOOP_HOME/etc/hadoop这个路径上，用户可以根据项目需求重新进行修改配置。

*3.完全分布式模式（开发重点）

*（1）虚拟机准备

*（2）编写集群分发脚本xsync

1）scp（secure copy）安全拷贝

（a）scp定义：

scp可以实现服务器与服务器之间的数据拷贝。（from server1 to server2）

（b）基本语法-

scp -r  pdir/fname   user@hadoophost:pdir/fname
命令   递归      要拷贝的文件路径/名称   	目的用户@主机:目的路径/名称

（c）案例

1、在hadoop01上，将hadoop01中/opt/module目录下的软件拷贝到hadoop02上。（推）

[admin@hadoop01 opt]$ scp -r module root@hadoop02:/opt/module

2、在hadoop03上，将hadoop01服务器上的/opt/module目录下的软件拷贝到hadoop03上。（拉）

[admin@hadoop03 opt]$ sudo scp -r admin@hadoop01:/opt/module ./

3、在hadoop03上操作将hadoop01中/opt/module目录下的软件拷贝到hadoop04上

[admin@hadoop03 opt]$ scp -r admin@hadoop01:/opt/module root@hadoop04:/opt/moudle

（d）修改文件所有者和所有组

[admin@hadoop02 opt]$ sudo chown admin:admin module/ -R

（e）将hadoop01中/etc/profile文件拷贝到hadoop02、03、04的/etc/profile上。

[admin@hadoop01~]$ sudo scp /etc/profile root@hadoop02:/etc/profile

注意：拷贝过来的配置文件别忘了source一下/etc/profile

（f）将hadoop01中/opt/software文件拷贝到hadoop02、03、04的/opt/software

[admin@hadoop01 opt]$ scp -r software root@hadoop02:/opt/software

拷贝完之后修改文件所有者和所有组：

[admin@hadoop02 opt]$ sudo chown admin:admin software/ -R

2） rsync 远程同步工具

（a）定义：rsync主要用于备份和镜像。具有速度快、避免复制相同内容和支持符号链接的优点。

（b）rsync和scp区别

用rsync做文件的复制要比scp的速度快，rsync只对差异文件做更新。scp是把所有文件都复制过去。

（c）基本语法

rsync    -rvl      pdir/fname              user@hadoophost:pdir/fname
命令   选项参数  要拷贝的文件路径/名称   目的用户@主机:目的路径/名称

选项参数说明：

选项	功能
-r	递归
-v	显示复制过程
-l	拷贝符号连接

（d）案例实操

1、把hadoop01机器上的/opt/software目录同步到hadoop02服务器的root用户下的/opt/目录

[admin@hadoop101 opt]$ rsync -rvl /opt/software/ root@hadoop02:/opt/software

3）xsync集群分发脚本

脚本案例

（a）在/home/admin目录下创建bin目录，并在bin目录下创建xsync文件

[admin@hadoop02 ~]$ mkdir bin
[admin@hadoop02 bin]$ touch xsync

（b）修改xsync文件

[admin@hadoop02 bin]$ vi xsync

脚本内容：

#!/bin/bash
#1 获取输入参数个数，如果没有参数，直接退出
pcount=$#
if((pcount==0)); then
echo no args;
exit;
fi

#2 获取文件名称
p1=$1
fname=`basename $p1`
echo fname=$fname

#3 获取上级目录到绝对路径
pdir=`cd -P $(dirname $p1); pwd`
echo pdir=$pdir

#4 获取当前用户名称
user=`whoami`

#5 循环
a=0
for((host=3; host<5; host++)); do
        echo ------------------- hadoop$host --------------
        rsync -rvl $pdir/$fname $user@hadoop$a$host:$pdir
done

（c）修改脚本 xsync 具有执行权限

[admin@hadoop02 bin]$ chmod 777 xsync

（d）调用脚本形式：xsync 文件名称

[admin@hadoop02 ~]$ xsync bin/

*（3）集群配置

1）集群部署规划

	hadoop02	hadoop03	hadoop04
HDFS	NameNode DataNode	DataNode	SecondaryNameNode DataNode
YARN	NodeManager	ResourceManager NodeManager	NodeManager

2）配置集群

（a）配置core-site.xml

[admin@hadoop02 hadoop]$ vi core-site.xml

<!-- 指定HDFS中NameNode的地址 -->
<property>
	<name>fs.defaultFS</name>
	<value>hdfs://hadoop02:9000</value>
</property>

（b）HDFS配置文件

配置hadoop-env.sh（伪不用修改）

[admin@hadoop02 hadoop]$ vi hadoop-env.sh
export JAVA_HOME=/opt/module/jdk1.8.0_144

配置hdfs-site.xml

[admin@hadoop02 hadoop]$ vi hdfs-site.xml

<!-- [指定HDFS副本的数量 -->
<property>
	<name>dfs.replication</name>
	<value>3</value>
</property>
<!-- 指定Hadoop辅助名称节点主机配置 -->
<property>
 	<name>dfs.namenode.secondary.http-address</name>
  	<value>hadoop04:50090</value>
</property>

（c）YARN配置文件

配置yarn-env.sh（伪不用修改）

[admin@hadoop02 hadoop]$ vi yarn-env.sh
export JAVA_HOME=/opt/module/jdk1.8.0_144

配置yarn-site.xml

[admin@hadoop02 hadoop]$ vi yarn-site.xml

<!-- 指定YARN的ResourceManager的地址-->
<property>
     <name>yarn.resourcemanager.hostname</name>
     <value>hadoop03</value>
</property>

（d）MapReduce配置文件

配置mapred-env.sh（伪不用修改）

[admin@hadoop02 hadoop]$  vi mapred-env.sh 
export JAVA_HOME=/opt/module/jdk1.8.0_144

配置mapred-site.xml

[admin@hadoop02 hadoop]$ cp mapred-site.xml.template mapred-site.xml
[admin@hadoop02 hadoop]$ vi mapred-site.xml

<!-- 指定MR运行在Yarn上 -->
<property>
		<name>mapreduce.framework.name</name>
		<value>yarn</value>
</property>

3）在集群上分发配置好的Hadoop配置文件

[admin@hadoop02 etc]$ xsync hadoop/

4）查看文件分发情况

[admin@hadoop04 hadoop-2.7.2]$ cat etc/hadoop/core-site.xml

（4）集群单点启动

1）如果集群是第一次启动，需要格式化NameNode

注意：先删除日志和之前的数据文件，三个都要删

[admin@hadoop02 hadoop-2.7.2]$ rm -rf data/ logs/ output/
[admin@hadoop02 hadoop-2.7.2]$ bin/hdfs namenode -format

2）在hadoop02上启动NameNode

[admin@hadoop02 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start namenode

3）在hadoop02、hadoop03以及hadoop04上分别启动DataNode

[admin@hadoop02 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start datanode
[admin@hadoop03 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start datanode
[admin@hadoop04 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start datanode

*（5）SSH无密登录配置

1）免密登录原理

在这里插入图片描述

2）生成公钥和私钥

[admin@hadoop02 .ssh]$ ssh-keygen -t rsa

（.ssh是家目录下的隐藏文件）

3）将公钥拷贝到要免密登录的目标机器上

[admin@hadoop02 .ssh]$ ssh-copy-id  hadoop03   #（复制到hadoop03）

然后hadoop03的.ssh文件夹下自动生成authorized_keys文件

[admin@hadoop03 .ssh]$ ll
total 8
-rw-------. 1 admin admin 396 Mar 30 13:56 authorized_keys
-rw-r--r--. 1 admin admin 372 Mar 30 13:50 known_hosts

[admin@hadoop02 .ssh]$ ssh-copy-id  hadoop04  #（复制到hadoop04）
[admin@hadoop02 .ssh]$ ssh-copy-id  hadoop02  #（复制到hadoop02，自己也需要）

4）hadoop03上同样配置

[admin@hadoop03 .ssh]$ ssh-keygen -t rsa
[admin@hadoop03 .ssh]$ ssh-copy-id hadoop02
[admin@hadoop03 .ssh]$ ssh-copy-id hadoop03
[admin@hadoop03 .ssh]$ ssh-copy-id hadoop04

5）hadoop02上采用root账号，配置一下无密登录

[admin@hadoop02 .ssh]$ su root
（root用户下没有.ssh的话，执行命令 ssh localhost）
[root@hadoop02 .ssh]# ssh-keygen -t rsa
[root@hadoop02 .ssh]# ssh-copy-id hadoop02
[root@hadoop02 .ssh]# ssh-copy-id hadoop03
[root@hadoop02 .ssh]# ssh-copy-id hadoop04

*（6）群起集群

1）配置slaves

配置slaves

# /opt/module/hadoop-2.7.2/etc/hadoop/slaves
[admin@hadoop02 hadoop]$ vi slaves

增加：

hadoop02
hadoop03
hadoop04

注意：该文件中添加的内容结尾不允许有空格，文件中不允许有空行。

2）启动集群

（a）若是之前有启动的集群，先关掉

[admin@hadoop02 hadoop-2.7.2]$ jps
2480 Jps
1860 DataNode
1770 NameNode

[admin@hadoop02 hadoop-2.7.2]$ sbin/hadoop-daemon.sh stop datanode
[admin@hadoop02 hadoop-2.7.2]$ sbin/hadoop-daemon.sh stop namenode

（hadoop03,04一样）

（b）启动

如果集群是第一次启动，需要格式化NameNode

[admin@hadoop01 hadoop-2.7.2]$ bin/hdfs namenode -format

启动HDFS

[admin@hadoop02 hadoop-2.7.2]$ sbin/start-dfs.sh

（c）启动YARN

[admin@hadoop03 hadoop-2.7.2]$ sbin/start-yarn.sh

（d）启动历史服务器

[admin@hadoop02 hadoop-2.7.2]$ sbin/mr-jobhistory-daemon.sh start historyserver

注意：NameNode和ResourceManger如果不是同一台机器，不能在NameNode上启动 YARN，应该在ResouceManager所在的机器上启动YARN。

3）集群基本测试

1）上传小文件

[admin@hadoop02 hadoop-2.7.2]$ bin/hdfs dfs -put wcinput/wc.input /

2）上传大文件

[admin@hadoop02 hadoop-2.7.2]$ bin/hdfs dfs -put /opt/software/hadoop-2.7.2.tar.gz /

3）查看HDFS文件存储路径

[admin@hadoop02 subdir0]$ pwd
/opt/module/hadoop-2.7.2/data/tmp/dfs/data/current/BP-938951106-192.168.10.107-1495462844069/current/finalized/subdir0/subdir0

*（7）集群启动/停止方式总结

1）各个服务组件逐一启动/停止（单节点启停）

（a）分别启动/停止HDFS组件

sbin/hadoop-daemon.sh  start / stop namenode / datanode / secondarynamenode

（b）启动/停止YARN

sbin/yarn-daemon.sh  start / stop resourcemanager / nodemanager

2）各个模块分开启动/停止（配置ssh是前提）常用

（a）整体启动/停止HDFS

sbin/start-dfs.sh   /  stop-dfs.sh

（b）整体启动/停止YARN

sbin/start-yarn.sh  /  stop-yarn.sh

（8）集群时间同步

时间同步的方式：找一个机器，作为时间服务器，所有的机器与这台集群时间进行定时的同步，比如，每隔十分钟，同步一次时间。

1）时间服务器配置（必须root用户）

（a）检查ntp是否安装

[root@hadoop02 hadoop-2.7.2]#rpm -qa |grep ntp
ntpdate-4.2.6p5-29.el7.centos.x86_64
ntp-4.2.6p5-29.el7.centos.x86_64
fontpackages-filesystem-1.44-8.el7.noarch

如果没有安装：

yum install ntp ntpdate -y
yum -y install fontconfig

（b）修改ntp配置文件

[root@hadoop02 hadoop-2.7.2]# vi /etc/ntp.conf

1、修改1（授权192.168.255.0-192.168.255.255网段上的所有机器可以从这台机器上查询和同步时间）

restrict 192.168.255.0 mask 255.255.255.0 nomodify notrap

2、修改2（集群在局域网中，不使用其他互联网上的时间）

#server 0.centos.pool.ntp.org iburst
#server 1.centos.pool.ntp.org iburst
#server 2.centos.pool.ntp.org iburst
#server 3.centos.pool.ntp.org iburst

3、添加3（当该节点丢失网络连接，依然可以采用本地时间作为时间服务器为集群中的其他节点提供时间同步）

server 127.127.1.0
fudge 127.127.1.0 stratum 10

（c）修改/etc/sysconfig/ntpd 文件

[root@hadoop02 hadoop-2.7.2]# vi /etc/sysconfig/ntpd

增加内容如下（让硬件时间与系统时间一起同步）

SYNC_HWCLOCK=yes

（d）重新启动ntpd服务

查看状态：

[root@hadoop02 hadoop-2.7.2]# service ntpd status

启动：

[root@hadoop02 hadoop-2.7.2]# service ntpd start

（e）设置ntpd服务开机启动

[root@hadoop02 hadoop-2.7.2]# chkconfig ntpd on

2）其他机器配置（必须root用户）

（a）测试：在其他机器配置1分钟与时间服务器同步一次

[root@hadoop03 hadoop-2.7.2]# crontab -e

输入：

*/1 * * * * /usr/sbin/ntpdate hadoop02

（b）修改任意机器时间

[root@hadoop03 hadoop-2.7.2]# date -s "2018-11-11 11:11:11"
Sun Nov 11 11:11:11 EST 2018

（c）1分钟后查看机器是否与时间服务器同步

[root@hadoop03 hadoop-2.7.2]# date
Mon Mar 30 16:25:52 EDT 2020

五、Hadoop编译源码（面试重点）

1.前期准备工作

（1）CentOS联网

配置CentOS能连接外网。Linux虚拟机ping www.baidu.com 是畅通的

（2）jar包准备

(hadoop源码、JDK8、maven、ant 、protobuf)

1）hadoop-2.7.2-src.tar.gz

2）jdk-8u144-linux-x64.tar.gz

3）apache-ant-1.9.9-bin.tar.gz（build工具，打包用的）

4）apache-maven-3.0.5-bin.tar.gz

5）protobuf-2.5.0.tar.gz（序列化的框架）

2.jar包安装

注意：所有操作必须在root用户下完成

（1）JDK解压、配置环境变量 JAVA_HOME和PATH，验证java-version(如下都需要验证是否配置成功)

[root@hadoop05 software]# tar -zxvf jdk-8u144-linux-x64.tar.gz -C /opt/module/
[root@hadoop05 software]# vi /etc/profile

##JAVA_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_144
export PATH=$PATH:$JAVA_HOME/bin

[root@hadoop05 software]# source /etc/profile
[root@hadoop05 software]# java -version

（2）Maven解压、配置 MAVEN_HOME和PATH

[root@hadoop05 software]# tar -zxvf apache-maven-3.0.5-bin.tar.gz -C /opt/module/
[root@hadoop05 software]# vi /etc/profile

#MAVEN_HOME
export MAVEN_HOME=/opt/module/apache-maven-3.0.5
export PATH=$PATH:$MAVEN_HOME/bin

[root@hadoop05 software]# source /etc/profile
[root@hadoop05 software]# mvn -version

（3）ant解压、配置 ANT _HOME和PATH

[root@hadoop05 software]# tar -zxvf apache-ant-1.9.9-bin.tar.gz -C /opt/module/
[root@hadoop05 software]# vi /etc/profile

#ANT_HOME
exportANT_HOME=/opt/module/apache-ant-1.9.9
export PATH=PATH:ANT_HOME/bin

[root@hadoop05 software]# source /etc//profile
[root@hadoop05 software]# ant -version

（4）安装glibc-headers 和 g++

[root@hadoop05 software]# yum install glibc-headers
[root@hadoop05 software]# yum install gcc-c++

（5）安装make和cmake

[root@hadoop05 software]# yum install make
[root@hadoop05 software]# yum install cmake

（6）解压protobuf ，进入到解压后protobuf主目录，/opt/module/protobuf-2.5.0，然后相继执行命令

[root@hadoop05 software]# tar -zxvf protobuf-2.5.0.tar.gz -C /opt/module/
[root@hadoop05 opt]# cd /opt/module/protobuf-2.5.0/
[root@hadoop05protobuf-2.5.0]# ./configure 
[root@hadoop05 protobuf-2.5.0]# make 
[root@hadoop05 protobuf-2.5.0]# make check 
[root@hadoop05 protobuf-2.5.0]# make install 
[root@hadoop05 protobuf-2.5.0]# ldconfig
[root@hadoop05 protobuf-2.5.0]# vi /etc/profile

#LD_LIBRARY_PATH
export LD_LIBRARY_PATH=/opt/module/protobuf-2.5.0
exportPATH=PATH:LD_LIBRARY_PATH

[root@hadoop05 protobuf-2.5.0]# source /etc/profile
[root@hadoop05 protobuf-2.5.0]# protoc --version

（7）安装openssl库

[root@hadoop05 protobuf-2.5.0]# yum install openssl-devel

（8）安装 ncurses-devel库

[root@hadoop05 protobuf-2.5.0]# yum install ncurses-devel

3.编译源码

（1）解压源码到/opt/目录

（2）进入到hadoop源码主目录

（3）通过maven执行编译命令

（4）成功的64位hadoop包在/opt/hadoop-2.7.2-src/hadoop-dist/target下

（5）编译源码过程中常见的问题及解决方案

Hadoop（HDFS）（存储）

一、概述

1.优缺点

（1）优点

高容错性
适合处理大数据
可构建在廉价机器上，通过多个副本机制，提高可靠性

（2）缺点

不适合低延时数据访问
无法高效的对大量小文件进行存储
- 仅支持数据的append(追加)

2.HDFS文件块大小（面试重点）

（1）默认大小在Hadoop2.x版本中是128M，老版本上是64M

（2）块大小的计算

如果寻址时间为10ms，即查找目标block的时间为10ms
寻址时间为传输时间的1%时为最佳状态，因此，传输时间=10ms/1%=1000ms=1s
磁盘目前的传输速率普遍为100MB/s
block大小=1s*100MB/s=100MB

总结：HDFS块的大小设置主要取决于磁盘传输速率

*二、Shell命令（开发重点）

1.基本语法

bin/hadoop fs 具体命令 OR bin/hdfs dfs 具体命令（这两个命令可互替换） dfs是fs的实现类。（）

2.命令大全

[admin@hadoop02 hadoop-2.7.2]$ hdfs dfs
Usage: hadoop fs [generic options]
	[-appendToFile <localsrc> ... <dst>]
	[-cat [-ignoreCrc] <src> ...]
	[-checksum <src> ...]
	[-chgrp [-R] GROUP PATH...]
	[-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]
	[-chown [-R] [OWNER][:[GROUP]] PATH...]
	[-copyFromLocal [-f] [-p] [-l] <localsrc> ... <dst>]
	[-copyToLocal [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
	[-count [-q] [-h] <path> ...]
	[-cp [-f] [-p | -p[topax]] <src> ... <dst>]
	[-createSnapshot <snapshotDir> [<snapshotName>]]
	[-deleteSnapshot <snapshotDir> <snapshotName>]
	[-df [-h] [<path> ...]]
	[-du [-s] [-h] <path> ...]
	[-expunge]
	[-find <path> ... <expression> ...]
	[-get [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
	[-getfacl [-R] <path>]
	[-getfattr [-R] {-n name | -d} [-e en] <path>]
	[-getmerge [-nl] <src> <localdst>]
	[-help [cmd ...]]
	[-ls [-d] [-h] [-R] [<path> ...]]
	[-mkdir [-p] <path> ...]
	[-moveFromLocal <localsrc> ... <dst>]
	[-moveToLocal <src> <localdst>]
	[-mv <src> ... <dst>]
	[-put [-f] [-p] [-l] <localsrc> ... <dst>]
	[-renameSnapshot <snapshotDir> <oldName> <newName>]
	[-rm [-f] [-r|-R] [-skipTrash] <src> ...]
	[-rmdir [--ignore-fail-on-non-empty] <dir> ...]
	[-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]
	[-setfattr {-n name [-v value] | -x name} <path>]
	[-setrep [-R] [-w] <rep> <path> ...]
	[-stat [format] <path> ...]
	[-tail [-f] <file>]
	[-test -[defsz] <path>]
	[-text [-ignoreCrc] <src> ...]
	[-touchz <path> ...]
	[-truncate [-w] <length> <path> ...]
	[-usage [cmd ...]]

3.常用命令实操（重）

（1）启动集群

[admin@hadoop02 hadoop-2.7.2]$ sbin/start-dfs.sh 
[admin@hadoop03 hadoop-2.7.2]$ sbin/start-yarn.sh

（2）-help：输出这个命令参数

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -help rm

（3）-ls: 显示目录信息

[admin@hadoop02 hadoop-2.7.2]$ hdfs dfs -ls /

（4）-mkdir：在HDFS上创建（多级）目录

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -mkdir -p /sanguo/weiguo

（5）-moveFromLocal：从本地剪切粘贴到HDFS

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -moveFromLocal ./panjinglian.txt /sanguo/weiguo/

（6）-appendToFile：追加一个文件到已经存在的文件末尾

[admin@hadoop02 hadoop-2.7.2]$ hdfs dfs -appendToFile ./liubei.txt /sanguo/weiguo/panjinglian.txt

（7）-cat：显示文件内容

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -cat /sanguo/weiguo/panjinglian.txt

（8）-chgrp 、-chmod、-chown：Linux文件系统中的用法一样，修改文件所属权限

[admin@hadoop02 hadoop-2.7.2] hadoop fs -chgrp admin /sanguo/weiguo/panjinglian.txt
[admin@hadoop02 hadoop-2.7.2] hadoop fs -chmod 666  /sanguo/weiguo/panjinglian.txt
[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -chown admin:admin  /sanguo/weiguo/panjinglian.txt

（9）-copyFromLocal：从本地文件系统中拷贝文件到HDFS路径去

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -copyFromLocal ./ximenqing.txt /sanguo/weiguo/

（10）-copyToLocal：从HDFS拷贝到本地

[admin@hadoop02 hadoop-2.7.2]$ hdfs dfs -copyToLocal /sanguo/weiguo/panjinglian.txt ./

（11）-cp ：从HDFS的一个路径拷贝到HDFS的另一个路径

[admin@hadoop02 hadoop-2.7.2]$ hdfs dfs -cp /sanguo/weiguo/panjinglian.txt /sanguo/

（12）-mv：在HDFS目录中移动文件

[admin@hadoop02 hadoop-2.7.2]$ hdfs dfs -mv /sanguo/panjinglian.txt /

（13）-get：等同于copyToLocal，就是从HDFS下载文件到本地

（14）-getmerge：合并下载多个文件

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -getmerge /sanguo/weiguo/* ./zaiyiqi.txt

（15）-put：等同于copyFromLocal

（16）-tail：显示一个文件的末尾

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -tail /sanguo/weiguo/LICENSE.txt

（17）-rm：删除文件或文件夹

（18）-rmdir：删除空目录

（19）-du：统计文件夹的大小信息

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -du /
	197657687  /hadoop-2.7.2.tar.gz
	29         /panjinglian.txt
	15474      /sanguo
	84         /wc.input

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -du -h /
	188.5 M  /hadoop-2.7.2.tar.gz
	29       /panjinglian.txt
	15.1 K   /sanguo
	84       /wc.input
[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -du -h -s /				#（看文件夹总大小）
	188.5 M  /

（20）-setrep：设置HDFS中文件的副本数量

[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -setrep 2 /panjinglian.txt

三、HDFS客户端操作（开发重点）

1.HDFS客户端环境准备

（1）根据自己电脑的操作系统拷贝对应的编译后的hadoop jar包到非中文路径

（2）配置HADOOP_HOME环境变量，配置Path环境变量
在这里插入图片描述

2.HDFS客户端环境

（1）创建一个Maven工程HDFS-0529

（2）导入相应的依赖坐标+日志添加

<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>RELEASE</version>
		</dependency>
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-core</artifactId>
			<version>2.8.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-common</artifactId>
			<version>2.7.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>2.7.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-hdfs</artifactId>
			<version>2.7.2</version>
		</dependency>
  
  		<!--使用IDEA不需要配置这个
		<dependency>
			<groupId>jdk.tools</groupId>
			<artifactId>jdk.tools</artifactId>
			<version>1.8</version>
			<scope>system</scope>
			<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
		</dependency> -->
</dependencies>

（3）创建日志文件log4j.properties

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（4）创建HDFSClient.java

public class HDFSClient {
    public static void main(String[] args) throws Exception{
        Configuration entries = new Configuration() 
        //1.获取hdfs客户端对象
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.在hdfs上创建路径
        fs.mkdirs(new Path("/0529/dashen/banzhang"));
        //3.关闭资源
        fs.close();
        System.out.println("执行结束");
    }
}

3.HDFS的API操作

（1）HDFS文件上传（测试参数优先级）

//1.文件上传
    @Test
    public void testCopyFromLocalFile()throws Exception{
        Configuration entries = new Configuration() ;
      	//设置副本数（这里的优先级最高）
        entries.set("dfs.replication", "2");
        //1.获取fs对象
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.执行上传API
        fs.copyFromLocalFile(new Path("F:/大数据/大数据知识/Hadoop/Code/banzhang.txt"), new Path("/bancao.txt"));
        //3.关闭资源
        fs.close();
    }

参数优先级：代码中>resources下配置文件>集群中>default默认3

（2）HDFS文件下载

//2.文件下载
    @Test
    public void testCopyToLocalFile() throws Exception {
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.执行下载API
//        fs.copyToLocalFile(new Path("/banhua.txt"), new Path("e:/banhua.txt"));
        //参数1：是否删除原数据（剪切），参数4：为true时不会产生.crc文件
        fs.copyToLocalFile(false, new Path("/banhua.txt"),new Path("e:/xioahua.txt"), true);
        //3.关闭资源
        fs.close();
    }

（3）HDFS文件夹删除

//3.文件的删除
    public void testDelete() throws Exception {
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.执行删除API
        //参数1：文件或者文件夹路径，参数2：当为文件夹是设置为true，表示递归
        fs.delete(new Path("/banhua.txt"),true);
        //3.关闭资源
        fs.close();
    }

（4）HDFS文件名更改

//4.更改文件名
    @Test
    public void testRename() throws Exception {
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.执行更名API
        fs.rename(new Path("/banzhang.txt"), new Path("/banhua.txt"));
        //3.关闭资源
        fs.close();
    }

（5）HDFS文件详情查看

//5.查看文件详情
    @Test
    public void testListFiles() throws Exception {
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.执行查看详情API
        //参数2：是否递归查看
        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.getPermission());
            System.out.println("长度："+fileStatus.getLen());
            //块信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();//副本
            for (BlockLocation blockLocation:blockLocations) {
                String[] hosts = blockLocation.getHosts();//具体存在哪些主机上
                for (String host:hosts) {
                    System.out.println("主机："+host);
                }
            }
            System.out.println("----------------------------------");
        }
        //3.关闭资源
        fs.close();
    }

（6）HDFS文件和文件夹判断

//6.判断文件还是文件夹
    @Test
    public void testListStatus() throws Exception {
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.执行判断状态API
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus:fileStatuses) {
            if(fileStatus.isFile()){
                System.out.println("f:"+fileStatus.getPath().getName()+"是文件");
            }else {
                System.out.println("d:"+fileStatus.getPath().getName()+"是文件夹");
            }
        }
        //3.关闭资源
        fs.close();
    }

4.HDFS的I/O流操作

（1） HDFS文件上传

//把本地e盘上的banhua.txt文件上传到HDFS根目录
    @Test
    public void putFileToHDFS() throws Exception {
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.获取输入流
        FileInputStream fis = new FileInputStream(new File("e:/banhua.txt"));
        //3.获取输出流
        FSDataOutputStream fos = fs.create(new Path("/banzhang.txt"));
        //4.流的对拷
        IOUtils.copyBytes(fis, fos, entries);
        //5.关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }

（2）HDFS文件下载

//从HDFS上下载banhua.txt文件到本地e盘上
    @Test
    public void getFileFromHDFS() throws Exception{
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.获取输入流
        FSDataInputStream fis = fs.open(new Path("/bancao.txt"));
        //3.获取输出流
        FileOutputStream fos = new FileOutputStream(new File("e:/bancao.txt"));
        //4.流的对拷
        IOUtils.copyBytes(fis, fos, entries);
        //5.关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }

（3）定位文件读取

//分块读取HDFS上的大文件，比如根目录下的/hadoop-2.7.2.tar.gz
    //下载第一块
    @Test
    public void readFileSeek1() throws Exception{
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.获取输入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
        //3.获取输出流
        FileOutputStream fos = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part1"));
        //4.流的对拷(只拷贝128M)
        byte[] buf = new byte[1024];
        for (int i = 0; i <1024*128 ; i++) {
            fis.read(buf);
            fos.write(buf);
        }
        //5.关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }
	//下载第二块
    @Test
    public void readFileSeek2() throws Exception{
        //1.获取fs对象
        Configuration entries = new Configuration() ;
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop02:9000"), entries, "admin");
        //2.获取输入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
        //3.设置指定读取的起点
        fis.seek(1024*1024*128);
        //4.获取输出流
        FileOutputStream fos = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part2"));
        //5.流的对拷
        IOUtils.copyBytes(fis, fos, entries);
        //6.关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }

合并文件

在Window命令窗口中进入到目录E:，然后执行如下命令，对数据进行合并
- type hadoop-2.7.2.tar.gz.part2 >>hadoop-2.7.2.tar.gz.part1
合并完成后，将hadoop-2.7.2.tar.gz.part1重新命名为hadoop-2.7.2.tar.gz。解压发现该tar包非常完整。

四、HDFS的数据流（面试重点）

1.客户端上传文件

上传到哪个DataNode，根据两个条件：

（1）节点距离客户端距离近

（3）节点负载小

2.计算节点距离

节点距离：两个节点到达最近的共同祖先的距离总和

3.副本节点选择

第一个副本在距离最近的机架上随机一个节点上

第二个副本与第一个副本位于相同机架，随机节点

第三个副本位于不同机架，随机节点

五、NameNode和SecondaryNameNode（面试开发重点）

1.NN和2NN工作机制

（1）问题和解决方法

NameNode中的元数据是存储在内存中（一关电就没了）
在磁盘中备份元数据的FsImage（发生一致性问题）
每当元数据有更新或者添加元数据时，修改内存中的元数据并追加到Edits中（Edits文件过大）
通过FsImage和Edits的合并，合成元数据（文件过大，合成效率低）
新的节点SecondaryNamenode，专门用于FsImage和Edits的合并

（2）NN和2NN工作机制

在这里插入图片描述

2NN的作用就是将fsimage和edits_001合并成新的Fsimage

2.Fsimage和Edits解析

（1）概念
在这里插入图片描述

（2）oiv查看Fsimage文件

hdfs oiv -p 文件类型 -i镜像文件 -o 转换后文件输出路径

（3）oev查看Edits文件

hdfs oev -p 文件类型 -i编辑日志 -o 转换后文件输出路径

（4）Fsimage中没有记录块所对应DataNode，为什么？

在集群启动后，要求DataNode上报数据块信息，并间隔一段时间后再次上报

（5）NameNode如何确定下次开机启动的时候合并哪些Edits？

根据seen_txid里面记录的最新的

3.CheckPoint时间设置

（1）通常情况下，SecondaryNameNode每隔一小时执行一次。[hdfs-default.xml]

（2）一分钟检查一次操作次数，当操作次数达到1百万时，SecondaryNameNode执行一次。

4.NameNode故障处理

（1）方法一：将SecondaryNameNode中数据拷贝到NameNode存储数据的目录

（2）方法二：使用-importCheckpoint选项启动NameNode守护进程，从而将SecondaryNameNode中数据拷贝到NameNode目录中

5.集群安全模式

（1）概述

1）NameNode启动

NameNode启动时，当NameNode开始监视DataNode请求时，NameNode一直运行在安全模式，即NameNode的文件系统对于客户端来说是只读的

2）DataNode启动

系统中的数据块的位置并不是有NameNode维护的，而是以块列表的形式存储在DataNode中

3）安全模式退出判断

满足最小副本条件，NameNode会在30秒之后退出安全模式

最小副本条件：在整个文件系统中99.9%的块满足最小副本级别。在启动一个刚刚格式化的HDFS集群时，因为系统中还没有任何块，所有NameNode不会进入安全模式

在hdfs-site.xml中设置安全阀值属性，属性值默认为0.999f，如果设为1则不进行安全检查

<property>
	<name>dfs.safemode.threshold.pct</name>
	<value>0.999f</value>
</property>

（2）基本语法

bin/hdfs dfsadmin -safemode get （功能描述：查看安全模式状态）
bin/hdfs dfsadmin -safemode enter （功能描述：进入安全模式状态）
bin/hdfs dfsadmin -safemode leave （功能描述：离开安全模式状态）
bin/hdfs dfsadmin -safemode wait （功能描述：等待安全模式状态）

六、DataNode（面试开发重点）

1.DataNode工作机制

在这里插入图片描述

2.掉线时限参数设置

HDFS默认超时时长是10分钟+30秒

3.服役新数据节点

（1）环境准备

1）在hadoop104主机上再克隆一台hadoop105主机

2）修改IP地址和主机名称

3）删除原来HDFS文件系统留存的文件（/opt/module/hadoop-2.7.2/data和log）

4）source一下配置文件

（2）具体步骤

1）直接启动DataNode，即可关联到集群

[admin@hadoop05 hadoop-2.7.2]$ sbin/hadoop-daemon.sh start datanode
[admin@hadoop05 hadoop-2.7.2]$ sbin/yarn-daemon.sh start nodemanager

2）在hadoop05上上传文件

[admin@hadoop05 hadoop-2.7.2]$ hadoop fs -mkdir -p /user/admin/input
[admin@hadoop05 hadoop-2.7.2]$ hadoop fs -put wcinput/wc.input /user/admin/input

3）如果数据不均衡，可以用命令实现集群的再平衡

[admin@hadoop02 sbin]$ ./start-balancer.sh

4.退役旧数据节点

（1）添加白名单

添加到白名单的主机节点，才允许访问NameNode

配置白名单具体步骤：

1）在NameNode的/opt/module/hadoop-2.7.2/etc/hadoop目录下创建dfs.hosts文件

[admin@hadoop05 hadoop]$ touch dfs.hosts

2）在NameNode的hdfs-site.xml配置文件中增加dfs.hosts属性

[admin@hadoop05 hadoop]$ vi hdfs-site.xml

<property>
	<name>dfs.hosts</name>
	<value>/opt/module/hadoop-2.7.2/etc/hadoop/dfs.hosts</value>
</property>

3）配置文件分发

[admin@hadoop02 hadoop]$ xsync hdfs-site.xml

4）刷新NameNode

[admin@hadoop02 hadoop]$ hdfs dfsadmin -refreshNodes

5）更新ResourceManager节点

[admin@hadoop02 hadoop]$ yarn rmadmin -refreshNodes

6）在web浏览器上查看
在这里插入图片描述

7）如果数据不均衡，可以用命令实现集群的再平衡

[admin@hadoop02 sbin]$ ./start-balancer.sh

（2）黑名单退役

在黑名单上的主机会被强制退出

1）在NameNode的/opt/module/hadoop-2.7.2/etc/hadoop目录下创建dfs.hosts.exclude文件

[admin@hadoop02 hadoop]$ vi dfs.hosts.exclude

2）在NameNode的hdfs-site.xml配置文件中增加dfs.hosts.exclude属性，并分发文件

<property>
	<name>dfs.hosts.exclude</name>
	<value>/opt/module/hadoop-2.7.2/etc/hadoop/dfs.hosts.exclude</value>
</property>

[admin@hadoop02 hadoop]$ xsync hdfs-site.xml

3）刷新NameNode、刷新ResourceManager

[admin@hadoop02 hadoop]$ hdfs dfsadmin -refreshNodes

[admin@hadoop02 hadoop]$ yarn rmadmin -refreshNodes

4）检查Web浏览器，退役节点的状态为decommission in progress（退役中），说明数据节点正在复制块到其他节点
在这里插入图片描述

5）等待退役节点状态为decommissioned（所有块已经复制完成），停止该节点及节点资源管理器

注意：如果副本数是3，服役的节点小于等于3，是不能退役成功的，需要修改副本数后才能退役

在这里插入图片描述

6）单节点退出


	[admin@hadoop05 hadoop-2.7.2]$ sbin/hadoop-daemon.sh stop datanode

	[admin@hadoop05 hadoop-2.7.2]$ sbin/yarn-daemon.sh stop nodemanager

7）如果数据不均衡，可以用命令实现集群的再平衡

	[admin@hadoop02 sbin]$ ./start-balancer.sh

注意：不允许白名单和黑名单中同时出现同一个主机名称。

七、HDFS 2.X新特性

1.小文件存档

（1）意义

大量的小文件会耗尽NameNode中的大部分内存

小文件归档：HDFS存档文件对内还是一个一个独立文件，对NameNode而言却是一个整体，减少了NameNode的内存占用

（2）案例实操

1）需要启动YARN进程

	[admin@hadoop03 hadoop-2.7.2]$ start-yarn.sh

2）归档文件

把/user/admin/input目录里面的所有文件归档成一个叫input.har的归档文件，并把归档后文件存储到/user/admin/output路径下。

	[admin@hadoop02 hadoop-2.7.2]$ bin/hadoop archive -archiveName input.har -p /user/admin/input/ /user/admin/output/

3）查看归档

	[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -ls -R har:///user/admin/output/input.har

4）解归档文件

	[admin@hadoop02 hadoop-2.7.2]$ hadoop fs -cp har:/// user/admin/output/input.har/* /user/admin

Hadoop(MapReduce)（计算）

一、MapReduce概述

1.定义

一个分布式运算程序的编程框架

核心功能：将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序

2.优缺点

（1）优点

易于编程
良好的扩展性
高容错性
适合PB级以上海量数据的离线处理

（2）缺点

不擅长实时计算

不擅长流式计算

不擅长DAG（有向图）计算

3.核心思想

Map阶段：分

Map阶段的并发MapTask，完全并行运行，互不相干

Reduce阶段：合

Reduce阶段的并发ReduceTask，完全互不相干，数据依赖于Map阶段的所有MapTask并发输出的结果

注意：一个MapReduce模型只能有一个Map和一个Reduce阶段，如果需要多个，只能多个MapReduce串行运行

4.MapReduce进程

三类实例进程：

（1）MrAppMaster：负责整个程序的过程调度及状态协调

（2）MapTask：负责Map阶段整个数据处理流程

（3）ReduceTask：负责Reduce阶段的整个数据处理流程

5.常用数据序列化类型

Java类型	Hadoop Writable类型
boolean	BooleanWritable
byte	ByteWritable
int	IntWritable
float	FloatWritable
long	LongWritable
double	DoubleWritable
String	Text
map	MapWritable
array	ArrayWritable

6.MapReduce编程规范（重）

分成三个部分：Mapper、Reducer和Driver。

（1）Mapper阶段

用户自定义的Mapper要继承自己的父类
Mapper的输入数据是KV对的形式（KV的类型可自定义）
Mapper中的业务逻辑写在map()方法中
Mapper的输出数据是KV对的形式（KV的类型可自定义）
map()方法（MapTask进程）对每个<K,V>调用一次

（2）Reducer阶段

用户自定义的Reducer要继承自己的父类
Reducer的输入类型对应Mapper的输出数据类型，也是KV
Reducer的业务逻辑写在reduce()方法中
ReduceTask进程对每一组相同k的<k,v>组调用一次reduce()方法

（3）Driver阶段

相当于YARN集群的客户端
用于提交整个程序到集群
提交的是封装了MapReduce程序相关运行参数的job对象

*7.WordCount案例实操（wordcount）

（1）需求分析

在这里插入图片描述

（2）环境准备

1）创建maven工程

2）导入pom依赖

<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>RELEASE</version>
		</dependency>
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-core</artifactId>
			<version>2.8.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-common</artifactId>
			<version>2.7.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>2.7.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-hdfs</artifactId>
			<version>2.7.2</version>
		</dependency>
</dependencies>

3）src/main/resources目录下，创建log4j.properties文件

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（3）编写程序

1）编写Mapper类

/**
 * @author hxc
 * @create 2020-04-08 17:34
 *
 * map阶段,继承Mapper
 * 参数1：输入数据key的类型
 * 参数2：输入数据value的类型
 * 参数3：输出数据key的类型
 * 参数4：输出数据value的类型
 */
public class WordcountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {

    Text k = new Text();
    IntWritable v = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        /*atguigu atguigu
        ss ss
        cls cls
        jiao
        banzhang
        xue
        hadoop*/
        //1.获取一行atguigu atguigu,转化成String类型
        String line = value.toString();
        //2.切割单词,按空格切
        String[] words = line.split(" ");
        //3.循环写出
        for (String word : words) {
            //atguigu
            k.set(word);
            context.write(k, v);
        }
    }
}

2）编写Reducer类

/**
 * @author hxc
 * @create 2020-04-08 18:18
 * reduce阶段，继承Reducer
 * 参数1,2：map阶段输出的key和value类型
 * 参数3,4：reduce阶段输出的key和value类型
 */
public class WordcountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
    IntWritable v = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //atguigu,1
        //atguigu,1
        //1.累加求和
        int sum=0;
        for (IntWritable value : values) {
            sum+=value.get();
        }
        //2.写出:atguigu,2
        v.set(sum);
        context.write(key, v);
    }
}

3）编写Driver驱动类

public class WordcountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1.获取Job对象
        Configuration conf=new Configuration();
        Job job = Job.getInstance(conf);

        //2.存储jar存储位置
        job.setJarByClass(WordcountDriver.class);

        //3.关联Map和Reduce类
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);

        //4.设置Mapper阶段输出数据的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5.设置最终数据输出key和value类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6.设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //7.提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

（4）本地测试

1）设置参数
在这里插入图片描述

2）运行程序

注意：output文件在运行前不存在

（5）集群测试

1）将程序打成jar包
在这里插入图片描述

2）拷贝到Hadoop集群（拷贝到/opt/module/hadoop-2.7.2文件夹下）

3）启动Hadoop集群

4）执行WordCount程序

[admin@hadoop02 hadoop-2.7.2]$ hadoop jar wc.jar com.ithxc.mr.wordcount.WordcountDriver /user/admin/input /user/admin/output

注意：output文件夹在执行程序之前不可已存在

二、Hadoop序列化

1.序列化概述

（1）什么是序列化

把内存中的对象（数组、集合等），转化成字节序列

（2）Hadoop序列化特点

紧凑：高效使用存储空间
快速：读写数据的额外开销小
可扩展：随着通信协议的升级而可升级
互操作：支持多语言的交互

2.自定义bean对象实现序列化接口（Writable）（重）

（1）具体实现bean对象序列化

1）必须实现Writable接口

2）反序列化时，需要反射调用空参构造函数，所以必须有空参构造

3）重写序列化方法

4）重写反序列化方法

5）注意反序列化的顺序和序列化的顺序完全一致

7）要想把结果显示在文件中，需要重写toString()，可用”t”分开，方便后续用

8）如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口，因为MapReduce框中的Shuffle过程要求对key必须能排序

*（2）序列化案例（flowsum）

1）实现bean对象序列化

//实现writable接口
public class FlowBean implements Writable {

    private long upFlow;//上行流量
    private long downFlow;//下行流量
    private long sumFlow;//总流量

    //反序列化时，需要反射调用空参构造函数，所以必须有空参构造器
    public FlowBean() {
    }
    public FlowBean(long upFlow, long downFlow, long sumFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        sumFlow = upFlow + downFlow;
    }

    //序列化方法
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(upFlow);
        dataOutput.writeLong(downFlow);
        dataOutput.writeLong(sumFlow);
    }

    //反序列化方法
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        //必须要求和序列化方法顺序一致
        upFlow = dataInput.readLong();
        downFlow = dataInput.readLong();
        sumFlow = dataInput.readLong();
    }

    @Override
    public String toString() {
        return  upFlow + "t" + downFlow + "t" + sumFlow ;
    }
  
  //计算sumFlow
  public void set(long upFlow2,long downFlow2){
        upFlow=upFlow2;
        downFlow=downFlow2;
        sumFlow=upFlow2+downFlow2;
    }
 // 此处get和set方法省略

2）编写Mapper类

public class FlowCountMapper extends Mapper<LongWritable,Text, Text,FlowBean> {
    Text k=new Text();
    FlowBean v=new FlowBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1.获取一行
        String line = value.toString();
        //2.切割 /t
        String[] fields = line.split("t");
        //3.封装对象
        k.set(fields[1]);//封装手机号

        long upFlow = Long.parseLong(fields[fields.length-3]);//封装上行流量
        long downFlow = Long.parseLong(fields[fields.length-2]);//封装下行流量

        v.setUpFlow(upFlow);
        v.setDownFlow(downFlow);
        //4.写出
        context.write(k, v);
    }
}

3）编写Reducer类

public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> {
    FlowBean v=new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        long sum_upFlow = 0;
        long sum_downFlow = 0;
        //1.累加求和
        for (FlowBean flowBean:values) {
            sum_upFlow += flowBean.getUpFlow();
            sum_downFlow += flowBean.getDownFlow();
        }
        v.set(sum_upFlow,sum_downFlow);
        //2.写出
        context.write(key, v);
    }
}

4）编写Driver

public class FlowsumDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        args = new String[]{"f:/input/inputflow","f:/output1"};
        //1.获取job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        //2.设置jar的路径
        job.setJarByClass(FlowsumDriver.class);

        //3.关联mapper和reducer
        job.setMapperClass(FlowCountMapper.class);
        job.setReducerClass(FlowCountReducer.class);

        //4.设置mapper输出的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //5.设置最终输出的key和value类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        //6.设置输入输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //7.提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

三、MapReduce框架原理（重）

1、InputFormat数据输入

（1）切片与MapTask并行度决定机制

**数据块：**Block是HDFS物理上把数据分成一块一块。

**数据切片：**数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。

1）一个Job的Map阶段并行度由客户端在提交Job时的切片数决定

2）每一个切片分配一个MapTask并行实例处理

3）默认情况下，切片大小=BlockSize（块大小：集群128M，本地32M）但是129M切一块

4）切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

（2）Job提交流程源码和切片源码详解

waitForCompletion()

submit();

// 1建立连接
	connect();	
		// 1）创建提交Job的代理
		new Cluster(getConfiguration());
			// （1）判断是本地yarn还是远程
			initialize(jobTrackAddr, conf); 

// 2 提交job
submitter.submitJobInternal(Job.this, cluster)
	// 1）创建给集群提交数据的Stag路径
	Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

	// 2）获取jobid ，并创建Job路径
	JobID jobId = submitClient.getNewJobID();

	// 3）拷贝jar包到集群
copyAndConfigureFiles(job, submitJobDir);	
	rUploader.uploadFiles(job, jobSubmitDir);

// 4）计算切片，生成切片规划文件
writeSplits(job, submitJobDir);
		maps = writeNewSplits(job, jobSubmitDir);
		input.getSplits(job);

// 5）向Stag路径写XML配置文件
writeConf(conf, submitJobFile);
	conf.writeXml(out);

// 6）提交Job,返回提交状态
status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

在这里插入图片描述

（3）FileInputFormat切片机制（默认）

1）切片机制
在这里插入图片描述

2）切片大小的参数配置
在这里插入图片描述

*（4）CombineTextInputFormat切片机制（小文件多）

1）切片机制
在这里插入图片描述

2）案例实操（开发中一般设置成128M）

在WordcountDriver（Driver类）中增加如下代码

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

//虚拟存储切片最大值设置4m
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

或者

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

//虚拟存储切片最大值设置20m
CombineTextInputFormat.setMaxInputSplitSize(job,20971520);

（5）FileInputFormat实现类

1）常见接口实现类

TextInputFormat(默认)
KeyValueTextInputFormat
NLineInputFormat
CombineTextInputFormat
自定义InputFormat

2）TextInputFormat(默认)

按行读取每条记录
键是存储该行在整个文件中的起始字节偏移量，LongWritable类型
值是这行的内容，Text类型
eg:

{0,atguigu atguigu hxc hxc}
{25,atguigu atguigu hxc}

3）KeyValueTextInputFormat

每一行作为一条记录
键是该行中第一次出现切割符之前的内容
值是该行中第一次出现切割符之后的内容
默认切割符为t,设置方式，在驱动类中加入如下代码

//设置切割符，默认是/t，这里设置成空格
conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");

需要在设置输出数据路径前添加设置输入格式

// 设置输入格式
job.setInputFormatClass(KeyValueTextInputFormat.class);
// 6 设置输出数据路径

eg：以空格为切割符

输入： atguigu atguigu hxc hxc
atguigu atguigu hxc

转换： {atguigu,atguigu hxc hxc}
{atguigu,atguigu hxc}

4）NLineInputFormat

每个map进程InputSplit不再按Block块来划分，而是按NlineInputFormat指定的行数N来划分
输入的文件总行数/N=切片数，不能整除时，切片数=商+1
键和值TextInputFormat是一样的
需要在驱动类中加入两行代码，设置jar包位置前面

// 设置每个切片InputSplit中划分三条记录
NLineInputFormat.setNumLinesPerSplit(job, 3);
          
// 使用NLineInputFormat处理记录数  
job.setInputFormatClass(NLineInputFormat.class);
// 2设置jar包位置，关联mapper和reducer

5）自定义InputFormat（合并小文件）（inputformat）

使用场景：合并小文件
步骤：

a.自定义一个类继承FileInputFormat

b.改写RecordReader，实现一次读取一个完整的文件封装为KV

c.在输出时使用SequenceFileOutPutFormat输出合并文件
键是每个小文件的名称
值是小文件的内容

案例：

需求：将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value。

分析：
在这里插入图片描述

1、自定义InputFromat

public class WholeFileInputformat extends FileInputFormat<Text, BytesWritable> {
    @Override
    public RecordReader createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        WholeRecordReader recordReader=new WholeRecordReader();
        recordReader.initialize(split, context);
        return recordReader;
    }
}

2、自定义RecordReader类

public class WholeRecordReader extends RecordReader<Text, BytesWritable> {
    FileSplit split;
    Configuration configuration;

    Text k=new Text();
    BytesWritable v=new BytesWritable();
    //每一次创建对象时，为true
    boolean isProgress=true;

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        //初始化
        this.split= (FileSplit) split;

        //获取配置信息
        configuration = context.getConfiguration();
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        //核心业务逻辑处理
        if(isProgress){
            byte[] buf=new byte[(int) split.getLength()];
            //1.获取fs对象
            Path path = split.getPath();
            FileSystem fs = path.getFileSystem(configuration);

            //2.获取输入流
            FSDataInputStream fis = fs.open(path);

            //3.拷贝
            IOUtils.readFully(fis,buf,0,buf.length);

            //4.封装v
            v.set(buf,0,buf.length);

            //5.封装k
            k.set(path.toString());

            //6.关闭资源
            IOUtils.closeStream(fis);

            isProgress=false;
            return true;
        }
        return false;
    }

    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        //获取输出的key
        return k;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        //获取输出的value
        return v;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {

    }
}

3、编写SequenceFileMapper类处理流程

public class SequenceFileMapper extends Mapper<Text, BytesWritable,Text, BytesWritable> {
    @Override
    protected void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException {
        context.write(key, value);
    }
}

4、编写SequenceFileReducer类处理流程

public class SequenceFileReducer extends Reducer<Text, BytesWritable,Text, BytesWritable> {
    @Override
    protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {
        //循环写出
        for (BytesWritable value:values) {
            context.write(key, value);
        }
    }
}

5、编写SequenceFileDriver类处理流程

public class SequenceFileDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
        args = new String[] { "f:/input/inputformat", "f:/output1" };

        // 1 获取job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 2 设置jar包存储位置、关联自定义的mapper和reducer
        job.setJarByClass(SequenceFileDriver.class);
        job.setMapperClass(SequenceFileMapper.class);
        job.setReducerClass(SequenceFileReducer.class);

        // 设置输入的inputFormat
        job.setInputFormatClass(WholeFileInputformat.class);

        // 设置输出的outputFormat
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

        // 3 设置map输出端的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(BytesWritable.class);

        // 4 设置最终输出端的kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BytesWritable.class);

        // 5 设置输入输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 6 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

2、MapReduce工作流程

在这里插入图片描述

3.Shuffle机制

在这里插入图片描述

*（1）Partition分区（flowsum）

1）默认Partition分区

默认分区是根据key的hashCode对ReduceTasks个数取模得到，用户没法控制哪个key存储到哪个分区

public int getPartition(K2 key, V2 value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

2）自定义Partition步骤

自定义类继承Partitioner,重写getPartition()方法
在Job驱动中，设置自定义Partitioner

public class ProvincePartitioner extends Partitioner<Text,FlowBean> {

    @Override
    public int getPartition(Text key, FlowBean value, int numPartitions) {
        //key是手机号
        //value是流量信息

        //获取手机号前三位
        String prePhoneNum = key.toString().substring(0, 3);
      	//分区数量为5:0-4
        int partition=4;
        if("136".equals(prePhoneNum)){
            partition=0;
        }else if ("137".equals(prePhoneNum)){
            partition=1;
        }else if ("138".equals(prePhoneNum)){
            partition=2;
        }else if ("139".equals(prePhoneNum)){
            partition=3;
        }
        return partition;
    }
}

自定义Partition后，要根据自定义Partitioner的逻辑设置相应数量的ReduceTask

//关联Partitioner
job.setPartitionerClass(ProvincePartitioner.class);
//设置ReduceTasks数量
job.setNumReduceTasks(5);

*（2）WritableComparable排序

1）排序概述

MapTesk和ReduceTask会对数据按照key进行排序
任何应用程序中的数据都会被排序，不管逻辑上是否需要
默认排序是按照字典顺序，实现方法是快速排序
在Map阶段至少要对key进行一次快速排序，对排序后溢写到磁盘的小文件进行一次归并排序合成一个大文件
在Reduce阶段，当所有数据拷贝完成，统一对内存和磁盘上的数据进行一次归并排序

2）排序的分类

部分排序
- 输出的每个文件内部有序
全排序
- 最终输出结果只有一个文件，且文件内部有序
辅助排序（GroupingComparator分组）
- 在Reduce端对key进行分组
二次排序
- 在自定义排序过程中，如果conpareTo中的判断条件为两个即为二次排序

*3）自定义排序WritableComparable（sort）

原理：bean对象作为key传输，需要实现WritableComparable接口重写conpareTo方法，就可以实现排序

案例1(全排序)：

FlowBean.java

public class FlowBean implements WritableComparable<FlowBean> {

    private long upFlow;    //上行流量
    private long downFlow;  //下行流量
    private long sumFlow;   //总流量

    public FlowBean() {
    }

    public FlowBean(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        sumFlow=upFlow+downFlow;
    }

    //比较
    @Override
    public int compareTo(FlowBean bean) {

        int result;
        //核心比较判断语句
        if(sumFlow>bean.getSumFlow()){
            result=-1;
        }else if(sumFlow<bean.getSumFlow()){
            result=1;
        }else {
            result=0;
        }
        return result;
    }

    //序列化
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(upFlow);
        dataOutput.writeLong(downFlow);
        dataOutput.writeLong(sumFlow);
    }

    //反序列化
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        upFlow=dataInput.readLong();
        downFlow=dataInput.readLong();
        sumFlow=dataInput.readLong();
    }
    
    @Override
    public String toString() {
        return upFlow + "t" + downFlow + "t" + sumFlow;
    }
    //省略get和set方法

FlowCountSortMapper.java

public class FlowCountSortMapper extends Mapper<LongWritable, Text,FlowBean,Text> {
    FlowBean k = new FlowBean();
    Text v = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //1.获取1行
        String line = value.toString();

        //2.切割
        String[] fields = line.split("t");

        //3.封装对象
        String phoneNum = fields[0];
        long upFlow = Long.parseLong(fields[1]);
        long downFlow = Long.parseLong(fields[2]);
        long sumFlow = Long.parseLong(fields[3]);

        k.setUpFlow(upFlow);
        k.setDownFlow(downFlow);
        k.setSumFlow(sumFlow);

        v.set(phoneNum);

        //4.写出
        context.write(k, v);
    }
}

FlowCountSortReducer.java

public class FlowCountSortReducer extends Reducer<FlowBean, Text,Text,FlowBean> {
    @Override
    protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for(Text value:values){
            context.write(value, key );
        }
    }
}

FlowCountSortDriver.java

public class FlowCountSortDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
        args = new String[]{"f:/output1","f:/output2"};

        // 1 获取配置信息，或者job对象实例
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        // 2 指定本程序的jar包所在的本地路径
        job.setJarByClass(FlowCountSortDriver.class);

        // 3 指定本业务job要使用的mapper/Reducer业务类
        job.setMapperClass(FlowCountSortMapper.class);
        job.setReducerClass(FlowCountSortReducer.class);

        // 4 指定mapper输出数据的kv类型
        job.setMapOutputKeyClass(FlowBean.class);
        job.setMapOutputValueClass(Text.class);

        // 5 指定最终输出的数据的kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        // 6 指定job的输入原始文件所在目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);

    }
}

案例2（区内排序）

分为多文件输出，每个输出都进行排序

添加一个ProvincePartitioner.java

public class ProvincePartitioner extends Partitioner<FlowBean, Text> {
    @Override
    public int getPartition(FlowBean key, Text value, int numPartitions) {

        //按照手机号的前三位分区
        String prePhoneNum = value.toString().substring(0, 3);

        int partition=4;
        if("136".equals(prePhoneNum)){
            partition=0;
        }else if ("137".equals(prePhoneNum)){
            partition=1;
        }else if ("138".equals(prePhoneNum)){
            partition=2;
        }else if ("139".equals(prePhoneNum)){
            partition=3;
        }

        return partition;
    }
}

在FlowCountSortDriver.java中添加如下代码

//关联分区
job.setPartitionerClass(ProvincePartitioner.class);
//设置NRTasks数量
job.setNumReduceTasks(5);

（3）Combiner合并

1）概念

Combiner的父类是Reducer

Combiner是在每一个MapTask所在的节点运行

其意义在于对每一个MapTask的输出进行局部汇总，减少网络IO

Combiner能够应用的前提是不能影响最终的业务逻辑（求平均值的不行，适合累加求和）

Combiner的输出kv要对应Reducer的输入kv

2）自定义Combiner实现

自定义一个Combiner继承Reducer，重写Reduce方法
在Job驱动类中设置关联（可直接将Reduce方法作为Combiner在驱动类中指定）

job.setCombinerClass(WordcountReducer.class);

（4）GroupingComparator分组（辅助排序）（order）

1）作用

对Reduce阶段的数据根据某一个或几个字段进行分组。

2）步骤

a）自定义类继承WritableComparator

public class OrderGroupingComparator extends WritableComparator {
   
}

b）重写compare()方法

 @Override
    public int compare(WritableComparable a, WritableComparable b) {
        // 比较的业务逻辑
		return result;
    }

c）创建一个构造将比较对象的类传给父类

protected OrderGroupingComparator() {
        super(OrderBean.class,true);
 }

d）在job驱动中关联分组

//设置reduce端的分组
job.setGroupingComparatorClass(OrderGroupingComparator.class);

4.MapTask工作机制

在这里插入图片描述

5.ReduceTask工作机制

（1）图解
在这里插入图片描述

（2）设置ReduceTask并行度（个数）

MapTask并发数由切片数决定，ReduceTask并发数可以手动设定

//默认为1，设置为5
job.setNumReduceTasks(5);

如果设置了分区数，ReduceTask个数一定与分区数相等

6.OutputFormat数据输出

（1）接口实现类

1）文本输出TextOutputFormat（系统默认）

2）SequenceFileOutputFormat

将SequenceFileOutputFormat输出作为后续MapReduce任务的输入，它格式紧凑，很容易被压缩

3）自定义OutputFormat

控制最终文件的输出路径和输出格式

*（2）自定义OutputFormat步骤（outputformat）

1）自定义一个类继承FileOutputFormat

public class FilterOutputFormat extends FileOutputFormat<Text, NullWritable> {
    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {

        // 创建一个RecordWriter
        return new FRecordWriter(job);
    }
}

2）改写RecordWriter，具体改写输出数据的方法write()

public class FRecordWriter extends RecordWriter<Text, NullWritable> {

    FSDataOutputStream fosatguigu;
    FSDataOutputStream fosother;
    public FRecordWriter(TaskAttemptContext job) {

        try {
            //1.获取文件系统
            FileSystem fs = FileSystem.get(job.getConfiguration());

            //2.创建输出到atguigu.log的输出流
            fosatguigu = fs.create(new Path("f:/atguigu.log"));
            //3.创建输出到other.log的输出流
            fosother = fs.create(new Path("f:/other.log"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {

        //主要业务逻辑写在这里面
        //判断key中有没有atguigu，如果有写出到atguigu.log,如果没有写出到other.log
        if(key.toString().contains("atguigu")){
            fosatguigu.write(key.toString().getBytes());
        }else {
            fosother.write(key.toString().getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {

        //关闭资源
        IOUtils.closeStream(fosatguigu);
        IOUtils.closeStream(fosother);
    }
}

7.Join多种应用

（1）Reduce Join

1）工作原理

Map端：对来与不同表或文件的kv对打标签以区分不同的来源记录，然后用连接字段作为key，其余部分和新加的标志作为value

Reduce端：将来源于不同文件的记录（在Map阶段已经打标签）分开，最后进行合并

2）缺点

合并的操作是在Reduce阶段完成的，Reduce端的处理压力太大，Map节点的运算负载很低，资源利用率不高，且在Reduce阶段极易产生数据倾斜

3）案例

TableBean.java

public class TableBean implements Writable {
    private String id;  //订单id
    private String pid; //产品id
    private int amount; //数量
    private String pname;   //产品名称
    private String flag;    //标记
  
    public TableBean() {
    }
    public TableBean(String id, String pid, int amount, String pname, String flag) {
        this.id = id;
        this.pid = pid;
        this.amount = amount;
        this.pname = pname;
        this.flag = flag;
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {

        //序列化方法
        dataOutput.writeUTF(id);
        dataOutput.writeUTF(pid);
        dataOutput.writeInt(amount);
        dataOutput.writeUTF(pname);
        dataOutput.writeUTF(flag);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {

        //反序列化方法
        id = dataInput.readUTF();
        pid = dataInput.readUTF();
        amount = dataInput.readInt();
        pname = dataInput.readUTF();
        flag = dataInput.readUTF();
    }
    @Override
    public String toString() {
        return id + "t"+ amount + "t" + pname;
    }
    //此处省略get和set方法

TableMapper.java

public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {

    String name;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {

        //获取文件名称
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        name = inputSplit.getPath().getName();

    }

    TableBean v = new TableBean();
    Text k = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

//        id	pid	amount
//        1001	01	1
//
//        pid	pname
//        01	小米

        //1.获取一行
        String line = value.toString();
        
        if(name.startsWith("order")){//订单表

            String[] fields = line.split("t");
            //封装key和value
            v.setId(fields[0]);
            v.setPid(fields[1]);
            v.setAmount(Integer.parseInt(fields[2]));
            v.setPname("");
            v.setFlag("order");
            k.set(fields[1]);
        }else {//产品表

            String[] fields = line.split("t");
            //封装key和value
            v.setId("");
            v.setPid(fields[0]);
            v.setAmount(0);
            v.setPname(fields[1]);
            v.setFlag("pd");
            k.set(fields[0]);
        }

        //写出
        context.write(k, v);
    }
}

TableReducer.java

public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {

        //存储所有订单集合
        ArrayList<TableBean> orderBeans = new ArrayList<>();
        //存储产品信息
        TableBean pdBean = new TableBean();

        for (TableBean tableBean : values) {

            if("order".equals(tableBean.getFlag())){//订单表

                TableBean tmpBean = new TableBean();
                try {
                    BeanUtils.copyProperties(tmpBean, tableBean);
                    orderBeans.add(tmpBean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }else {
                try {
                    BeanUtils.copyProperties(pdBean, tableBean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }
        for (TableBean tableBean : orderBeans) {
            tableBean.setPname(pdBean.getPname());
            context.write(tableBean, NullWritable.get());
        }
    }
}

TableDriver.java

public class TableDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        args = new String[]{"f:/input/inputtable","f:/output1"};

        // 1 获取配置信息，或者job对象实例
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        // 2 指定本程序的jar包所在的本地路径
        job.setJarByClass(TableDriver.class);

        // 3 指定本业务job要使用的Mapper/Reducer业务类
        job.setMapperClass(TableMapper.class);
        job.setReducerClass(TableReducer.class);

        // 4 指定Mapper输出数据的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(TableBean.class);

        // 5 指定最终输出的数据的kv类型
        job.setOutputKeyClass(TableBean.class);
        job.setOutputValueClass(NullWritable.class);

        // 6 指定job的输入原始文件所在目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);

    }
}

*（2）Map Join（cache）

1）使用场景

适用于一张表十分小，一张表很大的场景

2）优点

在Map端缓存多张表，提前处理业务逻辑，这样增加Map端业务，减少Reduce端的数据压力，尽可能的减少数据倾斜

3）具体办法：采用DistributedCache

在Mapper的setup阶段，将文件读取到缓存集合中。

在驱动函数中加载缓存。

// 缓存普通文件到Task运行节点。
job.addCacheFile(newURI("file://e:/cache/pd.txt"));

4）案例

DistributedCacheDriver.java

public class DistributedCacheDriver {

    public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {

        args = new String[]{"f:/input/inputtable2", "f:/output1"};

        // 1 获取job信息
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        // 2 设置加载jar包路径
        job.setJarByClass(DistributedCacheDriver.class);

        // 3 关联map
        job.setMapperClass(DistributedCacheMapper.class);

        // 4 设置最终输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        // 5 设置输入输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 6 加载缓存数据
        job.addCacheFile(new URI("file:///f:/input/inputcache/pd.txt"));

        // 7 Map端Join的逻辑不需要Reduce阶段，设置reduceTask数量为0
        job.setNumReduceTasks(0);

        // 8 提交
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);

    }
}

DistributedCacheMapper.java

public class DistributedCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    HashMap<String, String> pdMap = new HashMap<>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //缓存小表
        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath().toString();

        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path), "UTF-8"));

        String line;
        while (StringUtils.isNotEmpty(line = reader.readLine())){
//            pid	pname
//            01	小米

            //1.切割
            String[] fileds = line.split("t");
            pdMap.put(fileds[0],fileds[1]);
        }
        //关闭资源
        IOUtils.closeStream(reader);
    }

    Text k = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//        id	pid	amount
//        1001	01	1

//        pid	pname
//        01	小米
        //1.获取一行
        String line = value.toString();

        //2.切割
        String[] fileds = line.split("t");

        //3.获取pid
        String pid = fileds[1];

        //4.取出pname
        String pname = pdMap.get(pid);

        //5.拼接
        line = fileds[0] +"t"+ pname +"t"+ fileds[2];

        k.set(line);
        //6.写出
        context.write(k,NullWritable.get());
    }
}

8.计数器应用

（1）Hadoop为每个作业维护若干内置计数器，以描述多项指标

（2）计数器API

1）采用枚举的方式

enum MyCounter{MALFORORMED,NORMAL};
//对枚举定义的自定义计数器加1
context.getCounter(MyCounter.MALFORORMED).increment(1);

2）采用计数器组、计数器名称的方式统计

//组名和计数器名称随便起，但最好有意义
context.getCounter("counterGroup","counter").increment(1);

3）计数结果可在程序运行后的控制台查看

9.数据清洗（ETL）

（1）概念

对脏数据的过滤和清理，处理掉不符合用要求的数据（各种Null、各种空数据、404信息）
清理过程一般只需要Mappe程序，不需要Reduce程序

*（2）案例（log）

LogMapper.java

public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //1.获取一行
        String line = value.toString();

        //2.解析数据
        boolean result = parseLog(line, context);
        if(!result){
            return;
        }

        //3.解析通过，写出去
        context.write(value, NullWritable.get());
    }

    private boolean parseLog(String line, Context context) {

        String[] fields = line.split(" ");

        if (fields.length > 11){

            context.getCounter("map", "true").increment(1);
            return true;
        }else {

            context.getCounter("map", "false").increment(1);
            return false;
        }
    }
}

LogDriver.java

public class LogDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        args = new String[] { "f:/input/inputlog", "f:/output1" };

        // 1 获取job信息
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 2 加载jar包
        job.setJarByClass(LogDriver.class);

        // 3 关联map
        job.setMapperClass(LogMapper.class);

        // 4 设置最终输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        // 设置reducetask个数为0
        job.setNumReduceTasks(0);

        // 5 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 6 提交
        job.waitForCompletion(true);

    }
}

四、Hadoop数据压缩（重）

1.概述

（1）作用

压缩技术能够有效减少底层存储系统（HDFS）读写字节数
在数据规模很大和工作负载密集的情况下，使用数据压缩非常重要
数据压缩对于节省资源、最小化磁盘I/O和网络传输非常有帮助
可以在任意MapReduce阶段启用压缩

（2）策略

通过对Mapper、Reducer运行过程的数据进行压缩，可以减少磁盘IO，提高MR程序运行速度
注意：采用压缩技术减少了磁盘IO，但同时增加了CPU运算负担

（3）原则

运算密集型的job，少用压缩
IO密集型的job，多用压缩

2.MR支持的压缩编码

（1）压缩方式

压缩格式	hadoop自带？	算法	文件扩展名	是否可切分	换成压缩格式后，原来的程序是否需要修改
DEFLATE	是，直接使用	DEFLATE	.deflate	否	和文本处理一样，不需要修改
Gzip	是，直接使用	DEFLATE	.gz	否	和文本处理一样，不需要修改
bzip2	是，直接使用	bzip2	.bz2	是	和文本处理一样，不需要修改
LZO	否，需要安装	LZO	.lzo	是	需要建索引，还需要指定输入格式
Snappy	否，需要安装	Snappy	.snappy	否	和文本处理一样，不需要修改

（2）Hadoop的编码/解码器

压缩格式	对应的编码/解码器
DEFLATE	org.apache.hadoop.io.compress.DefaultCodec
gzip	org.apache.hadoop.io.compress.GzipCodec
bzip2	org.apache.hadoop.io.compress.BZip2Codec
LZO	com.hadoop.compression.lzo.LzopCodec
Snappy	org.apache.hadoop.io.compress.SnappyCodec

（3）压缩性能的比较

压缩算法	原始文件大小	压缩文件大小	压缩速度	解压速度
gzip	8.3GB	1.8GB	17.5MB/s	58MB/s
bzip2	8.3GB	1.1GB	2.4MB/s	9.5MB/s
LZO	8.3GB	2.9GB	49.3MB/s	74.6MB/s

3.压缩方式选择

（1）Gzip压缩

优点：

压缩率比较高，压缩/解压速度比较快；
Hadoop本身支持，在应用中处理Gzip格式的文件就和直接处理文本一样；
大部分Linux系统都自带Gzip命令

缺点：

不支持Split（切分）

应用场景：

当每个文件压缩之后在130M以内的（1个块大小内）

（2）Bzip2压缩

优点：

支持Split（切分）
很高的压缩率，高于Gzip
Hadoop本身自带

缺点：

压缩/解压缩速度慢

应用场景：

对速度要求不高，但需要较高的压缩率
输出之后的数据比较大，处理之后的数据需要压缩存档减少磁盘空间，并且以后数据用的少
单个很大的文本文件想压缩减少存储空间，同时又需要支持Split，而且兼容之前的应用程序

*（3）Lzo压缩

优点：

压缩/解压缩速度比较快，合理的压缩率
支持Split，是Hadoop中流行的压缩格式
可以在Linux系统下安装lzop命令，使用方便

缺点：

压缩率比Gzip要低一些
Hadoop本身不支持，需要安装
在应用中对Lzo格式的文件需要做一些特殊处理（为了支持Split需要建索引，还需要指定InputFormat为Lzo格式）

应用场景：

一个很大的文本文件，压缩之后还要大于200M以上的可以考虑，单个文件越大，Lzo优点越明显

*（4）Snappy压缩

优点：

高速压缩速度和合理的压缩率

缺点：

不支持Split
压缩率比Gzip要低
Hadoop本身不支持，需要安装

应用场景：

当MapReduce作业的Map输出数据比较大时，作为Map到Reduce的中间数据的压缩格式
作为MapReduce作业的输出和另一个MapReduce作业的输入

4.压缩位置选择

在这里插入图片描述

5.压缩参数配置

参数	默认值	阶段	建议
io.compression.codecs （在core-site.xml中配置）	org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec	输入压缩	Hadoop使用文件扩展名判断是否支持某种编解码器
mapreduce.map.output.compress（在mapred-site.xml中配置）	false	mapper输出	这个参数设为true启用压缩
mapreduce.map.output.compress.codec（在mapred-site.xml中配置）	org.apache.hadoop.io.compress.DefaultCodec	mapper输出	企业多使用LZO或Snappy编解码器在此阶段压缩数据
mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置）	false	reducer输出	这个参数设为true启用压缩
mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置）	org.apache.hadoop.io.compress. DefaultCodec	reducer输出	使用标准工具或者编解码器，如gzip和bzip2
mapreduce.output.fileoutputformat.compress.type（在mapred-site.xml中配置）	RECORD	reducer输出	SequenceFile输出使用的压缩类型：NONE和BLOCK

6.压缩实操案例

（1）数据流的压缩和解压缩

压缩：

使用createOutputStream(OutputStreamout)方法创建一个CompressionOutputStream

public class TestCompress {
    public static void main(String[] args) throws Exception{
		//压缩(目标文件,压缩格式)
//        compress("f:/hello.txt","org.apache.hadoop.io.compress.BZip2Codec");
//        compress("f:/hello.txt","org.apache.hadoop.io.compress.GzipCodec");
        compress("f:/hello.txt","org.apache.hadoop.io.compress.DefaultCodec");
    } 
    
    private static void compress(String fileName, String method) throws IOException, ClassNotFoundException {

        //获取输入流
        FileInputStream fis = new FileInputStream(new File(fileName));

        Class classCodec = Class.forName(method);
        CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(classCodec, new Configuration());

        //获取输出流
        FileOutputStream fos = new FileOutputStream(new File(fileName + codec.getDefaultExtension()));
        CompressionOutputStream cos = codec.createOutputStream(fos);//获取有压缩功能的输出流

        //流的对拷
        IOUtils.copyBytes(fis, cos, 1024*1024*5, false);

        //关闭资源
        IOUtils.closeStream(cos);
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
    }
}

解压缩：

使用createInputStream(InputStreamin)方法创建一个CompressionInputStream

public class TestCompress {
    public static void main(String[] args) throws Exception{
        //解压(目标文件)
        decompress("f:/hello.txt.bz2");
    }
    
     private static void decompress(String fileName) throws IOException {
        //1.压缩方式检查
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
        CompressionCodec codec = factory.getCodec(new Path(fileName));
        if(codec == null){
            System.out.println("无法处理这种压缩文件");
            return;
        }

        //获取输入流
        FileInputStream fis = new FileInputStream(new File(fileName));
        CompressionInputStream cis = codec.createInputStream(fis);//对输出流进行包装

        //获取输出流
        FileOutputStream fos = new FileOutputStream(new File(fileName + ".decode"));

        //流的对拷
        IOUtils.copyBytes(cis, fos, 1024*1024*5, false);

        //关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(cis);
        IOUtils.closeStream(fis);

    }
}

（2）Map输出端采用压缩

在Driver驱动类中加上两行代码，Mapper和Reducer不变，输出结果格式不影响

public class WordcountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        args = new String[]{"f:/input/inputwc","f:/output1"};
        //1.获取Job对象
        Configuration conf=new Configuration();
        Job job = Job.getInstance(conf);

        // 开启map端输出压缩
        conf.setBoolean("mapreduce.map.output.compress", true);
        // 设置map端输出压缩方式
        conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

        //2.存储jar存储位置
        job.setJarByClass(WordcountDriver.class);

        //3.关联Map和Reduce类
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);

        //4.设置Mapper阶段输出数据的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5.设置最终数据输出key和value类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //6.设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //7.提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

（3）Reduce输出端采用压缩

最终输出文件采用压缩方式

public class WordcountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        args = new String[]{"f:/input/inputwc","f:/output1"};
        //1.获取Job对象
        Configuration conf=new Configuration();
        Job job = Job.getInstance(conf);

        // 开启map端输出压缩
        conf.setBoolean("mapreduce.map.output.compress", true);
        // 设置map端输出压缩方式
        conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

        //2.存储jar存储位置
        job.setJarByClass(WordcountDriver.class);

        //3.关联Map和Reduce类
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);

        //4.设置Mapper阶段输出数据的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5.设置最终数据输出key和value类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置reduce端输出压缩开启
        FileOutputFormat.setCompressOutput(job, true);
        // 设置压缩的方式
//        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
//        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
        FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);
        
        //6.设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //7.提交job
        boolean result = job.waitForCompletion(true);
        System

五、Yarn资源调度器（面试重）

1.Yarn基本架构

YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成。

ResourceManage：整个集群资源调度的老大
NodeManager：单个几点资源调度的老大
ApplicationMaster：单个Job资源调度的老大
Container：资源的抽象，虚拟化CPU、内存等

2.Yarn的工作机制

在这里插入图片描述

3.作业提交全过程

（1）作业提交过程之YARN，过程如Yarn的工作机制流程

（2）作业提交过程之MapReduce
在这里插入图片描述

4.资源调度器

Hadoop作业调度器主要有三种：FIFO（队列）、Capacity Scheduler（容量）和Fair Scheduler（公平）

具体见yarn-default.xml

（1）队列调度器（FIFO）

先进先出，按照到达时间排序，先到先服务
只有一个队列

（2）容量调度器（Capacity Scheduler）

支持多个队列，每个队列可配置一定的资源量，每个队列采用先进先出策略
为了防止同一个用户的作业独占队列中的资源，该调度器会对同一用户提交的作业所占资源量进行限定
首先，计算每个队列中正在运行的任务数与其应该分配的计算资源比值，选择一个比值比较小的队列–最闲的
其次，按照作业优先级和提交时间顺序，同时考虑用户资源量限制和内存限制对队列内任务排序
多个队列同时按照任务的先后顺序依次执行，这几个队列并行运行

（3）公平调度器（Fair Scheduler）

支持多队列多用户，每个队列中的资源量可配置，同一队列中的作业公平共享队列中所有资源

每个队列中的job按照优先级分配资源，优先级越高的分配资源越多，但是每个job都会分配到

缺额：每个job理想获得的计算资源与实际获得的计算资源存在的差距

在同一个队列中，job的资源缺额越大，越优先获得资源优先执行

多个队列同时运行；同一个队列，多个job也同时运行，并发量高

5.任务的推测执行

（1）作业完成时间取决于最慢的任务完成时间

（2）推测执行机制

发现拖后腿的任务，比如某个任务运行速度远慢于任务平均速度。为拖后腿任务启动一个备份任务，同时运行。谁先运行完，则采用谁的结果。

（3）执行推测任务的前提条件

每个Task只能有一个备份任务
当前Job已完成的Task必须不小于0.05（5%）
推测执行参数设置。mapred-site.xml文件中默认是打开的

（4）不能启用推测执行机制情况

任务间存在严重的负载倾斜；
特殊任务，比如任务向数据库中写数据。

（5）算法原理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wqm6V53j-1604044596012)(images任务推测执行算法原理.png)]

六、Hadoop企业优化（重重）

1.MapReduce跑的慢的原因

（1）计算机性能

CPU、内存、磁盘健康、网络

（2）I/O操作优化

数据倾斜
MapTask和ReduceTask数设置不合理
Map运行时间太长，导致Reduce等候太久
小文件过多
大量的不可分块的超大文件
Spill（溢写）次数过多
Merge（归并排序）次数过多

2.MapReduce优化方法

主要从六个方面考虑：数据输入、Map阶段、Reduce阶段、IO传输、数据倾斜问题和常用的调优参数

（1）数据输入

合并小文件：在执行MR任务前将小文件进行合并
采用CombineTextInputFormat来作为输入

（2）Map阶段

减少溢写次数：通过调整环形缓冲区大小（io.sort.mb）及触发溢写比例80%（sort.spill.percent）参数
减少合并的次数：调整io.sort.factor,增大Merge的文件数目，减少Merge的次数
在Map之后，不影响业务的前提下，先进行Combine处理，减少I/O

（3）Reduce阶段

合理设置Map和Reduce数：两个都不太少，也不能太多
设置Map、Reduce共存：调整slowstart.completedmaps参数，使map运行到一定程度后，Reduce也开始运行，减少Reduce的等待时间
规避使用Reduce：Reduce在用于连接数据集的时候将会产生大量网络消耗
合理设置Reduce端的Buffer：使得buffer中一部分数据可以直接输送到Reduce端
- mapred.job.reduce.input.buffer.percen默认值为0.0，当>0时，会保留指定比例的内存读Buffer中的数据直接拿给Reduce。这样就很占内存，合理调整

（4）I/O传输

采用数据压缩方式，减少网络IO的时间。可以安装Snappy和LZO压缩编码器
使用SequenceFile二进制文件

（5）数据倾斜问题

1）数据倾斜现象

数据频率倾斜：某一个区域的数据量要远远大于其他区域

数据大小倾斜：部分记录的大小远远大于平均值

2）减少数据倾斜的方法

抽样和范围分区
自定义分区
使用Conbine可以大量的减少数据倾斜
采用Map Join，尽量避免Reduce Join

*（6）常用的调优参数

1）资源相关参数

MR应用程序中配置生效（mapred-default.xml）

配置参数	参数说明
mapreduce.map.memory.mb	一个MapTask可使用的资源上限（单位:MB），默认为1024。如果MapTask实际使用的资源量超过该值，则会被强制杀死。
mapreduce.reduce.memory.mb	一个ReduceTask可使用的资源上限（单位:MB），默认为1024。如果ReduceTask实际使用的资源量超过该值，则会被强制杀死。
mapreduce.map.cpu.vcores	每个MapTask可使用的最多cpu core数目，默认值: 1
mapreduce.reduce.cpu.vcores	每个ReduceTask可使用的最多cpu core数目，默认值: 1
mapreduce.reduce.shuffle.parallelcopies	每个Reduce去Map中取数据的并行数。默认值是5
mapreduce.reduce.shuffle.merge.percent	Buffer中的数据达到多少比例开始写入磁盘。默认值0.66
mapreduce.reduce.shuffle.input.buffer.percent	Buffer大小占Reduce可用内存的比例。默认值0.7
mapreduce.reduce.input.buffer.percent	指定多少比例的内存用来存放Buffer中的数据，默认值是0.0

YARN启动之前就配置在服务器的配置文件中才能生效（yarn-default.xml）

配置参数	参数说明
yarn.scheduler.minimum-allocation-mb	给应用程序Container分配的最小内存，默认值：1024
yarn.scheduler.maximum-allocation-mb	给应用程序Container分配的最大内存，默认值：8192
yarn.scheduler.minimum-allocation-vcores	每个Container申请的最小CPU核数，默认值：1
yarn.scheduler.maximum-allocation-vcores	每个Container申请的最大CPU核数，默认值：32
yarn.nodemanager.resource.memory-mb	给Containers分配的最大物理内存，默认值：8192

Shuffle性能优化的关键参数，应在YARN启动之前就配置好（mapred-default.xml）

配置参数	参数说明
mapreduce.task.io.sort.mb	Shuffle的环形缓冲区大小，默认100m
mapreduce.map.sort.spill.percent	环形缓冲区溢出的阈值，默认80%

2）容错相关参数(MapReduce性能优化)

配置参数	参数说明
mapreduce.map.maxattempts	每个Map Task最大重试次数，一旦重试参数超过该值，则认为Map Task运行失败，默认值：4。
mapreduce.reduce.maxattempts	每个Reduce Task最大重试次数，一旦重试参数超过该值，则认为Map Task运行失败，默认值：4。
mapreduce.task.timeout	Task超时时间，经常需要设置的一个参数，该参数表达的意思为：如果一个Task在一定时间内没有任何进入，即不会读取新的数据，也没有输出数据，则认为该Task处于Block状态，可能是卡住了，也许永远会卡住，为了防止因为用户程序永远Block住不退出，则强制设置了一个该超时时间（单位毫秒），默认是600000。如果你的程序对每条输入数据的处理时间过长（比如会访问数据库，通过网络拉取数据等），建议将该参数调大，该参数过小常出现的错误提示是“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.”。

3.HDFS小文件优化方法

（1）HDFS小文件弊端

一方面会大量占用NameNode的内存空间，另一方面就是索引文件过大使得索引速度变慢

（2）HDFS小文件解决方案

在数据采集的时候，就将小文件或小批数据合成大文件再上传HDFS。
在业务处理之前，在HDFS上使用MapReduce程序对小文件进行合并。
在MapReduce处理时，可采用CombineTextInputFormat提高效率。

实现方法：

Hadoop Archive：将多个文件打包成一个HAR文件，对外是一个整体，对内是一个一个小文件
Sequence File：由一系列的二进制key/value组成，可以为文件名，value为文件内容
CombineFileInputFormat：将多个文件合并成一个单独的Split，会考虑数据的存储位置
开启JVM重用：开启后会减少45%运行时间
- 原理：一个Map运行在一个JVM上，开启重用的话，该Map在JVM上运行完毕后，JVM继续运行其它Map
- 具体设置：mapreduce.job.jvm.numtasks值在10-20之间

七、MapReduce扩展案例（掌握）

1.倒排索引案例（多job串联）

第一次MR

OneIndexMapper.java

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

    String name;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取文件名称
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        name= inputSplit.getPath().getName();

    }

    Text k = new Text();
    IntWritable v = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //atguigu pingping
        
        //1.获取一行
        String line = value.toString();

        //2.切割
        String[] fields = line.split(" ");

        //3.写出
        for (String word : fields) {
            k.set(word+"--"+name);
            context.write(k, v);
        }
    }
}

OneIndexReducer.java

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

    IntWritable v=new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        int sum = 0;
        //1.累加求和
        for (IntWritable value : values) {
            sum += value.get();
        }

        v.set(sum);
        //2.写出
        context.write(key, v);
    }
}

OneIndexDriver.java这里省略

第二次MR

TwoIndexMapper.java

public class TwoIndexMapper extends Mapper<LongWritable, Text,Text,Text> {

    Text k = new Text();
    Text v = new Text();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //atguigu--a.txt	3
        //1.读取一行
        String line = value.toString();

        //2.切割
        String[] fields = line.split("--");

        //3.封装
        k.set(fields[0]);
        v.set(fields[1]);

        //4.写出
        context.write(k, v);
    }
}

TwoIndexReducer.java

public class TwoIndexReducer extends Reducer<Text,Text,Text,Text> {

    Text v = new Text();
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        // atguigu a.txt 3
        // atguigu b.txt 2
        // atguigu c.txt 2

        // atguigu c.txt-->2 b.txt-->2 a.txt-->3

        StringBuffer sb = new StringBuffer();
        //1.拼接
        for (Text value : values) {
            sb.append(value.toString().replace("t", "-->")+"t");
        }

        v.set(sb.toString());

        //2.写出
        context.write(key, v);
    }
}

TwoIndexDriver.java此处省略

*2.TopN案例（topN）

输入数据文件为第二章第二节输出数据文件

FlowBean.java

public class FlowBean implements WritableComparable<FlowBean> {

    private long upFlow;
    private long downFlow;
    private long sumFlow;

    public FlowBean() {
    }

    public FlowBean(long upFlow, long downFlow, long sumFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = sumFlow;
    }

    @Override
    public int compareTo(FlowBean bean) {

        int result;

        if (this.sumFlow > bean.getSumFlow()) {
            //降序为-1
            result = -1;
        }else if (this.sumFlow < bean.getSumFlow()) {
            result = 1;
        }else {
            result = 0;
        }

        return result;

    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(upFlow);
        dataOutput.writeLong(downFlow);
        dataOutput.writeLong(sumFlow);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        upFlow=dataInput.readLong();
        downFlow=dataInput.readLong();
        sumFlow=dataInput.readLong();
    }
	
	//此处省略get和set方法

    @Override
    public String toString() {
        return upFlow + "t" + downFlow + "t" + sumFlow ;
    }

    public void set(long downFlow2, long upFlow2) {
        downFlow = downFlow2;
        upFlow = upFlow2;
        sumFlow = downFlow2 + upFlow2;
    }

}

TopNMapper.java

public class TopNMapper extends Mapper<LongWritable, Text, FlowBean, Text> {

    // 定义一个TreeMap作为存储数据的容器（天然按key排序）
    private TreeMap<FlowBean,Text> flowMap = new TreeMap<FlowBean,Text>();

    private FlowBean kBean;

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        kBean = new FlowBean();
        Text v = new Text();
      
     	//13470253144	180	180	360
        //1.获取一行
        String line = value.toString();

        //2.切割
        String[] fields = line.split("t");

        //3.封装数据
        String phoneNum = fields[0];
        long upFlow = Long.parseLong(fields[1]);
        long downFlow = Long.parseLong(fields[2]);
        long sumFlow = Long.parseLong(fields[3]);

        kBean.setUpFlow(upFlow);
        kBean.setDownFlow(downFlow);
        kBean.setSumFlow(sumFlow);

        v.set(phoneNum);
        //4.向TreeMap中添加数据
        flowMap.put(kBean, v);

        //5.限制TreeMap的数据量，超过10条就删除掉流量最小的一条数据
        if(flowMap.size() > 10){
            //删除第一条，取最后十条
//            flowMap.remove(flowMap.firstKey());
            //删除最后一条，取前十条
            flowMap.remove(flowMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {

        // 6.遍历treeMap集合，输出数据
        Iterator<FlowBean> bean = flowMap.keySet().iterator();

        while (bean.hasNext()){
            FlowBean k = bean.next();
            context.write(k, flowMap.get(k));
        }
    }
}

TopNReducer.java

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

    @Override
    protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        for(Text k : values){
            context.write(k, key);
        }
    }
}