概述
一、Hadoop
第1章 HDFS概述
1.1 HDFS产出背景及定义
HDFS（Hadoop Distributed File System），分布式文件系统。通过目录树来定位存储文件，分布式集群系统。
应用场景：适合一次写入，多次读出，不支持修改文件。
1.2 HDFS优缺点
1.3 HDFS组织架构
1）NameNode（nn）–Master：
（1）管理HDFS的名称空间；
（2）配置副本策略；
（3）管理数据块映射信息；
（4）处理客户单读写请求。
2）DataNode（dn）–Salve：
（1）存储实际的数据块；
（2）执行数据块的读/写操作；
3）Client --客户端：
（1）将文件切分为多个Block；
（2）与NameNode、DataNode交互；
（3）可以通过一些命令管理和访问HDFS
4）Secondary NameNode：
（1）辅助NameNode
（2）辅助恢复NameNode
1.4 HDFS文件块大小
思考：为什么HDFS块大小不能太大也不能太小？
（1）块设置太小会增加寻址时间；
（2）块如果设置太大，从磁盘传输数据的时间会明显大于定位这个块开始位置所需的时间，处理块数据速度慢
HDFS块的大小设置主要取决于磁盘传输速度。
第2章 HDFS的Shell操作
2.1 基本语法
（1）bin/ hdfs dfs
（2）bin/ hadoop fs
2.2 命令大全
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] [-d] [-t <thread count>] <localsrc> ... <dst>]
[-copyToLocal [-f] [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
[-count [-q] [-h] [-v] [-t [<storage type>]] [-u] [-x] [-e] <path> ...]
[-cp [-f] [-p | -p[topax]] [-d] <src> ... <dst>]
[-createSnapshot <snapshotDir> [<snapshotName>]]
[-deleteSnapshot <snapshotDir> <snapshotName>]
[-df [-h] [<path> ...]]
[-du [-s] [-h] [-v] [-x] <path> ...]
[-expunge]
[-find <path> ... <expression> ...]
[-get [-f] [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
[-getfacl [-R] <path>]
[-getfattr [-R] {-n name | -d} [-e en] <path>]
[-getmerge [-nl] [-skip-empty-file] <src> <localdst>]
[-head <file>]
[-help [cmd ...]]
[-ls [-C] [-d] [-h] [-q] [-R] [-t] [-S] [-r] [-u] [-e] [<path> ...]]
[-mkdir [-p] <path> ...]
[-moveFromLocal <localsrc> ... <dst>]
[-moveToLocal <src> <localdst>]
[-mv <src> ... <dst>]
[-put [-f] [-p] [-l] [-d] <localsrc> ... <dst>]
[-renameSnapshot <snapshotDir> <oldName> <newName>]
[-rm [-f] [-r|-R] [-skipTrash] [-safely] <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] [-s <sleep interval>] <file>]
[-test -[defsz] <path>]
[-text [-ignoreCrc] <src> ...]
[-touch [-a] [-m] [-t TIMESTAMP ] [-c] <path> ...]
[-touchz <path> ...]
[-truncate [-w] <length> <path> ...]
[-usage [cmd ...]]
2.3 常用命令 --help查询
2.3.1 上传
1）-moveFromLocal：从本地剪切粘贴到HDFS
[atguigu@hadoop102 hadoop-3.1.3]$ touch kongming.txt
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -moveFromLocal ./kongming.txt /sanguo/shuguo
2）-copyFromLocal：从本地文件系统中拷贝文件到HDFS路径去
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -copyFromLocal README.txt /
3）-appendToFile：追加一个文件到已经存在的文件末尾
[atguigu@hadoop102 hadoop-3.1.3]$ touch liubei.txt
[atguigu@hadoop102 hadoop-3.1.3]$ vi liubei.txt
输入
san gu mao lu
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -appendToFile liubei.txt /sanguo/shuguo/kongming.txt
4）-put：等同于copyFromLocal
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -put ./zaiyiqi.txt /user/atguigu/test/
2.3.2 下载
1）-copyToLocal：从HDFS拷贝到本地
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -copyToLocal /sanguo/shuguo/kongming.txt ./
2）-get：等同于copyToLocal，就是从HDFS下载文件到本地
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -get /sanguo/shuguo/kongming.txt ./
3）-getmerge：合并下载多个文件，比如HDFS的目录 /user/atguigu/test下有多个文件:log.1, log.2,log.3,…
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -getmerge /user/atguigu/test/* ./zaiyiqi.txt
2.3.3 HDFS直接操作
1）-ls: 显示目录信息
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -ls /
2）-mkdir：在HDFS上创建目录
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -mkdir -p /sanguo/shuguo
3）-cat：显示文件内容
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -cat /sanguo/shuguo/kongming.txt
4）-chgrp 、-chmod、-chown：Linux文件系统中的用法一样，修改文件所属权限
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -chmod 666 /sanguo/shuguo/kongming.txt
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -chown atguigu:atguigu /sanguo/shuguo/kongming.txt
5）-cp ：从HDFS的一个路径拷贝到HDFS的另一个路径
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -cp /sanguo/shuguo/kongming.txt /zhuge.txt
6）-mv：在HDFS目录中移动文件
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -mv /zhuge.txt /sanguo/shuguo/
7）-tail：显示一个文件的末尾
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -tail /sanguo/shuguo/kongming.txt
8）-rm：删除文件或文件夹
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -rm /user/atguigu/test/jinlian2.txt
9）-rmdir：删除空目录
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -mkdir /test
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -rmdir /test
10）-du统计文件夹的大小信息
-s 汇总所有文件大小
-h 单位换算，以最合适的单位进行显示
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -du -s -h /user/atguigu/test
2.7 K /user/atguigu/test
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -du -h /user/atguigu/test
1.3 K /user/atguigu/test/README.txt
15 /user/atguigu/test/jinlian.txt
1.4 K /user/atguigu/test/zaiyiqi.txt
11）-setrep：设置HDFS中文件的副本数量
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -setrep 10 /sanguo/shuguo/kongming.txt
第3章 HDFS客户端操作
3.1 客户端环境准备
1）找到相关依赖并拷贝
2）配置环境变量
3）创建一个Maven工程HdfsClientDemo，并导入相应的依赖坐标+日志添加
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
</dependency>
</dependencies>
在项目的src/main/resources目录下，新建一个文件，命名为“log4j2.xml”，在文件中填入
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="error" strict="true" name="XMLConfig">
<Appenders>
<!-- 类型名为Console，名称为必须属性 -->
<Appender type="Console" name="STDOUT">
<!-- 布局为PatternLayout的方式，
输出样式为[INFO] [2018-01-22 17:34:01][org.test.Console]I'm here -->
<Layout type="PatternLayout"
pattern="[%p] [%d{yyyy-MM-dd HH:mm:ss}][%c{10}]%m%n" />
</Appender>
</Appenders>
<Loggers>
<!-- 可加性为false -->
<Logger name="test" level="info" additivity="false">
<AppenderRef ref="STDOUT" />
</Logger>
<!-- root loggerConfig设置 -->
<Root level="info">
<AppenderRef ref="STDOUT" />
</Root>
</Loggers>
</Configuration>
4）创建包/类
public class HdfsClient{
@Test
public void testMkdirs() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
// 配置在集群上运行
// configuration.set("fs.defaultFS", "hdfs://hadoop102:8020");
// FileSystem fs = FileSystem.get(configuration);
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "atguigu");
// 2 创建目录
fs.mkdirs(new Path("/1108/daxian/banzhang"));
// 3 关闭资源
fs.close();
}
}
5）执行程序
3.2 HDFS API操作
创建Junit下After和Before测试
FileSystem fs = null;
@Before
public void before() throws URISyntaxException, IOException, InterruptedException {
Configuration conf = new Configuration();
//创建文件系统的对象
/**
* get(final URI uri, final Configuration conf,
* final String user)
* uri : HDFS的地址
* conf : 配置文件的对象
* user : 操作HDFS的用户
*/
fs = FileSystem.get(
new URI("hdfs://hadoop102:9820"), conf, "atguigu"
);
}
@After
public void after() throws IOException {
if (fs != null) {
fs.close();
}
}
3.2.1HDFS文件上传
@Test
public void test3() throws IOException {
/*
copyFromLocalFile(boolean delSrc, boolean overwrite,
Path src, Path dst)
delSrc : 是否删除本地文件（剪切 or 复制） true-剪切 false-复制
overwrite : 是否覆盖HDFS中相同的文件。
true : 如果文件名相同则进行覆盖
false : 如果文件名相同则抛异常
src : 本地路径
dst : HDFS路径
*/
fs.copyFromLocalFile(true,false,
new Path("D:\aa.txt"),new Path("/"));
}
3.2.2 HDFS文件下载
@Test
public void test4() throws IOException {
/*
copyToLocalFile(boolean delSrc, Path src, Path dst,
boolean useRawLocalFileSystem)
delSrc : HDFS上的文件是否删除
src : HDFS的路径
dst : 本地路径
useRawLocalFileSystem ： 是否使用useRawLocalFileSystem系统
（是否获取校验和文件）
true : 不会创建crc文件
false : 会创建crc文件
*/
fs.copyToLocalFile(false,new Path("/aa.txt"),
new Path("D://"),true);
}
3.2.3 删除
@Test
public void test5() throws IOException {
/*
boolean delete(Path f, boolean recursive)
f : HDFS的路径
recursive : 如果f是一个文件那么true和false都可以
如果f是一个目录，那么必须是true,如果是false就抛异常
*/
fs.delete(new Path("/demo"),true);
}
3.2.4 改名和移动
@Test
public void test6() throws IOException {
/*
rename(Path src, Path dst)
src : 源文件路径
dst ：目标文件路径
*/
//fs.rename(new Path("/aaa.txt"),new Path("/ccc.txt"));
fs.rename(new Path("/ccc.txt"),
new Path("/input/ccc.txt"));
}
3.2.5 查看文件详情
@Test
public void test7() throws IOException {
/*
listFiles(
final Path f, final boolean recursive)
f ： 路径
recursive : 是否对子目录进行递归
*/
RemoteIterator<LocatedFileStatus> fi
= fs.listFiles(new Path("/"), true);
while(fi.hasNext()){
System.out.println("================================");
//获取一个文件
LocatedFileStatus file = fi.next();
//输出文件信息
System.out.println(file.getPath().getName());//输出文件名
System.out.println(file.getReplication());//输出副本数
//获取块信息
BlockLocation[] blockLocations = file.getBlockLocations();
//输出块信息
System.out.println(Arrays.toString(blockLocations));
}
}
3.2.6 判断文件和目录
@Test
public void test8() throws IOException {
FileStatus[] files = fs.listStatus(new Path("/"));
for (FileStatus file : files) {
if (file.isDirectory()){
System.out.println(file.getPath().getName() + "是一个目录");
}else if(file.isFile()){
System.out.println(file.getPath().getName() + "是一个文件");
}
}
}
3.2.7 通过流向HDFS上传、下载文件
@Test
public void test9() throws Exception {
//上传
//创建输入流读取本地文件的内容
FileInputStream fis = new FileInputStream(new Path("D:\aa.txt"));
//创建输出流将文件写到HDFS上
FSDataOutputStream fos
= fs.create(new Path("/aa.txt"));
//文件对拷
IOUtils.copyBytes(fis,fos,2000);
//关闭流
IOUtils.closeStream(fis);
IOUtils.closeStream(fos);
}
//下载
@Test
public void test10() throws Exception {
FSDataInputStream fis = fs.open(new Path("/aa.txt"));
FileOutputStream fos = new FileOutputStream(new Path("D://aa.txt"));
//最后一个参数 ：true就会关流 false不会关流
IOUtils.copyBytes(fis,fos,2000,true);
}
第4章 HDFS的数据流
4.1 HDFS写数据流程
4.1.1 剖析文件写入
（1）客户端通过Distributed FileSystem模块向NameNode请求上传文件，NameNode检查目标文件是否已存在，父目录是否存在。
（2）NameNode返回是否可以上传。
（3）客户端请求第一个 Block上传到哪几个DataNode服务器上。
（4）NameNode返回3个DataNode节点，分别为dn1、dn2、dn3。
（5）客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成。
（6）dn1、dn2、dn3逐级应答客户端。
（7）客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答。[packet为64kb里面包含n个（truck 512b+检验值4b）]
（8）当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block的服务器。（重复执行3-7步）。
4.1.2 节点距离计算
4.1.3 机架感知
1）官方说明
http://hadoop.apache.org/docs/r3.1.3/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html#Data_Replication
For the common case, when the replication factor is three, HDFS’s placement policy is to put one replica on the local machine if the writer is on a datanode, otherwise on a random datanode, another replica on a node in a different (remote) rack, and the last on a different node in the same remote rack.
2）副本节点选择
4.2 HDFS读数据流程
（1）客户端通过Distributed FileSystem向NameNode请求下载文件，NameNode通过查询元数据，找到文件块所在的DataNode地址。
（2）挑选一台DataNode（就近原则，然后随机）服务器，请求读取数据。
（3）DataNode开始传输数据给客户端（从磁盘里面读取数据输入流，以Packet为单位来做校验）。
（4）客户端以Packet为单位接收，先在本地缓存，然后写入目标文件。
第5章 NameNode 和 Secondary NameNode
5.1 NN和2NN工作机制
1）第一阶段：NameNode启动
（1）第一次启动NameNode格式化后，创建Fsimage和Edits文件。如果不是第一次启动，直接加载编辑日志和镜像文件到内存。
（2）客户端对元数据进行增删改的请求。
（3）NameNode记录操作日志，更新滚动日志。
（4）NameNode在内存中对元数据进行增删改。
2）第二阶段：Secondary NameNode工作
（1）Secondary NameNode询问NameNode是否需要CheckPoint。直接带回NameNode是否检查结果。
（2）Secondary NameNode请求执行CheckPoint。
（3）NameNode滚动正在写的Edits日志。
（4）将滚动前的编辑日志和镜像文件拷贝到Secondary NameNode。
（5）Secondary NameNode加载编辑日志和镜像文件到内存，并合并。
（6）生成新的镜像文件fsimage.chkpoint。
（7）拷贝fsimage.chkpoint到NameNode。
（8）NameNode将fsimage.chkpoint重新命名成fsimage。
5.2 Fsimage 和 Edits解析
5.2.1 oiv查看Fsimage文件
（1）查看oiv和oev命令
[atguigu@hadoop102 current]$ hdfs
oiv apply the offline fsimage viewer to an fsimage
（2）基本语法
hdfs oiv -p 文件类型 -i镜像文件 -o 转换后文件输出路径
（3）案例实操
[atguigu@hadoop102 current]$ pwd
/opt/module/hadoop-3.1.3/data/tmp/dfs/name/current
[atguigu@hadoop102 current]$ hdfs oiv -p XML -i fsimage_0000000000000000025 -o /opt/module/hadoop-3.1.3/fsimage.xml
[atguigu@hadoop102 current]$ cat /opt/module/hadoop-3.1.3/fsimage.xml
将显示的xml文件内容拷贝到Eclipse中创建的xml文件中，并格式化。部分显示结果如下。
<inode>
<id>16386</id>
<type>DIRECTORY</type>
<name>user</name>
<mtime>1512722284477</mtime>
<permission>atguigu:supergroup:rwxr-xr-x</permission>
<nsquota>-1</nsquota>
<dsquota>-1</dsquota>
</inode>
<inode>
<id>16387</id>
<type>DIRECTORY</type>
<name>atguigu</name>
<mtime>1512790549080</mtime>
<permission>atguigu:supergroup:rwxr-xr-x</permission>
<nsquota>-1</nsquota>
<dsquota>-1</dsquota>
</inode>
<inode>
<id>16389</id>
<type>FILE</type>
<name>wc.input</name>
<replication>3</replication>
<mtime>1512722322219</mtime>
<atime>1512722321610</atime>
<perferredBlockSize>134217728</perferredBlockSize>
<permission>atguigu:supergroup:rw-r--r--</permission>
<blocks>
<block>
<id>1073741825</id>
<genstamp>1001</genstamp>
<numBytes>59</numBytes>
</block>
</blocks>
</inode >
5.2.2 oev 查看Edits文件
（1）基本语法
hdfs oev -p 文件类型 -i编辑日志 -o 转换后文件输出路径
（2）案例实操
[atguigu@hadoop102 current]$ hdfs oev -p XML -i edits_0000000000000000012-0000000000000000013 -o /opt/module/hadoop-3.1.3/edits.xml
[atguigu@hadoop102 current]$ cat /opt/module/hadoop-3.1.3/edits.xml
将显示的xml文件内容拷贝到Eclipse中创建的xml文件中，并格式化。显示结果如下。
<?xml version="1.0" encoding="UTF-8"?>
<EDITS>
<EDITS_VERSION>-63</EDITS_VERSION>
<RECORD>
<OPCODE>OP_START_LOG_SEGMENT</OPCODE>
<DATA>
<TXID>129</TXID>
</DATA>
</RECORD>
<RECORD>
<OPCODE>OP_ADD</OPCODE>
<DATA>
<TXID>130</TXID>
<LENGTH>0</LENGTH>
<INODEID>16407</INODEID>
<PATH>/hello7.txt</PATH>
<REPLICATION>2</REPLICATION>
<MTIME>1512943607866</MTIME>
<ATIME>1512943607866</ATIME>
<BLOCKSIZE>134217728</BLOCKSIZE>
<CLIENT_NAME>DFSClient_NONMAPREDUCE_-1544295051_1</CLIENT_NAME>
<CLIENT_MACHINE>192.168.1.5</CLIENT_MACHINE>
<OVERWRITE>true</OVERWRITE>
<PERMISSION_STATUS>
<USERNAME>atguigu</USERNAME>
<GROUPNAME>supergroup</GROUPNAME>
<MODE>420</MODE>
</PERMISSION_STATUS>
<RPC_CLIENTID>908eafd4-9aec-4288-96f1-e8011d181561</RPC_CLIENTID>
<RPC_CALLID>0</RPC_CALLID>
</DATA>
</RECORD>
<RECORD>
<OPCODE>OP_ALLOCATE_BLOCK_ID</OPCODE>
<DATA>
<TXID>131</TXID>
<BLOCK_ID>1073741839</BLOCK_ID>
</DATA>
</RECORD>
<RECORD>
<OPCODE>OP_SET_GENSTAMP_V2</OPCODE>
<DATA>
<TXID>132</TXID>
<GENSTAMPV2>1016</GENSTAMPV2>
</DATA>
</RECORD>
<RECORD>
<OPCODE>OP_ADD_BLOCK</OPCODE>
<DATA>
<TXID>133</TXID>
<PATH>/hello7.txt</PATH>
<BLOCK>
<BLOCK_ID>1073741839</BLOCK_ID>
<NUM_BYTES>0</NUM_BYTES>
<GENSTAMP>1016</GENSTAMP>
</BLOCK>
<RPC_CLIENTID></RPC_CLIENTID>
<RPC_CALLID>-2</RPC_CALLID>
</DATA>
</RECORD>
<RECORD>
<OPCODE>OP_CLOSE</OPCODE>
<DATA>
<TXID>134</TXID>
<LENGTH>0</LENGTH>
<INODEID>0</INODEID>
<PATH>/hello7.txt</PATH>
<REPLICATION>2</REPLICATION>
<MTIME>1512943608761</MTIME>
<ATIME>1512943607866</ATIME>
<BLOCKSIZE>134217728</BLOCKSIZE>
<CLIENT_NAME></CLIENT_NAME>
<CLIENT_MACHINE></CLIENT_MACHINE>
<OVERWRITE>false</OVERWRITE>
<BLOCK>
<BLOCK_ID>1073741839</BLOCK_ID>
<NUM_BYTES>25</NUM_BYTES>
<GENSTAMP>1016</GENSTAMP>
</BLOCK>
<PERMISSION_STATUS>
<USERNAME>atguigu</USERNAME>
<GROUPNAME>supergroup</GROUPNAME>
<MODE>420</MODE>
</PERMISSION_STATUS>
</DATA>
</RECORD>
</EDITS >
5.3 CheckPoint 时间设置
1）通常情况下，SecondaryNameNode每隔一小时执行一次。
[hdfs-default.xml]
<property>
<name>dfs.namenode.checkpoint.period</name>
<value>3600</value>
</property>
2）一分钟检查一次操作次数，当操作次数达到1百万时，SecondaryNameNode执行一次。
<property>
<name>dfs.namenode.checkpoint.txns</name>
<value>1000000</value>
<description>操作动作次数</description>
</property>
<property>
<name>dfs.namenode.checkpoint.check.period</name>
<value>60</value>
<description> 1分钟检查一次操作次数</description>
</property >
5.4 NameNode 故障处理
NameNode故障后，可以采用如下两种方法恢复数据。
1）将SecondaryNameNode中数据拷贝到NameNode存储数据的目录；
（1）kill -9 NameNode进程
（2）删除NameNode存储的数据（/opt/module/hadoop-3.1.3/data/tmp/dfs/name）
[atguigu@hadoop102 hadoop-3.1.3]$ rm -rf /opt/module/hadoop-3.1.3/data/tmp/dfs/name/*
（3）拷贝SecondaryNameNode中数据到原NameNode存储数据目录
[atguigu@hadoop102 dfs]$ scp -r atguigu@hadoop104:/opt/module/hadoop-3.1.3/data/tmp/dfs/namesecondary/* ./name/
（4）重新启动NameNode
[atguigu@hadoop102 hadoop-3.1.3]$ hdfs --daemon start namenode
2）使用-importCheckpoint选项启动NameNode守护进程，从而将SecondaryNameNode中数据拷贝到NameNode目录中。
（1）修改hdfs-site.xml中的
<property>
<name>dfs.namenode.checkpoint.period</name>
<value>120</value>
</property>
<property>
<name>dfs.namenode.name.dir</name>
<value>/opt/module/hadoop-3.1.3/data/tmp/dfs/name</value>
</property>
（2）kill -9 NameNode进程
（3）删除NameNode存储的数据（/opt/module/hadoop-3.1.3/data/tmp/dfs/name）
[atguigu@hadoop102 hadoop-3.1.3]$ rm -rf /opt/module/hadoop-3.1.3/data/tmp/dfs/name/*
（4）如果SecondaryNameNode不和NameNode在一个主机节点上，需要将SecondaryNameNode存储数据的目录拷贝到NameNode存储数据的平级目录，并删除in_use.lock文件
[atguigu@hadoop102 dfs]$ scp -r atguigu@hadoop104:/opt/module/hadoop-3.1.3/data/tmp/dfs/namesecondary ./
[atguigu@hadoop102 namesecondary]$ rm -rf in_use.lock
[atguigu@hadoop102 dfs]$ pwd
/opt/module/hadoop-3.1.3/data/tmp/dfs
[atguigu@hadoop102 dfs]$ ls
data name namesecondary
（5）导入检查点数据（等待一会ctrl+c结束掉）
[atguigu@hadoop102 hadoop-3.1.3]$ bin/hdfs namenode -importCheckpoint
（6）启动NameNode
[atguigu@hadoop102 hadoop-3.1.3]$ hdfs --daemon start namenode
5.5 集群安全模式
1）基本语法
集群处于安全模式，不能执行重要操作（写操作）。集群启动完成后，自动退出安全模式。
（1）bin/hdfs dfsadmin -safemode get （功能描述：查看安全模式状态）
（2）bin/hdfs dfsadmin -safemode enter （功能描述：进入安全模式状态）
（3）bin/hdfs dfsadmin -safemode leave （功能描述：离开安全模式状态）
（4）bin/hdfs dfsadmin -safemode wait （功能描述：等待安全模式状态）
2）案例
模拟等待安全模式
3）查看当前模式
[atguigu@hadoop102 hadoop-3.1.3]$ hdfs dfsadmin -safemode get
Safe mode is OFF
4）先进入安全模式
[atguigu@hadoop102 hadoop-3.1.3]$ bin/hdfs dfsadmin -safemode enter
5）创建并执行下面的脚本
在/opt/module/hadoop-3.1.3路径上，编辑一个脚本safemode.sh
[atguigu@hadoop102 hadoop-3.1.3]$ touch safemode.sh
[atguigu@hadoop102 hadoop-3.1.3]$ vim safemode.sh
#!/bin/bash
hdfs dfsadmin -safemode wait
hdfs dfs -put /opt/module/hadoop-3.1.3/README.txt
[atguigu@hadoop102 hadoop-3.1.3]$ chmod 777 safemode.sh
[atguigu@hadoop102 hadoop-3.1.3]$ ./safemode.sh
6）再打开一个窗口，执行
[atguigu@hadoop102 hadoop-3.1.3]$ bin/hdfs dfsadmin -safemode leave
7）观察
8）再观察上一个窗口
Safe mode is OFF
9）HDFS集群上已经有上传的数据了。
第6章 DataNode
6.1 DataNode工作机制
（1）一个数据块在DataNode上以文件形式存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据包括数据块的长度，块数据的校验和，以及时间戳。
（2）DataNode启动后向NameNode注册，通过后，周期性（1小时）的向NameNode上报所有的块信息。
（3）心跳是每3秒一次，心跳返回结果带有NameNode给该DataNode的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10分钟没有收到某个DataNode的心跳，则认为该节点不可用。
（4）集群运行中可以安全加入和退出一些机器。
6.2 数据完整性
思考：如果电脑磁盘里面存储的数据是控制高铁信号灯的红灯信号（1）和绿灯信号（0），但是存储该数据的磁盘坏了，一直显示是绿灯，是否很危险？同理DataNode节点上的数据损坏了，却没有发现，是否也很危险，那么如何解决呢？
如下是DataNode节点保证数据完整性的方法。
（1）当DataNode读取Block的时候，它会计算CheckSum。
（2）如果计算后的CheckSum，与Block创建时值不一样，说明Block已经损坏。
（3）Client读取其他DataNode上的Block。
（4）DataNode在其文件创建后周期验证CheckSum。
6.3 掉线时限参数设置
需要注意的是hdfs-site.xml 配置文件中的heartbeat.recheck.interval的单位为毫秒，dfs.heartbeat.interval的单位为秒。
<property>
<name>dfs.namenode.heartbeat.recheck-interval</name>
<value>300000</value>
</property>
<property>
<name>dfs.heartbeat.interval</name>
<value>3</value>
</property>
6.4 服役新数据节点
1）需求
随着公司业务的增长，数据量越来越大，原有的数据节点的容量已经不能满足存储数据的需求，需要在原有集群基础上动态添加新的数据节点。
2）环境准备
（1）在hadoop104主机上再克隆一台hadoop105主机
（2）修改IP地址和主机名称
（3）删除原来HDFS文件系统留存的文件（/opt/module/hadoop-3.1.3/data和log）
（4）source一下配置文件
[atguigu@hadoop105 hadoop-3.1.3]$ source /etc/profile
3）服役新节点具体步骤
（1）直接启动DataNode，即可关联到集群
[atguigu@hadoop105 hadoop-3.1.3]$ hdfs --daemon start datanode
[atguigu@hadoop105 hadoop-3.1.3]$ sbin/yarn-daemon.sh start nodemanager
（2）在hadoop105上上传文件
[atguigu@hadoop105 hadoop-3.1.3]$ hadoop fs -put /opt/module/hadoop-3.1.3/LICENSE.txt /
（3）如果数据不均衡，可以用命令实现集群的再平衡
[atguigu@hadoop102 sbin]$ ./start-balancer.sh
starting balancer, logging to /opt/module/hadoop-3.1.3/logs/hadoop-atguigu-balancer-hadoop102.out
Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved
6.5 退役旧数据节点
6.5.1 白名单退役
——（直接杀死，不推荐）
1）在NameNode的/opt/module/hadoop-3.1.3/etc/hadoop目录下创建dfs.hosts文件
[atguigu@hadoop102 hadoop]$ pwd
/opt/module/hadoop-3.1.3/etc/hadoop
[atguigu@hadoop102 hadoop]$ touch dfs.hosts
[atguigu@hadoop102 hadoop]$ vi dfs.hosts
添加如下主机名称（不添加hadoop105）
hadoop102
hadoop103
hadoop104
2）在NameNode的hdfs-site.xml配置文件中增加dfs.hosts属性
<property>
<name>dfs.hosts</name>
<value>/opt/module/hadoop-3.1.3/etc/hadoop/dfs.hosts</value>
</property>
3）配置文件分发
[atguigu@hadoop102 hadoop]$ xsync hdfs-site.xml
4）刷新NameNode
[atguigu@hadoop102 hadoop-3.1.3]$ hdfs dfsadmin -refreshNodes
Refresh nodes successful
5）更新ResourceManager节点
[atguigu@hadoop102 hadoop-3.1.3]$ yarn rmadmin -refreshNodes
17/06/24 14:17:11 INFO client.RMProxy: Connecting to ResourceManager at hadoop103/192.168.1.103:8033
6）在web浏览器上查看
7）如果数据不均衡，可以用命令实现集群的再平衡
[atguigu@hadoop102 sbin]$ ./start-balancer.sh
starting balancer, logging to /opt/module/hadoop-3.1.3/logs/hadoop-atguigu-balancer-hadoop102.out
Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved
6.5.2 黑名单退役
1）在NameNode的/opt/module/hadoop-3.1.3/etc/hadoop目录下创建dfs.hosts.exclude文件
[atguigu@hadoop102 hadoop]$ pwd
/opt/module/hadoop-3.1.3/etc/hadoop
[atguigu@hadoop102 hadoop]$ touch dfs.hosts.exclude
[atguigu@hadoop102 hadoop]$ vi dfs.hosts.exclude
添加如下主机名称（要退役的节点）
hadoop105
2）在NameNode的hdfs-site.xml配置文件中增加dfs.hosts.exclude属性
<property>
<name>dfs.hosts.exclude</name>
<value>/opt/module/hadoop-3.1.3/etc/hadoop/dfs.hosts.exclude</value>
</property>
3）刷新NameNode、刷新ResourceManager
[atguigu@hadoop102 hadoop-3.1.3]$ hdfs dfsadmin -refreshNodes
Refresh nodes successful
[atguigu@hadoop102 hadoop-3.1.3]$ yarn rmadmin -refreshNodes
17/06/24 14:55:56 INFO client.RMProxy: Connecting to ResourceManager at hadoop103/192.168.1.103:8033
4）检查Web浏览器，退役节点的状态为decommission in progress（退役中），说明数据节点正在复制块到其他节点
5）等待退役节点状态为decommissioned（所有块已经复制完成），停止该节点及节点资源管理器。注意：如果副本数是3，服役的节点小于等于3，是不能退役成功的，需要修改副本数后才能退役
[atguigu@hadoop105 hadoop-3.1.3]$ hdfs --daemon stop datanode
stopping datanode
[atguigu@hadoop105 hadoop-3.1.3]$ sbin/yarn-daemon.sh stop nodemanager
stopping nodemanager
6）如果数据不均衡，可以用命令实现集群的再平衡
[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-balancer.sh
starting balancer, logging to /opt/module/hadoop-3.1.3/logs/hadoop-atguigu-balancer-hadoop102.out
Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved
注意：不允许白名单和黑名单中同时出现同一个主机名称。
6.6 DataNode 多目录配置
1）DataNode也可以配置成多个目录，每个目录存储的数据不一样。即：数据不是副本
2）具体配置如下
[hdfs-site.xml]
<property>
<name>dfs.datanode.data.dir</name>
<value>file:///${hadoop.tmp.dir}/dfs/data1,file:///${hadoop.tmp.dir}/dfs/data2</value>
</property>
二、MapReduce&Yarn
第1章 MapReduce概述
1.1 MapReduce定义
MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架。
MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上。
1.2 MapReduce优缺点
1.2.1 优点
1）MapReduce易于编程
它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的服务器上运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。就是因为这个特点使得MapReduce编程变得非常流行。
2）良好的扩展性
当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。
3）高容错性
MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由Hadoop内部完成的。
4）适合PB级以上海量数据的离线处理
可以实现上千台服务器集群并发工作，提供数据处理能力。
1.2.2 缺点
1）不擅长实时计算
MapReduce无法像MySQL一样，在毫秒或者秒级内返回结果。
2）不擅长流式计算
流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。
3）不擅长DAG（有向图）计算
多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。
1.3 MapReduce核心思想
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cTBWS7Gb-1625191603054)(img/1597288759247.png)]
（1）分布式的运算程序往往需要分成至少2个阶段。
（2）第一个阶段的MapTask并发实例，完全并行运行，互不相干。
（3）第二个阶段的ReduceTask并发实例互不相干，但是他们的数据依赖于上一个阶段的所有MapTask并发实例的输出。
（4）MapReduce编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个MapReduce程序，串行运行。
总结：分析WordCount数据流走向深入理解MapReduce核心思想。
1.4 MapReduce进程
一个完整的MapReduce程序在分布式运行时有三类实例进程：
（1）MrAppMaster：负责整个程序的过程调度及状态协调。
（2）MapTask：负责Map阶段的整个数据处理流程。
（3）ReduceTask：负责Reduce阶段的整个数据处理流程。
1.5 官方WordCount源码
采用反编译工具反编译源码，发现WordCount案例有Map类、Reduce类和驱动类。且数据的类型是Hadoop自身封装的序列化类型。
1.6 常用数据序列化类型
| Java 类型 | **Hadoop Writable ** 类型 |
|---|---|
| Boolean | BooleanWritable |
| Byte | ByteWritable |
| Int | IntWritable |
| Float | FloatWritable |
| Long | LongWritable |
| Double | DoubleWritable |
| String | Text |
| Map | MapWritable |
| Array | ArrayWritable |
1.7 WordCount案例实操
1）创建maven工程
2）添加依赖
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
</dependency>
</dependencies>
3）创建log4j2.xml
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="error" strict="true" name="XMLConfig">
<Appenders>
<!-- 类型名为Console，名称为必须属性 -->
<Appender type="Console" name="STDOUT">
<!-- 布局为PatternLayout的方式，
输出样式为[INFO] [2018-01-22 17:34:01][org.test.Console]I'm here -->
<Layout type="PatternLayout"
pattern="[%p] [%d{yyyy-MM-dd HH:mm:ss}][%c{10}]%m%n" />
</Appender>
</Appenders>
<Loggers>
<!-- 可加性为false -->
<Logger name="test" level="info" additivity="false">
<AppenderRef ref="STDOUT" />
</Logger>
<!-- root loggerConfig设置 -->
<Root level="info">
<AppenderRef ref="STDOUT" />
</Root>
</Loggers>
</Configuration>
4）编写程序
//<Mapper类>
package com.atguigu.mapreduce;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
Text k = new Text();
IntWritable v = new IntWritable(1);
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 1 获取一行
String line = value.toString();
// 2 切割
String[] words = line.split(" ");
// 3 输出
for (String word : words) {
k.set(word);
context.write(k, v);
}
}
}
//<Reducer类>
package com.atguigu.mapreduce.wordcount;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
int sum;
IntWritable v = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
// 1 累加求和
sum = 0;
for (IntWritable count : values) {
sum += count.get();
}
// 2 输出
v.set(sum);
context.write(key,v);
}
}
<Driver驱动类>
package com.atguigu.mapreduce.wordcount;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordcountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 1 获取配置信息以及封装任务
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);
// 2 设置jar加载路径
job.setJarByClass(WordcountDriver.class);
// 3 设置map和reduce类
job.setMapperClass(WordcountMapper.class);
job.setReducerClass(WordcountReducer.class);
// 4 设置map输出
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
// 5 设置最终输出kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 6 设置输入和输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7 提交
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
5）本地测试
（1）需要首先配置好HadoopHome变量以及Windows运行依赖
（2）在Eclipse/Idea上运行程序
6）集群测试
（1）用maven打jar包，需要添加的打包插件依赖
注意：标记红颜色的部分需要替换为自己工程主类
<build>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin </artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>com.atguigu.mr.WordcountDriver</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
（2）将程序打成jar包，然后拷贝到Hadoop集群中
步骤详情：右键->Run as->maven install。等待编译完成就会在项目的target文件夹中生成jar包。如果看不到。在项目上右键-》Refresh，即可看到。修改不带依赖的jar包名称为wc.jar，并拷贝该jar包到Hadoop集群。
（3）启动Hadoop集群
（4）执行WordCount程序
7）windows集群测试
（1）添加必要配置信息
public class WordcountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 1 获取配置信息以及封装任务
Configuration configuration = new Configuration();
configuration.set("fs.defaultFS", "hdfs://hadoop102:8020");
configuration.set("mapreduce.framework.name","yarn");
configuration.set("mapreduce.app-submission.cross-platform","true");
configuration.set("yarn.resourcemanager.hostname","hadoop103");
Job job = Job.getInstance(configuration);
// 2 设置jar加载路径
job.setJarByClass(WordcountDriver.class);
// 3 设置map和reduce类
job.setMapperClass(WordcountMapper.class);
job.setReducerClass(WordcountReducer.class);
// 4 设置map输出
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
// 5 设置最终输出kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 6 设置输入和输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7 提交
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
（2）编辑任务配置
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zJbvTp02-1625191603055)(img/1597289968385.png)]
（3）打包，并将Jar包设置到Driver中
public class WordcountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 1 获取配置信息以及封装任务
Configuration configuration = new Configuration();
configuration.set("fs.defaultFS", "hdfs://hadoop102:8020");
configuration.set("mapreduce.framework.name","yarn");
configuration.set("mapreduce.app-submission.cross-platform","true");
configuration.set("yarn.resourcemanager.hostname","hadoop103");
Job job = Job.getInstance(configuration);
// 2 设置jar加载路径
job.setJar("C:\Users\skiin\IdeaProjects\mapreduce1021\target\mapreduce1021-1.0-SNAPSHOT.jar");
// 3 设置map和reduce类
job.setMapperClass(WordcountMapper.class);
job.setReducerClass(WordcountReducer.class);
// 4 设置map输出
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
// 5 设置最终输出kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 6 设置输入和输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7 提交
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
（4）提交并查看结果
第2章 Hadoop序列化
2.1 序列化概述
2.2 自定义bean对象实现序列化接口（Writable）
具体实现bean对象序列化步骤如下7步。
（1）必须实现Writable接口
（2）反序列化时，需要反射调用空参构造函数，所以必须有空参构造
public FlowBean() {
super();
}
（3）重写序列化方法
@Override
public void write(DataOutput out) throws IOException {
out.writeLong(upFlow);
out.writeLong(downFlow);
out.writeLong(sumFlow);
}
（4）重写反序列化方法
@Override
public void readFields(DataInput in) throws IOException {
upFlow = in.readLong();
downFlow = in.readLong();
sumFlow = in.readLong();
}
（5）注意反序列化的顺序和序列化的顺序完全一致
（6）要想把结果显示在文件中，需要重写toString()，可用”t”分开，方便后续用。
（7）如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口/继承Comparator类，因为MapReduce框中的Shuffle过程要求对key必须能排序。详见后面排序案例。
第3章 MapReduce框架原理
3.1 InputFormat数据输入
3.1.1 切片与MapTask并行度决定机制
1）问题引出
MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。
思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？
2）MapTask并行度决定机制
**数据块：**Block是HDFS物理上把数据分成一块一块。
**数据切片：**数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。
3.1.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.1.3 InputFormat数据输入
1）InputFormat : 抽象类 数据输入
——getSplits( ) : 生成切片方法
——creatRecordReader( ) : 创建RecordReader对象，真正负责数据读取的对象
2）主要抽象子类：FileInputFormat
——getSplits( ) : 做出了具体的实现.
——createRecordReader( ) : 没有做任何的改动.
——isSplitable( ) : 当前输入的数据集是否可切分.
3）具体实现类：
——TextInputFormat : MapReduce默认使用的InputFormat
——CombineTextInputFormat
3.1.4 FileInputFormat切片机制
1）切片机制：
（1）简单地按照文件的内容长度进行排序；
（2）切片大小默认等于Block大小；
（3）切片时不考虑数据集整体，而是逐个对每一个文件单独切片。
2）切片大小计算公式
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
==>return Math.max(minSize, Math.min(maxSize, blockSize));
minSize默认大小为1，maxSize默认大小为Long型最大值
3）修改切片大小配置
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); // 1
相关配置项: “mapreduce.input.fileinputformat.split.minsize”=“0”
long maxSize = getMaxSplitSize(job); // Long.MAX_VALUE
相关配置项: “mapreduce.input.fileinputformat.split.maxsize” 默认没有配置这一项
4）切片源码
long blockSize = file.getBlockSize(); //获取文件的块大小
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) //SPLIT_SLOP=1
5）切片总结
（1） 每个切片都需要由一个MapTask来处理 ， 也就意味着在一个MR中，有多少个切片，就会有多少个MapTask。
（2）切片的大小默认情况下等于块的大小（默认128M）
（3）切片的时候每个文件单独切片，不会整体切片.
（4）切片的个数不是越多越好，也不是越少越少，按照实际情况，处理成合适的切片数.
6）TextInputFormat实现类
TextInputFormat : 默认使用的InputFormat类
切片的规则: 用的就是父亲FileInputFormat中的切片规则.
读取数据: LineRecordReader 按行读取数据
3.1.5 CombineFileInputFormat切片机制
1) 用于小文件过多的场景，解决过多的小文件最终生成太多的切片的问题
2) 在Driver中设置
job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job,20971520); //20M
3) 会按照设置的虚拟存储大小进行输入数据的逻辑上的规划
（1）如果文件的大小小于MaxInputSplitSize ， 则文件规划成一个；
（2）如果文件的大小 大于MaxInputSplitSize ，但是小于 MaxInputSplitSize*2 ， 则文件规划成两个(对半分)；
（3）如果文件的大小大于 MaxInputSplitSize2 ，先按照MaxInputSplitSize的大小规划一个， 剩余的再进行规划.
最终按照 MaxInputSplitSize 大小来生成切片；
（4）将规划好的每个虚拟文件逐个累加，只要不超过 MaxInputSplitSize大小，则都是规划到一个切片中的。
3.2 MapReduce工作流程
上面的流程是整个MapReduce最全工作流程，但是Shuffle过程只是从第7步开始到第16步结束，具体Shuffle过程详解，如下：
（1）MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中
（2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
（3）多个溢出文件会被合并成大的溢出文件
（4）在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序
（5）ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据
（6）ReduceTask会取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）
（7）合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）
注意：
（1）Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。
（2）缓冲区的大小可以通过参数调整，参数：io.sort.mb默认100M。
3.3 Shuffle机制
3.3.1 Shuffle流程
map( )方法之后，reduce( )方法之前的数据处理过程称为Shuffle
1). map方法之后， reduce方法之前的处理过程就是shuffle过程.
2). map方法写出去的kv， 会被一个收集线程收集到缓冲区中.
3). 缓冲区的大小默认是100M,达到80%发生溢写.
4). 缓冲区中记录的是 kv 、 kv的下标 、 kv的分区 等.
5). 溢写的时候，是按照kv的分区进行排序(快排，只排索引) ，再按照分区溢写. -->map端的第一次排序
6). 每个MapTask有可能发生多次溢写，最终需要将多次溢写的文件归并成一个大的文件. --> map端的第二次排序.
7). 在溢写 和归并过程中，都可以采用combiner.(用于分组合并数据，节约磁盘空间)
8). 每个ReduceTask按照所要处理的分区， 到每个MapTask中拷贝对应的分区的数据.
拷贝过程中，先放内存，放不下写磁盘. 等数据全部都拷贝过来后，进行归并排序。 --> reduce端的排序
9). reduce端排好序的数据进行分组，然后进入reduce方法进行业务处理.
3.3.2 Partition分区
1）分区概念：
用以按照条件将不同的数据输出到不同的文件当中
2）默认分区：
//获取ReduceTask的数量，默认为1
partitions = jobContext.getNumReduceTasks();
partitioner = new org.apache.hadoop.mapreduce.Partitioner<K,V>() {
@Override
public int getPartition(K key, V value, int numPartitions) {
//获取分区号 此处为0
return partitions - 1;
}
}
默认分区器：HashPartitioner
默认的分区器就会按照key的hashcode值 先对Integer的最大值做&运算，在对reduce的个数取余，得到分区号。
//有多个分区时（ReduceTask为多个）
public int getPartition(K key, V value,int numReduceTasks) {
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
分区总结：
（1）如果ReduceTask数量 > getPartition的结果，会产生多个空的输出文件part-r-000xx;
（2）如果1 < ReduceTask数量 < getPartition的结果，则有一部分数据无处安放，会Exception；
（3）如果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终只会交给一个ReduceTask，因此也只会生成一个part-r-00000文件；
（4）分区必须从零开始，逐一累加。
3）数据分区方式
（1）数据的分区由分区器(Partitioner)来决定.
（2） hadoop有默认的分区器对象 HashPartitioner .
HashPartitioner会按照k的hash值对Reduce的个数进行取余操作得到k所对应的分区.
（3） hadoop也支持用户自定义分区器
4）自定义分区器
3.3.3 自定义排序的两种方式
1）排序概述
（1）MapTask和ReduceTask均会对数据按照key进行排序，默认是按照字典序进行排序；
（2）环形缓冲区溢写到磁盘时使用快速排序，多个溢写文件合并、多个MapTask输出文件合并时使用归并排序。
2）自定义排序——WritableComparable接口
（1）自定义bean对象需实现WritableComparable接口
public class FlowBean implements WritableComparable<FlowBean> {}
（2）在自定义bean类内重写compareTo方法
@Override
public int compareTo(FlowBean o) {
//指定在缓形缓冲区时通过哪个属性进行排序
return Long.compare(this.sumFlow,o.sumFlow);
}
3）自定义排序——WritableComparator类
（1）自定义比较器类继承WritableComparator
public class MyComparator extends WritableComparator {}
（2）调用父类有参构造器
public MyComparator(){
//需要调用父类中的有参构造器
/*
第一个参数 ：排序的数据的类型
第二个参数 ：排序的数据是否每一个都需要创建一个新的对象
*/
super(FlowBean.class,true);
}
（3）重写compare方法
@Override
public int compare(WritableComparable a, WritableComparable b) {
FlowBean f1 = (FlowBean) a;
FlowBean f2 = (FlowBean) b;
//按照上传流量的大小排序
return Long.compare(f1.getUpFlow(),f2.getUpFlow());
}
（4）在Driver类中指定自定义比较器类
//设置排序的自定义类
job.setSortComparatorClass(MyComparator.class);
4）排序方式对比
（1）当上面的两种方式都使用的情况下下WritableComparator起作用；
（2）就算使用WritableComparator 那么JavaBean也得实现WritableComparable接口；
（3）如果继承WritableComparator中没有重写方法那么默认调用的是WritableComparable的比较方法。
3.3.4 Combiner合并
1）Combiner概述
2）自定义Combiner实现
（1）自定义一个Combiner继承Reducer，重写Reduce方法；
public class MyCombiner extends Reducer<Text, IntWritable,Text,IntWritable>{
//泛型类型应与Mapper输出泛型类型一致
private IntWritable outv = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int sum = 0;
//将当前单词出现的次数进行汇总
for (IntWritable value : values) {
sum += value.get();
}
outv.set(sum);
//写出
context.write(key,outv);
}
}
（2）在Job驱动类中设置
//Driver中指定Combiner类
job.setCombinerClass(MyCombiner.class);
//需求合理情况下可直接指定WordCountReducer作为Combiner类
job.setCombinerClass(WordcountReducer.class);
3.4 MapTask工作机制
**（1）Read阶段：**MapTask通过用户编写的RecordReader，从输入InputSplit中解析出一个个key/value。
**（2）Map阶段：**该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。
**（3）Collect收集阶段：**在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。
**（4）Spill阶段：**即“溢写”，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。
溢写阶段详情：
步骤1：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号Partition进行排序，然后按照key进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。
步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。
步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。
**（5）Combine阶段：**当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。
当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。
在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。
让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。
3.5 ReduceTask工作机制
3.5.1 工作机制
（1）Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
**（2）Merge阶段：**在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。
**（3）Sort阶段：**按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。
**（4）Reduce阶段：**reduce()函数将计算结果写到HDFS上。
3.5.2 ReduceTask并行度测试
1）设置ReduceTask并行度（个数）
ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：
// 默认值是1，手动设置为4
job.setNumReduceTasks(4);
实验：测试ReduceTask多少合适（MapTask=16，数据量为1GB）
（1）实验环境：1个Master节点，16个Slave节点：CPU:8GHZ，内存: 2G
（2）实验结论：
| ReduceTask | 1 | 5 | 10 | 15 | 16 | 20 | 25 | 30 | 45 | 60 |
|---|---|---|---|---|---|---|---|---|---|---|
| 总时间 | 892 | 146 | 110 | 92 | 88 | 100 | 128 | 101 | 145 | 104 |
3）注意事项
3.6 OutputFormat数据输出
3.6.1 OutputFormat概述
3.6.2 自定义OutputFormat
1）自定义类继承FileOutputFormat
重写getRecordWriter方法，将配置对象conf 作为构造器形参传入到自定义MyRecordWriter 中
/*
泛型：FileOutputFormat<K, V>
K : 写出数据的偏移量
V : 写出的数据
*/
public class MyOutputFormat extends FileOutputFormat<LongWritable, Text> {
/*
返回RecordWriter类型的对象
*/
@Override
public RecordWriter<LongWritable, Text> getRecordWriter(TaskAttemptContext job)
throws IOException, InterruptedException {
//将conf传到MyrecordWriter构造器中
return new MyRecordWriter(job.getConfiguration());
}
}
2）自定义RecordWriter类
（1）创建文件系统对象，通过形参获取配置文件；
（2）通过文件系统对象创建流；
（3）定义输出流对象并创建输出路径；
（4）重写write方法，根据文件内容输出不同文件；
public class MyRecordWriter extends RecordWriter<LongWritable, Text> {
//定义FSData..具体子类目的为调用其更全的方法，OutputStream抽象类中方法较少
private FSDataOutputStream atguiguFS = null;
private FSDataOutputStream otherFS = null;
//开流 ：用文件系统FileSystem
public MyRecordWriter(Configuration conf) throws IOException {
//需要该形参的目的是为了拿配置文件
//创建文件系统的对象为了创建流
FileSystem fs = FileSystem.get(conf);
//开流
atguiguFS = fs.create(new Path("D:\atguigu.txt"));
otherFS = fs.create(new Path("D:\other.txt"));
}
/*
将数据写出
key : 写出数据的偏移量
value : 写出的数据
*/
@Override
public void write(LongWritable key, Text value) throws IOException, InterruptedException {
String str = value.toString();
//判断是否包含atguigu
if (str.contains("atguigu")){
//输出到atguigu.txt文件中
atguiguFS.write((str+"n").getBytes());
}else{
//输出到other.txt文件中
otherFS.write((str+"n").getBytes());
}
}
/*
关闭资源
*/
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
IOUtils.closeStream(atguiguFS);
IOUtils.closeStream(otherFS);
}
}
3）驱动类中指定
（1）驱动类中指定自定义输出类；
（2）指定文件输入输出路径（不可以省略，还有其它文件_SUCCESS要输出）
//设置自定义的OutputFormat
job.setOutputFormatClass(MyOutputFormat.class);
//设置输入输出路径
FileInputFormat.setInputPaths(job,new Path("D:\demo\input"));
FileOutputFormat.setOutputPath(job,new Path("D:\demo\output"));
3.7 Join的多种应用
3.7.1 Reduce Join
1）实现原理
2）案例实操
（1）需求说明
表一：订单数据表 order
| id | pid | amount |
|---|---|---|
| 1001 | 01 | 1 |
| 1002 | 02 | 2 |
| 1003 | 03 | 3 |
| 1004 | 01 | 4 |
| 1005 | 02 | 5 |
| 1006 | 03 | 6 |
表二：商品信息表 product
| pid | pname |
|---|---|
| 01 | 小米 |
| 02 | 华为 |
| 03 | 格力 |
需求表：最终数据形式
| id | pname | amount |
|---|---|---|
| 1001 | 小米 | 1 |
| 1004 | 小米 | 4 |
| 1002 | 华为 | 2 |
| 1005 | 华为 | 5 |
| 1003 | 格力 | 3 |
| 1006 | 格力 | 6 |
（2）需求分析
（3）代码实现
A. 创建javabean用来存储两张表join后的结果
//自定义类实现WritableComparable接口
public class OrderBean implements WritableComparable<OrderBean> {
//定义表中所有信息属性
private String pid;
private String pname;
private String id;
private int amount;
//所有属性的set/get方法
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public String getPname() {
return pname;
}
public void setPname(String pname) {
this.pname = pname;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public int getAmount() {
return amount;
}
public void setAmount(int amount) {
this.amount = amount;
}
//必须提供一个空参构造器方便创建对象
public OrderBean(){
}
public OrderBean(String id, String pid, String pname, int amount) {
this.pid = pid;
this.pname = pname;
this.id = id;
this.amount = amount;
}
/*
排序方式 - 先按照pid排序，pid相同按照pname排序
*/
@Override
public int compareTo(OrderBean o) {
int pd = this.pid.compareTo(o.pid);
if (pd == 0){
return -this.pname.compareTo(o.pname);
}
return pd;
}
/*
序列化调用的方法
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(pid);
out.writeUTF(pname);
out.writeUTF(id);
out.writeInt(amount);
}
/*
反序列化调用的方法
*/
@Override
public void readFields(DataInput in) throws IOException {
pid = in.readUTF();
pname = in.readUTF();
id = in.readUTF();
amount = in.readInt();
}
/*
最终输出内容的格式
*/
@Override
public String toString() {
return id + " " + pname + " " + amount;
}
}
B. 编写JoinMapper类
public class JoinMapper extends Mapper<LongWritable, Text,Text,Order> {
private String filename;
@Override
protected void setup(Context context) throws IOException, InterruptedException {
//获取切片对象（切片信息中有源文件路径和文件名）
FileSplit fs = (FileSplit) context.getInputSplit();
//获取文件名
filename = fs.getPath().getName();
}
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] order = line.split("t");
//判断是哪个文件中的内容
if ("order.txt".equals(fileName)){
//封装k --- String id, String pid, String pname, int amount
OrderBean orderBean = new OrderBean(order[0], order[1], "",
Integer.parseInt(order[2]));
//写出去
context.write(orderBean,NullWritable.get());
}else if("pd.txt".equals(fileName)){
//封装k
//封装k --- String id, String pid, String pname, int amount
OrderBean orderBean = new OrderBean("", order[0], order[1], 0);
//写出去
context.write(orderBean,NullWritable.get());
}
}
}
C. 编写JoinReducer类
public class JoinReducer extends Reducer<OrderBean, NullWritable,OrderBean,NullWritable> {
/*
null 1 null 小米
1001 1 1 null
1004 1 4 null
*/
@Override
protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
Iterator<NullWritable> iterator = values.iterator();
//这个方法本身返回value值，key值也会发生改变-变成value对应的key
iterator.next();
//获取第一条数据的key
String pname = key.getPname();
//遍历第二条及以后的所有数据
while(iterator.hasNext()){
iterator.next();//每next一下key值就会变成value所对应的key值数据
key.setPname(pname);
//写出该数据
context.write(key,NullWritable.get());
}
}
}
D. 自定义分组比较器（由于bean中定义了两种排序方法，会分成多个组，所以此处需要重新定义）
//分组比较器需继承WritableComparator类，并重写compare方法
public class MyGroupingCompartor extends WritableComparator {
//调用父类构造器，指定排序类，true为每次排序创建一个新的对象
public MyGroupingCompartor(){
super(OrderBean.class,true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
OrderBean o1 = (OrderBean) a;
OrderBean o2 = (OrderBean) b;
//按照pid分组
return o1.getPid().compareTo(o2.getPid());
}
}
E. 在驱动类中指定分组比较器
public class JoinDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Job job = Job.getInstance(new Configuration());
job.setJarByClass(JoinDriver.class);
job.setMapperClass(JoinMapper.class);
job.setReducerClass(JoinReducer.class);
job.setMapOutputKeyClass(OrderBean.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(OrderBean.class);
job.setOutputValueClass(NullWritable.class);
//设置分组比较器 --- 默认的是按照缓形缓冲区的排序方式分组。
job.setGroupingComparatorClass(MyGroupingCompartor.class);
//读取的文件
FileInputFormat.setInputPaths(job,
new Path("D:/input"));
//输出路径
FileOutputFormat.setOutputPath(job,new Path("D:/output/"));
job.waitForCompletion(true);
}
}
3.7.2 Mapper Join
1）使用场景
Map Join适用于一张表十分小、一张表很大的场景。
2）优点
思考：在Reduce端处理过多的表，非常容易产生数据倾斜。怎么办？
在Map端缓存多张表，提前处理业务逻辑，这样增加Map端业务，减少Reduce端数据的压力，尽可能的减少数据倾斜。
3）具体办法：采用DistributedCache
（1）在Mapper的setup阶段，将文件读取到缓存集合中。
（2）在驱动函数中加载缓存。
// 缓存普通文件到Task运行节点。
job.addCacheFile(new URI("file://e:/cache/pd.txt"));
4）代码实现
A. 在驱动类Driver中添加缓存文件路径
public class DistributedCacheDriver {
public static void main(String[] args) throws Exception {
// 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:///e:/input/inputcache/pd.txt"));
// 7 Map端Join的逻辑不需要Reduce阶段，设置reduceTask数量为0
job.setNumReduceTasks(0);
// 8 提交
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
B. 同样需要一个JavaBean容器来存放Join后的结果
//自定义类实现WritableComparable接口
public class OrderBean implements WritableComparable<OrderBean> {
//定义表中所有信息属性
private String pid;
private String pname;
private String id;
private int amount;
//所有属性的set/get方法
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public String getPname() {
return pname;
}
public void setPname(String pname) {
this.pname = pname;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public int getAmount() {
return amount;
}
public void setAmount(int amount) {
this.amount = amount;
}
//必须提供一个空参构造器方便创建对象
public OrderBean(){
}
public OrderBean(String id, String pid, String pname, int amount) {
this.pid = pid;
this.pname = pname;
this.id = id;
this.amount = amount;
}
/*
排序方式 - 先按照pid排序，pid相同按照pname排序
*/
@Override
public int compareTo(OrderBean o) {
int pd = this.pid.compareTo(o.pid);
if (pd == 0){
return -this.pname.compareTo(o.pname);
}
return pd;
}
/*
序列化调用的方法
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(pid);
out.writeUTF(pname);
out.writeUTF(id);
out.writeInt(amount);
}
/*
反序列化调用的方法
*/
@Override
public void readFields(DataInput in) throws IOException {
pid = in.readUTF();
pname = in.readUTF();
id = in.readUTF();
amount = in.readInt();
}
/*
最终输出内容的格式
*/
@Override
public String toString() {
return id + " " + pname + " " + amount;
}
}
C.Mapper类：
① 重写setup方法（在最初只执行一次）中获取到缓存文件内容，并存放到HashMap集合中；
②重写Map方法，读取并封装outK，最后写出context;
public class JoinMapper extends Mapper<LongWritable, Text,OrderBean, NullWritable> {
//用来存储 1 小米 （pd.txt中的内容）
private Map<String,String> map = new HashMap<String,String>();
/*
setup方法只调用一次，我们只需要调用一次的代码或者初始化的代码都可以放在该方法中
缓存pd.txt
*/
@Override
protected void setup(Context context) throws IOException, InterruptedException {
//创建文件系统
FileSystem fs = FileSystem.get(context.getConfiguration());
//开流
//获取缓存文件路径
URI[] cacheFiles = context.getCacheFiles();
//创建输入流
FSDataInputStream fis = fs.open(new Path(cacheFiles[0]));
//一行一行的读取数据 ：BufferedReader
BufferedReader br = new BufferedReader(new InputStreamReader(fis,"UTF-8"));
String line = null;
while((line = br.readLine()) != null){
//切割数据
String[] lineInfo = line.split("t");
map.put(lineInfo[0],lineInfo[1]);
}
//关资源
IOUtils.closeStream(br);
IOUtils.closeStream(fis);
fs.close();
}
/*
读取order.txt
*/
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//1006 03 6
String line = value.toString();
String[] order = line.split("t");
//封装K
//String id,String pid, String pname, int amount
OrderBean orderBean = new OrderBean(order[0], order[1],
map.get(order[1]), Integer.parseInt(order[2]));
//写出去
context.write(orderBean,NullWritable.get());
}
//关闭资源
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
}
}
3.8 计数器应用
3.9 数据清洗（ETL）
1) 清洗说明
清理的过程往往只需要运行Mapper程序，不需要运行Reduce程序。
2）案例说明
（1）需求：去除日志中字段个数小于等于11的日志
（2）代码实现：
public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable>{
Text k = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 1 获取1行数据
String line = value.toString();
// 2 解析日志
boolean result = parseLog(line,context);
// 3 日志不合法退出
if (!result) {
return;
}
// 4 设置key
k.set(line);
// 5 写出数据
context.write(k, NullWritable.get());
}
// 2 解析日志
private boolean parseLog(String line, Context context) {
// 1 截取
String[] fields = line.split(" ");
// 2 日志长度大于11的为合法
if (fields.length > 11) {
// 系统计数器
context.getCounter("map", "true").increment(1);
return true;
}else {
context.getCounter("map", "false").increment(1);
return false;
}
}
}
3.10 MapReduce开发总结
第4章 Yarn资源调度器
Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而MapReduce等运算程序则相当于运行于操作系统之上的应用程序。
4.1 Yarn基本架构
4.2 Yarn工作机制
（1）MR程序提交到客户端所在的节点。
（2）YarnRunner向ResourceManager申请一个Application。
（3）RM将该应用程序的资源路径返回给YarnRunner。
（4）该程序将运行所需资源提交到HDFS上。
（5）程序资源提交完毕后，申请运行mrAppMaster。
（6）RM将用户的请求初始化成一个Task。
（7）其中一个NodeManager领取到Task任务。
（8）该NodeManager创建容器Container，并产生MRAppmaster。
（9）Container从HDFS上拷贝资源到本地。
（10）MRAppmaster向RM 申请运行MapTask资源。
（11）RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（12）MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。
（13）MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。
（14）ReduceTask向MapTask获取相应分区的数据。
（15）程序运行完毕后，MR会向RM申请注销自己。
4.3 作业提交全过程
4.3.1 作业提交之Yarn
作业提交全过程详解
（1）作业提交
第1步：Client调用job.waitForCompletion方法，向整个集群提交MapReduce作业。
第2步：Client向RM申请一个作业id。
第3步：RM给Client返回该job资源的提交路径和作业id。
第4步：Client提交jar包、切片信息和配置文件到指定的资源提交路径。
第5步：Client提交完资源后，向RM申请运行MrAppMaster。
（2）作业初始化
第6步：当RM收到Client的请求后，将该job添加到容量调度器中。
第7步：某一个空闲的NM领取到该Job。
第8步：该NM创建Container，并产生MRAppmaster。
第9步：下载Client提交的资源到本地。
（3）任务分配
第10步：MrAppMaster向RM申请运行多个MapTask任务资源。
第11步：RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（4）任务运行
第12步：MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。
第13步：MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。
第14步：ReduceTask向MapTask获取相应分区的数据。
第15步：程序运行完毕后，MR会向RM申请注销自己。
（5）进度和状态更新
YARN中的任务将其进度和状态(包括counter)返回给应用管理器, 客户端每秒(通过mapreduce.client.progressmonitor.pollinterval设置)向应用管理器请求进度更新, 展示给用户。
（6）作业完成
除了向应用管理器请求作业进度外, 客户端每5秒都会通过调用waitForCompletion()来检查作业是否完成。时间间隔可以通过mapreduce.client.completion.pollinterval来设置。作业完成之后, 应用管理器和Container会清理工作状态。作业的信息会被作业历史服务器存储以备之后用户核查。
4.3.2 作业提交过程之MapReduce
4.4 资源调度器
目前，Hadoop作业调度器主要有三种：FIFO、Capacity Scheduler和Fair Scheduler。Hadoop3.1.3默认的资源调度器是Capacity Scheduler。
具体设置详见：yarn-default.xml文件
<property>
<description>The class to use as the resource scheduler.</description>
<name>yarn.resourcemanager.scheduler.class</name> <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>
4.4.1 先进先出调度器（FIFO）
4.4.2 容量调度器（Capacity Scheduler)
4.4.3 公平调度器（Fair Scheduler）
4.5 容量调度器多队列提交案例
4.5.1 需求
Yarn默认的容量调度器是一条单队列的调度器，在实际使用中会出现单个任务阻塞整个队列的情况。同时，随着业务的增长，公司需要分业务限制集群使用率。这就需要我们按照业务种类配置多条任务队列。
4.5.2 配置多队列的容量调度器
默认Yarn的配置下，容量调度器只有一条Default队列。在capacity-scheduler.xml中可以配置多条队列，并降低default队列资源占比：
<property>
<name>yarn.scheduler.capacity.root.queues</name>
<value>default,hive</value>
<description>
The queues at the this level (root is the root queue).
</description>
</property>
<property>
<name>yarn.scheduler.capacity.root.default.capacity</name>
<value>40</value>
</property>
同时为新加队列添加必要属性：
<property>
<name>yarn.scheduler.capacity.root.hive.capacity</name>
<value>60</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.user-limit-factor</name>
<value>1</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.maximum-capacity</name>
<value>80</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.state</name>
<value>RUNNING</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.acl_submit_applications</name>
<value>*</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.acl_administer_queue</name>
<value>*</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.acl_application_max_priority</name>
<value>*</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.maximum-application-lifetime</name>
<value>-1</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.hive.default-application-lifetime</name>
<value>-1</value>
</property>
在配置完成后，source配置文件，重启Yarn，就可以看到两条队列：
4.5.3 向Hive队列提交任务
默认的任务提交都是提交到default队列的。如果希望向其他队列提交任务，需要在Driver中声明：
public class WcDrvier {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration configuration = new Configuration();
configuration.set("mapred.job.queue.name", "hive");
//1. 获取一个Job实例
Job job = Job.getInstance(configuration);
//2. 设置类路径
job.setJarByClass(WcDrvier.class);
//3. 设置Mapper和Reducer
job.setMapperClass(WcMapper.class);
job.setReducerClass(WcReducer.class);
//4. 设置Mapper和Reducer的输出类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setCombinerClass(WcReducer.class);
//5. 设置输入输出文件
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//6. 提交Job
boolean b = job.waitForCompletion(true);
System.exit(b ? 0 : 1);
}
}
这样，这个任务在集群提交时，就会提交到hive队列：
三、Hadoop优化与新特性
第1章 数据压缩
1.1 概述
压缩是提高hadoop运行效率的一种优化策略，在Mapper和Reducer过程中国使用压缩可以减少磁盘IO；
注：使用压缩技术可以减少磁盘的IO，但是也增加了CPU的运行负担，因此需要根据具体实际应用场景来选择是否压缩。
压缩基本原则：
① 运算密集型的job，少用压缩；
②IO密集型的job，多用压缩。
1.2 MR支持的压缩编码
1.2.1 常见的压缩方式
| 压缩格式 | 自带 | 算法 | 文件扩展名 | 是否可切分 | 换成压缩格式，原来的程序是否需修改 |
|---|---|---|---|---|---|
| DEFLATE | 是 | DEFLATE | .deflate | 否 | 和文本处理一样，不需要修改 |
| Gzip | 是 | DEFLATE | .gz | 否 | 和文本处理一样，不需要修改 |
| bzip2 | 是 | bzip2 | .bz2 | 是 | 和文本处理一样，不需要修改 |
| LZO | 是 | LZO | .lzo | 是 | 需要建索引，还需要指定输入格式 |
| Snappy | 是 | Snappy | .snappy | 否 | 和文本处理一样，不需要修改 |
压缩文件是否可切分决定了其是否可应用于Map阶段的压缩。
1.2.2 压缩性能比较
| 压缩算法 | 原始文件大小 | 压缩文件大小 | 压缩速度 | 解压速度 |
|---|---|---|---|---|
| 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 |
常用LZO压缩方式进行压缩处理。
1.2.3 MR各阶段的压缩
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9lJTz9d1-1625191603089)(…/…/4-Hadoop总结资料/img/1597888789850.png)]
1.2.4 压缩参数配置
| 参数 | 默认值 | 阶段 | 建议 |
|---|---|---|---|
| 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 |
第2章 企业级调优
2.1 MR运行慢原因分析
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xJsMQyzb-1625191603090)(…/…/4-Hadoop总结资料/img/1597889198963.png)]
2.2 优化方法
2.2.1 数据输入
减少小文件输入，采用combineTestInputFormat来输入
2.2.2 Map阶段
**①减少溢写次数（spill）：**通过调整mapreduce.io.sort.mb及mapreduce.map.sort.spill.percent参数，增大出发spill内存上限，减少次数；
②减少合并次数（merge）:通过io.sort.factor 参数，增大Merge的文件数；
③Map之后不影响业务逻辑的情况下使用Combine。
2.2.3 Reduce阶段
①合理设置Map、Reduce数；
②设置Map、Reduce共存；
③规避使用Reduce；
④合理设置Reduce的Buffer
2.2.4 IO输入
①使用数据压缩的方式减少网络IO的时间；
②使用SequenceFile二进制文件；
2.2.5 调优参数详见课件
最后
以上就是疯狂小伙为你收集整理的Hadoop复习笔记一、Hadoop二、MapReduce&Yarn三、Hadoop优化与新特性的全部内容，希望文章能够帮你解决Hadoop复习笔记一、Hadoop二、MapReduce&Yarn三、Hadoop优化与新特性所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复