# Hadoop-demo
**Repository Path**: CandyPop/Hadoop-demo
## Basic Information
- **Project Name**: Hadoop-demo
- **Description**: hadoop使用说明
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 0
- **Forks**: 0
- **Created**: 2022-04-02
- **Last Updated**: 2022-08-24
## Categories & Tags
**Categories**: Uncategorized
**Tags**: hadoop
## README
#### Hadoop 3.x
##### 什么是大数据
顾名思义，指的是，数据的体量十分庞大。
大数据的体量一般在TB-EB级别
bit、Byte、KB、MB、GB、**TB、PB、EB、**ZB、YB、BB、NB、DB
大数据技术主要解决，海量数据的的，**采集**，**存储**，**分析计算**问题。
用一个通俗的话来说，如果你有一份100T的资料，但是你只有一块1T的硬盘该如何存储，当你想要看着1T的资料中的某个月份的资料的时候，该如何获取。
##### 大数据特点(4V)
* Volume 大量
* 会产生大量的数据，级别至少在TB级别
* Velocity 高速
* 处理这些数据的速度，效率
* Variety 多样
* 不止能存储结构化的数据，也可以处理非结构化的数据
* Value 低价值密度
* 如此众多的数据，我们只关心其中有价值的数据，称为**提纯**
#### Hadoop 是什么
存储大数据的分布式框架。
* Hadoop1.x组成
* HDFS 数据存储
* MapReduce 计算+资源调度
* Common 辅助工具
Hadoop 中MapReduce中责任太繁重了，要负责计算和资源调度，所以在2.x进行了改善
* Hadoop2.x
* MapReduce 计算
* Yarn 资源调度
* HDFS 数据存储
* Common 辅助工具
* Hadoop3.x并没有明显变化，只有内部细节优化
##### HDFS 架构概述
Hadoop Distributed File System 简称HDFS 是一个分布式文件系统
有以下数据结构
* NameNode(nn)存储文件的元数据，如文件名，文件目录结构，文件属性(生成时间、副本数、文件权限)，以及每个文件的块列表，和快所在的DataNode等
* DataNode(dn)在本地文件系统存储文件块数据，以及块数据的校验和
* Seconday NameNode(2nn)每隔一段时间对NameNode元数据备份(副本机制)
做一个比喻就是，NameNode是一个文件目录，有指向文件的所在，DataNode是真正存储数据的地方，而2nn会对目录进行备份，保证分布式环境下的高可用。
##### YARN 架构概述
Yet Another Resource Negotiator 简称YARN，另一种资源协调者，是Hadoop的资源管理器
管理什么资源，主要是CPU，和内存。
* ResourceManager（RM）：整个集群资源(内存、CPU等)的老大
* NodeManager（NN）：单个节点服务器资源老大
* ApplicationMaster（AM）：单个任务运行的老大
* Container：容器，相当一台独立的服务器，里面封装了任务运行所需的资源，如内存、CPU、磁盘、网络等。
需要说明的是，一个appliationMaster负责运行任务，当他要运行任务的时候，会向ResourceManager申请资源来运行任务，RM将把任务分配到符合要求的NodeManager中的容器中开始运行任务，container可以理解成不同的小型虚拟机，一个container至少要分配一g内存，一核cpu，一个nodeManager以2c4g为例子，那么他最多只能创建2个container，当一个容器中的任务还有后续任务，或者所需的资源有所增加，可以由RM分配到其他地方去。
##### MapReduce 架构概述
MapReduce 将计算过程分为两个阶段，Map和Reduce
* Map 阶段并行处理输入数据
* Reduce 阶段对Map结果进行汇总
假设你需要在100t的数据中，找到ss1505_wuma.avi这个文件，由于100t是分布在不同的hadoop服务器上，所以，让不同的服务器自己去检索，然后将结果汇总到汇总服务器上。
第一步，Map阶段，也就是将任务分布到不同节点，然后去执行，最后Reduce阶段，然后将结果进行汇总。
##### HDFS、YARN、MapReduce 三者关系
HDFS作为存储海量数据的底层结构，用于存储错综复杂的数据，YARN将会分配资源去运行检索数据任务，具体的查找流程由MapReduce完成，通过分发任务，和汇总任务，将结果反映到目录上。然后2nn会定期进行备份。
当一个客户端发起一个查找资源的请求，首先YARN会在连接到某个NodeManager上，创建一个容器来运行这个请求的任务，这个ApplicationMaster会向RM申请资源来完成任务，RM收到请求后，会在不同的Hadoop服务器上按照申请资源要求创建容器来运行一个MapTask，这一步就是MapReduce的工作，每一台Hadoop服务器上都会运行这个MapTask用于检索自己的HDFS存储的文件，最后Reduce汇总到另一台服务器上，每个Reduce阶段都会返回结果，通知反写到对应的NameNode(目录)节点上。
##### 大数据技术生态体系
##### 准备环境
准备三台linux的虚拟机，且都装好了jdk环境。
安装epel-release
Extra Packages For Enterprise 是Linux为红帽系的操作系统提供的额外的软件包，适用于RHEL,CentOs和Scientific Linux，相当于一个软件仓库，大多数rpm包在官方repository中那是找到不到的
```
yum install -y epel-release
```
如果你是最小安装，需要安装vim和net-tool
```
yum install -y net-tools
yum install -y epel-release
```
接着关闭防火墙，一般服务器都会将防火墙关闭，但是如果要对外开放服务，你也需要在访问前设置防火墙，来保护你的服务器。
```
systemctl stop firewalld
systemctl disable firewalld.service
```
```
https://hadoop.apache.org/release/3.1.3.html
```
下载好后，配置环境变量
```
sudo vim /etc/profile.d/my_env.sh
```
```
#HADOOP_HOME
export HADOOP_HOME= 安装目录
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin
```
```
source /etc/profile
```
目录结构
bin、etc、sbin用的比较多，bin一些命令，etc存放一些配置文件，sbin启动服务的命令，bin
##### Hadoop的三种模式
* 本地模式
* 数据存储在linux本地
* 伪分布
* 数据存储在HDFS，分布在另一台hadoop服务器
* 完全分布式
* 拥有2n+1节点的服务器，分别存在hadoop服务器
##### 本地模式下运行一个官方小案例
在hadoop的目录下，创建一个wcinput的文件夹
```shell
[root@hadoop102 hadoop-3.1.3]# mkdir wcinput
[root@hadoop102 hadoop-3.1.3]# cd wcinput/
[root@hadoop102 wcinput]# ls
[root@hadoop102 wcinput]# ll
total 0
[root@hadoop102 wcinput]# vim word.txt
[root@hadoop102 wcinput]# ll
total 4
-rw-r--r--. 1 root root 35 Apr 3 07:37 word.txt
[root@hadoop102 wcinput]#
```
在word文件里写下这些内容。
```
ss ss
clas clas
banz
pop opop pop
```
我们希望能够统计这些单词出现的次数。
到hadoop的安装目录下，使用hadoop指令，运行`hadoop-mapreduce-examples-3.1.3.jar`的众多案例的wordcount案例，从wcinput目录下，输出到wcouput目录下，请注意，输出**路径一定要不存在**。
```sh
[root@hadoop102 hadoop-3.1.3]# hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar wordcount wcinput/ wcouput
2022-04-03 07:41:47,380 INFO impl.MetricsConfig: loaded properties from hadoop-metrics2.properties
2022-04-03 07:41:47,443 INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).
2022-04-03 07:41:47,443 INFO impl.MetricsSystemImpl: JobTracker metrics system started
#...
Bytes Written=44
[root@hadoop102 hadoop-3.1.3]# cd wcouput/
[root@hadoop102 wcouput]# ll
total 4
-rw-r--r--. 1 root root 32 Apr 3 07:41 part-r-00000
-rw-r--r--. 1 root root 0 Apr 3 07:41 _SUCCESS
[root@hadoop102 wcouput]# cat part-r-00000
banz 1
clas 2
opop 1
pop 2
ss 2
[root@hadoop102 wcouput]#
```
##### 不同服务器之间的拷贝
我们在上面创建了Hadoop102-104，那么服务器之间是否可以传输文件。答案是可以的
使用scp命令就可以传输文件
```
这是在103 执行另外两个服务器之间的文件传输
scp -r root@hadoop102:/usr/local/* root@hadoop104:/usr/local/*
执行完该命令后，需要输入两个服务器root账号的密码，全部完成后，开始传输
如果你希望直接在本机传输到其他服务器上
scp -r /usr/local/* root@hadoop104:/usr/local/*
```
##### 不同服务器之间的同步
上面的命令可以100%的拷贝，但是可能存在这样一个情况，如果只是某一个文件发生了改变，却还是要复制整个文件夹，其实有些不必要，而rsync可以只针对某个修改的文件进行同步，而不影响其他没有修改的文件，所以效率大大提升。
```
-a 归档考别
-v 显示复制过程
rsync -av /usr/local/* root@hadoop104:/usr/local/*
```
第一次同步，使用scp命令，第二次使用rsync命令
##### 集群分发脚本
但是每次同步文件都需要写命令，有点太过繁琐。
如果有一个命令可以直接让所有服务器都执行已经设定好的命令，就可以大大节省时间。
```
#!/bin/bash
#1. 判断参数个数
if [ $# -lt 1 ]
then
echo Not Enough Arguement!
exit;
fi
#2. 遍历集群所有机器
for host in hadoop102 hadoop103 hadoop104
do
echo ==================== $host ====================
#3. 遍历所有目录，挨个发送
for file in $@
do
#4. 判断文件是否存在
if [ -e $file ]
then
#5. 获取父目录
pdir=$(cd -P $(dirname $file); pwd)
#6. 获取当前文件的名称
fname=$(basename $file)
ssh $host "mkdir -p $pdir"
rsync -av $pdir/$fname $host:$pdir
else
echo $file does not exists!
fi
done
done
```
```
赋予脚本执行权限
chmod +777 xsync
测试脚本
xsync /home/atguigu/bin
将脚本复制到bin中，可以全局调用
sudo cp xsync /bin/
//同步到其他机器
sudo ./bin/xsync /etc/profile.d/my_env.sh
```
##### 免秘登陆
（2）生成公钥和私钥
[atguigu@hadoop102 .ssh]$ pwd
/home/atguigu/.ssh
[atguigu@hadoop102 .ssh]$ ssh-keygen -t rsa
然后敲（三个回车），就会生成两个文件id_rsa（私钥）、id_rsa.pub（公钥）
（3）将公钥拷贝到要免密登录的目标机器上
[atguigu@hadoop102 .ssh]$ ssh-copy-id hadoop102
[atguigu@hadoop102 .ssh]$ ssh-copy-id hadoop103
[atguigu@hadoop102 .ssh]$ ssh-copy-id hadoop104
##### 集群配置
* NameNode和SecondaryNameNode不要安装在同一台服务器
* ResourceManager也很消耗内存，不要和NameNode、SecondaryNameNode配置在同一台机器上。
| | hadoop102 | hadoop103 | hadoop104 |
| ---- | ------------------ | ---------------------------- | --------------------------- |
| HDFS | NameNode\|DataNode | DataNode | SecondaryName\|NodeDataNode |
| YARN | NodeManager | ResourceManager\|NodeManager | NodeManager |
修改如下配置
```xml
fs.defaultFS
hdfs://hadoop102:8020
hadoop.tmp.dir
/opt/module/hadoop-3.1.3/data
hadoop.http.staticuser.user
atguigu
dfs.namenode.http-address
hadoop102:9870
dfs.namenode.secondary.http-address
hadoop104:9868
yarn.nodemanager.aux-services
mapreduce_shuffle
yarn.resourcemanager.hostname
hadoop103
yarn.nodemanager.env-whitelist
JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_MAPRED_HOME
mapreduce.framework.name
yarn
```
配置集群节点
```
hadoop-3.1.3/etc/hadoop/workers
```
在该文件中增加如下内容：
hadoop102
hadoop103
hadoop104
不允许有空格
第一次启动需要初始化
**如果集群是第一次启动**，需要在**hadoop102**节点格式化NameNode（注意：格式化NameNode，会产生新的集群id，导致NameNode和DataNode的集群id不一致，集群找不到已往数据。如果集群在运行过程中报错，需要重新格式化NameNode的话，一定要先停止namenode和datanode进程，并且要删除所有机器的data和logs目录，然后再进行格式化。）,在安装目录下
```
hdfs namenode -format
```
开始启动，在102输入以下命令
```
sbin/start-dfs.sh
```
也许会存在找到一些配置的环境变量，这里统一配置下
```
export HDFS_NAMENODE_USER=root
export HDFS_DATANODE_USER=root
export HDFS_SECONDARYNAMENODE_USER=root
export YARN_RESOURCEMANAGER_USER=root
export YARN_NODEMANAGER_USER=root
export JAVA_HOME=/usr/local/java
```
看到三台服务器已经启动了，说明启动成功。完成后，我们可以访问web页面，端口就是之前我们配置好的端口
```
http://192.168.10.102:9870/
```
这个功能比较常用
接着我们启动yarn资源管理器，之前我们定好了需要在103上启动，**就一定不要再其他服务器上启动**。
切换到103
```
可以看到，我们之前约定好的一模一样。一个都不一样都不行
-
| | hadoop102 | hadoop103 | hadoop104 |
| ---- | ------------------ | ---------------------------- | --------------------------- |
| HDFS | NameNode\|DataNode | DataNode | SecondaryName\|NodeDataNode |
| YARN | NodeManager | ResourceManager\|NodeManager | NodeManager |
同样yarn也提供了web页面，可以展示现在运行的job
```
http://192.168.10.103:8088/cluster
```
##### 集群测试
尝试在hadoop创建一个文件夹
```
hadoop fs -mkdir /input
```
上传一个文件
```
hadoop fs -put word.txt /input
```
数据存储在哪里
我们之前说过，一份文件存在三个副本，我们可以看到103也同样存在相同的副本。
运行一个任务
这里需要注意的是，由于是集群环境，你的输入路径也应该是集群环境下的路径，而不是本地路径
```
hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar wordcount /input /ouput
/input 是我们定义好的路径 /output是输出路径
```
执行后，我们可以发现，ResourceManager也出现了任务
执行完毕后，结果也输出到了指定路径
但是我们发现，当点击历史记录的时候，是无法查看的，这是因为还没配置历史服务器。
##### 关闭服务器
```
先关闭yarn
切换到103服务器
sbin/stop-yarn.sh
然后切换到102
sbin/start-dfs.sh
```
如果发生故障，导致服务有问题，首先关闭掉所有的服务，然后清除掉每个节点的data和logs，然后再nameNode节点上初始化，再启动。
由于每个DataNode节点都是有自己的版本号的，也就是VERSION，当不匹配的时候，就无法正常启动。
##### 配置历史服务器
```xml
mapreduce.jobhistory.address
hadoop102:10020
mapreduce.jobhistory.webapp.address
hadoop102:19888
```
然后先暂停103的yarn服务，在重启一下
接着手动启动历史服务器，切换到102启动
```
```
##### 配置日志聚集
hdfs将会收集hadoop各个节点的日志，进行汇总展示，而我们就需要配置这个。
```xml
yarn.log-aggregation-enable
true
yarn.log.server.url
http://hadoop102:19888/jobhistory/logs
yarn.log-aggregation.retain-seconds
604800
```
修改完毕后，你需要重启历史服务器和yarn
```
mapred --daemon stop historyserver
sbin/stop-yarn.sh
start-yarn.sh
mapred --daemon start historyserver
```
各个模块分开启动/停止（配置ssh是前提）
常用
（1）整体启动/停止HDFS
start-dfs.sh/stop-dfs.sh
（2）整体启动/停止YARN
start-yarn.sh/stop-yarn.sh
2）各个服务组件逐一启动停止
（1）分别启动/停止HDFS组件
hdfs --daemon start/stop namenode/datanode/secondarynamenode
（2）启动/停止YARN
yarn --daemon start/stop resourcemanager/nodemanager
##### 两个面试题
* 常用端口号
* hadoop3.x
* HDFS NameNode 内部通用端口:8020/900/9820
* HDFS NameNode 对用户的查询端口：9870
* Yarn 查看任务运行情况：8088
* 历史服务器：19888/10020
* hadoop2.x
* HDFS NameNode 内部通用端口:8020/900
* HDFS NameNode 对用户的查询端口：50070
* Yarn 查看任务运行情况：8088
* 历史服务器：19888/10020
* 常用的配置文件
* 3.x
```
core-site.xml hdfs-site.xml yarn-site.xml mapred-site.xml workers
```
* 2.x
```
core-site.xml hdfs-site.xml yarn-site.xml mapred-site.xml slaves
```
##### 时间同步
在外网环境下，服务器会自动与外网时间同步，但是当是访问不到外网的内网环境下，由于各个服务器之间可能存在不同步，也许执行任务的时候，因为时间不对，可能有点问题。
##### HDFS
hadoop distribute file system ，分布式文件系统。
hdfs适合于，一次写入，多次读出的操作，也就是修改的场景的极少的，而且在大数据场景环境下，也不会要求你对数据进行修改。
###### HDFS的优缺点
* 优点
* 数据自动保存多个副本，他会增加副本的形式，提高容错
* 某一个副本丢失以后，他可以自动恢复
* 适合处理大数据
* 数据规模：能够处理数据规模达到GB,TB,甚至PB级别数据
* 文件规模：能够处理百万规模以上的文件数量，数量相当之大
* 可以构建在廉价的机器上，通过多副本机制，提高可靠性
* 缺点
* 不适合低延迟的数据访问，比如毫秒级的存储数据，无法做到。
* 无法高校对大量小文件进行存储
* 存储大量小文件的话，他会占用NameNode大量的内存来存储文件目录和块信息，这样是不可取的，因为NameNode的内存总是有限的
* 小文件存储的寻址时间会超过读取时间，他违反了HDFS的设计目标
* 不支持并发写入，文件的随机修改
* 一个文件只能有一个写，不允许多个线程同时写
* 只支持数据的append，不支持文件的随机修改
###### HDFS的架构
* NameNode（nn）：就是Master，管理者
* 管理HDFS的名称空间
* 元数据的命名空间的存储文件位置，会在这里记录
* 配置副本策略
* 可以分别配置，不同文件的副本策略，例如文件的个数据，会告诉DateNode的信息。
* 管理数据块（Block）映射信息
* Block是HDFS的最小存储单元，可以在hdfs-default.xml文件中配置大小，hadoop底层会将上传的文件切分成若干个block来存储，默认是128M，当你上传一个200M的文件时，这个文件在128M的设置下，会变成128M和72M，并且，当你的多集群的情况下，会分别存在不同的机器中，且128M与72M不一定存在同一个机器上，而这些信息被DataNode记录。他们的副本也是如此。
* 处理客户端的读写请求
* DataNode：就是Slave。NameNode 下达命令，DataNode执行实际的操作。
* 存储实际的数据块（Block）
* 执行数据块的读写操作
* Secondary NameNode：**并非NameNode的热备**。当NameNode挂掉的时候，它并不能替换NameNode并提供服务。
* 辅助NameNode，分担其工作量，比如定期合并Fsimage和Edits，并推送给NameNode
* 在紧急情况下，可辅助恢复NameNode。但无法完全恢复NameNode所持有的数据，只能还原snn自己持有的一部分。
* Client 客户端
* 文件切分。文件上传HDFS的时候，Client将文件切分成一个一个的Block
* 与NameNode交互，获取文件位置信息
* 与DataNode交互，读取或者写入数据
* Client提供一些命令来管理HDFS，比如NameNode格式化
###### HDFS文件块大小
```xml
dfs.blocksize
134217728
```
**为什么块的大小不能设置太小，也不能设置太大**
* HDFS的块设置太小，会增加寻址时间，程序会一直找块的开始时间，因为房间多，找人就比较麻烦。因为是线性的
* 如果设置太大，从数据传输数据时间，会明显大于定位这个快开始位置所需的时间，都只程序在处理这块数据时候，会非常慢
HDFS块的大小设置主要取决于**硬盘的传输速率**
###### HDFS的Shell操作
hadoop fs 具体命令 和 hdfs dfs 具体命令是一样的
```xml
查看命令
hadoop fs -help rm
查看rm的命令操作
```
创建文件夹
```
hadoop fs -mkdir /sanguo
```
**上传**
-moveFromLocal：从本地**剪切**粘贴到HDFS
```
hadoop fs -moveFromLocal ./shuguo.txt /sanguo
```
-copyFormLocal：从本地文件系统**拷贝**文件到HDFS路径去
```
hadoop fs -copyFromLocal weiguo.txt /sanguo
```
-put：等同于 copyFromLocal，生产环境更习惯用put
```
hadoop fs -put ./wuguo.txt /sanguo
```
-appendToFile：追加一个文件内容到已经存在文件末尾
```
hadoop fs -appendToFile liubei.txt /sanguo/shuguo.txt
```
**下载**
-copyToLocal ：从HDFS拷贝到本地
```
hadoop fs -copyToLocal /sanguo/shuguo.txt ./
```
-get：等同于copyToLocal，生产环境更习惯用get
```
hadoop fs -get /sanguo/shuguo.txt ./shuguo2.txt
```
**常用操作**
-ls 显示目录信息
```
hadoop -fs -ls /sanguo
```
-cat 查看文件内容
```
hadoop fs -cat /snaguo/shuguo.txt
```
-chgrp -chmod -chown:Linux文件系统中的用法一样，修改文件所属权限
```
hadoop fs -chmod 666 /sanguo/shuguo.txt
hadoop fs -chown root:root /sanguo/shuguo.txt
```
-mkdir 创建路径
```
hadoop fs -mkdir /jinguo
```
-cp 从HDFS的路径拷贝到HDFS的另一个路径
```
hadoop fs -cp /sanguo/shuguo.txt /jinguo
```
-mv 从HDFS 目录中移动文件，也就是剪切
```
hadoop fs -mv /sanguo/wuguo.txt /jinguo
```
-tail 显示一个文件末尾1kb的数据
```
hadoop fs -tail /jinguo/shuguo.txt
```
-rm删除文件或者文件夹，rm -r 递归删除目录及里面内容
```
hadoop fs rm /sanguo/shuguo.txt
```
-du 统计文件夹大小
```
hadoop fs -du -s -h /jinguo
输出：
27 81 /jinguo
27 表示文件的大小，81表示27*3个副本，/jinguo表示查看的目录
hadoop fs -du -h /jinguo
14 42 /jinguo/shuguo.txt
7 21 /jinguo/weiguo.txt
6 18 /jinguo/wuguo.txt
```
-setrep 设置HDFS中文件的副本数量
```
hadoop fs -setrep 10 /jinguo/shuguo.txt
```
这里设置的副本数，只是记录在NameNode的元数据中，是否真的会有这么多的副本，还需要看DataNode的数量，因为目前只有三台设备，最后也就是3个副本。只有节点数增加到10台时，副本数才能达到10
###### 使用HDFS API
首先配置Hadoop的环境变量
看源码
###### HDFS的写数据流程
* 当你想要使用客户端上传一个`ss.avi`，一个200m的文件
* 由于你使用的是分布式文件系统，所以创建的客户端也是`DistributedFileSystem`，创建完成后，你请求的上传地址是`/user/atguigu/ss.avi`，这个请求是发送给NameNode，也就是目录，他会检查权限，例如是否是这个节点的所有者，还有是否目录文件重复等。
* 当检查没有问题时会响应会客户端标识可以上传，由于所需上传的文件大于128M，所以会切分成足够128M大小的Block，这是个200M的文件，所以会切割成128M和72M两个Block，首先准备发送第一个128M，请求NameNode返回其管理的所有DataNode集群信息，包括这些节点的地理位置，机架等信息。
* 挑选好节点后，就可以尝试写出数据了。你需要准备`FSDataOutputStream`对象来写数据。传输数据的最小单位是packet，大小为64k，而一个packet中会若干个chunk，这些chunk由512byte+chunksum4byte组成，当数量达到packet大小就会输出到一个等待**发送的缓冲队列**中，缓冲队列中可能含有多个packet。同时客户端还会持有一个**ack回执队列**，此队列中会有刚刚发送出去的packet的备份，目的在于当发送packet写入DataNode节点失败后，可以快速从回执队列中拿到这个packet发起重试，如果发送成功后，会将此packet删除。
* 接着发送到第一个选择好的DataNode节点，建立好通信通道，接着DataNode1会自己找到下一个DataNode以此类推，形成一个串行通道。当第一个packet到达第一个DataNode的时候，是首先保存在内存中的，接着在第一个DataNode写入磁盘的过程中，内存中的数据会接着传递到下一个，基于DataNode2会再次开始写，这也是replication的副本机制。当一个packet完成所有节点的写入后，会一个接着一个返回回执ack，等所有的DataNode都回执并完成。
* packet所有的数据包完成写入后，就写入完成。
###### 节点距离计算
服务器放在不同的机架中，机架又放在不同的机房中，机房又放在不同的地区。
如果说看d1/r1/n1 单元的服务器与 d1/r2/n2 的节点距离，那就是先找到他们两个的共同祖先，很明显，对于n1来说，首先找到r1，然后找到d1，d1是他们的共同祖先，到这里是2步，然后再向下找。找到r2，然后找到n1，总共是4步，所以他们的节点距离是4