标签：HDFS node01 文件 -- 数据 Hadoop NameNode root 节点

Hadoop 3.1.2

算法复杂度

]孙发复杂度分为时间复杂度和空间复杂度
- 时间复杂度执行算需要计算工作量
- 而空间复杂度是指执行这个算法所需要的内存空间；
时间和空间都是计算机资源的重要体现，而算法的复杂性就是体现在运行该算法时的计算机所需的资源

空间复杂度

一个程序的空间复杂度是指运行完一个程序所需内存的大小。
利用程序的空间复杂度，可以对程序的运行所需要的内存多少有个预先估计。
程序执行时所需存储空间包括以下两部分
- 固定部分:主要包括指令空间（即代码空间）、数据空间（常量、简单变量）等所占的空间。
- 可变空间:这部分空间的主要包括动态分配的空间，以及递归栈所需的空间等。
如果判断一个年份是平年还是闰年？
- 方法1：使用年份计算公式进行计算
- 方法2：创建一个长度为4000的数组，以年份为下标，直接取值即可
- 如果计算的年份超过1万年？

时间复杂度

一个算法执行所耗费的时间，从理论上是不能算出来的，必须上机运行测试才能知道。但我们不可能也没有必要对每个算法都上机测试，只需知道哪个算法花费的时间多，哪个算法花费的时间少就可以了。并且一个算法花费的时间与算法中语句的执行次数成正比例，哪个算法中语句执行次数多，它花费时间就多。

一个算法中的语句执行次数称为语句频度或时间频度。记为T(n)

T(n)=n^2+3n+4
T(n)=4n^2+2n+1
他们执行的频度是不同的，但是时间复杂度？

为了描述时间频度变化引起的变化规律，引入时间复杂度概念。记为O(…)

时间复杂度计算规则只需要遵循如下守则

用常数1来取代运行时间中所有加法常数；
只要高阶项，不要低阶项；
不要高阶项系数；

时间频度不同，但时间复杂度可能相同。

T(n)=n^2+3n+4-->n2+3n-->n^2
T(n)=4n^2+2n+1-->4n2+2n-->4n^2-->n2
它们的频度不同但时间复杂度相同，都为O(n2)
T(n)=log2nn+2n+10-->log2nn+2n-->log2n*n-->logN * n

常见的时间复杂度场景
O(1)—常数阶
O(N)—线性阶
O(logN)—对数阶
O(n^2)—平方阶
O(nlogn)—线性对数阶
O(1)<O(logn)<O(n)<O(nlogn)<O(n2)<O(n3)<O(2^n)<O(n!)<O(nn)

时间复杂度去估算算法优劣的时候注重的是算法的潜力
- 有100条数据的时候算法a比算法b快
- 有10000条数据的时候算法b比算法a快
- 算法b的时间复杂对更优

时间与空间的取舍

日常操作中：我们都是更加注重时间复杂度，有些特殊情况下，空间复杂度可能会更加重要

因为很多时候空间复杂度可以花钱解决
获取明天现在这个时刻？
//请用代码获取明天这个时刻？
Thread.sleep(3600*24);
Date date = new Date();
以后时间复杂度的问题具体情况还要具体分析，两行代码执行一天
编程的精髓和美，并不在于一方的退让和妥协。而是在于如何在二者之间取一个平衡点，完成华丽
变身
但是将来工作中一切以时间复杂度为标准

十大排序算法

默认以升序排列(小-->大)

冒泡
- 从位置0开始，当前位置数字和后面位置的数字进行比较
- 如果前面的大于后面，数据交换
- 将位置向后移动一位，重复第一个过程，直到最后一个
- 重复刚才的过程
- 时间复杂度O=n^2

十大经典排序算法动画与解析，看我就够了！（配代码完全版）

选择
假设第一个数字就是最大数字，记录他的索引
然后和后面的数字进行比较，如果后面的数字大于最大数字，重新记录新数字的索引
直到最后一个数字，然后将最大数字索引与最后一个数字进行交换
时间复杂度 O=n^2 十大经典排序算法动画与解析，看我就够了！（配代码完全版）

插入
假设原始数列是有序的
每次向数列中插入一个数字，从右向左依次比较，找到合适的位置
数列重新变为有序数列
在向数列插入一个数字
时间复杂度 O=n^2 十大经典排序算法动画与解析，看我就够了！（配代码完全版）

希尔排序

算法步骤

选择一个增量序列 t1，t2，……，tk，其中 ti > tj, tk = 1；
按增量序列个数 k，对序列进行 k 趟排序；
每趟排序，根据对应的增量 ti，将待排序列分割成若干长度为 m 的子序列，分别对各子表进行直接插入排序。仅增量因子为 1 时，整个序列作为一个表来处理，表长度即为整个序列的长度。

归并排序

让两个有序队列进行比较，然后每个队列依次取出1个数字比较，然后小的数字被去除
分割
依次将数列二等分，二等分之后的子序列继续二等分
直到每个子序列只有一个数字，停止分割
合并
按照分割的顺序进行合并

十大经典排序算法动画与解析，看我就够了！（配代码完全版）

快速
三种（挖坑法左右法前后法）
操作：选择一个数字作为参照物（基准数字）5
思想：然后和另一个的进行比较，如果是右边的小于5就交换，5左面的大于5就交换
整体排序之后，5的位置是正确的，左右两个是无序
将左右两个都当做一个新序列进行排序基准数
时间复杂度O=n*log2n

十大经典排序算法动画与解析，看我就够了！（配代码完全版）

堆
一种基于树的排序算法
二分查找树
任意节点
最多有两个子节点
左面的节点都小于父节点
右面的节点都大于父节点

十大经典排序算法动画与解析，看我就够了！（配代码完全版）

计数
一种很巧妙的排序算法
比较特殊的排序算法，只针对特殊数据的排序
将对应的数据让数组的索引对应值+1
最后遍历数组，索引对应的值是几就打印几

十大经典排序算法动画与解析，看我就够了！（配代码完全版）

桶
将数字按照范围进行分割
然后对分割后的数据进行排序（继续分割）
将来数据分桶要特别注意数据倾斜的问题

基数
基数排序是按照地位先排序，然后手机，在按照高位排序，然后再收集，依次类推，直到最高
位

将所有待比较数值（正整数）统一为同样的数位长度，数位较短的数前面补零
从最低位开始，依次进行一次排序
从最低位排序一直到最高位排序完成以后, 数列就变成一个有序序列

大数据思维

查重

有1T数据，这些数据中心有两行数据是重复的，如果找出重复行

方案1
从硬盘读取数据，按照首字母放回硬盘，首字母相同的肯定在同一个的区域
依次类推，如果到最后一个文件中，有两个相同的字符串，那么就是相同的字符串
优点：
- 分拆开
缺点：
- 长度不等的时候，有可能中途的时候已经有相同字符串了，但是不能找出
方案2
- 读取数据按照长度分桶
- 然后空桶或者只有一个数据的桶就排除掉
- 缺点：
  - 如果长度相差不大，桶数量会急剧减少，导致每个桶数量过多

方案3

我们一定要把相同的两条记录放入到同一个桶中
需要选用一种相同均匀的分拆数据的方式，要求数据散列的非常完美：
采用Hash算法对每一行数据计算Hash值
相同的数据Hash值一定相同，但是Hash值相同字符串不一定相同
根据文件的大小，计算文件的数量，然后让Hash对数量取余
将余数相同的数据存放到同一个文件中
查重的时候只需要将一个文件读到内存中即可

排序

如何用1G内存的电脑对1T的数据进行排序

方案1

设置一个阈值（500M），按行读取数据,如果读取数据量超过500M就产生一个新文件
大略算一下一共切分成2048个文件
现在的情况是：
- 文件内部和文件外部全部无序
使用归并算法对每个文件进行排序,让文件内部有序
然后使用归并让多个文件合并到一起

方案2

按照字母组合建立桶，桶的序号就是前2个字母的组合（52*52）
然后将数据依次放入到桶中
- 桶与桶之间有序，桶内无序
我们只需要将桶内的数据排序好，然后将桶链接到一起数据有序

Hadoop的历史

Hadoop进化史

Nutch
Hadoop最早起源于Nutch。
Nutch的设计目标是构建一个大型的全网搜索引擎，包括网页抓取、索引、查询等功能
但随着抓取网页数量的增加，遇到了严重的可扩展性问题——如何解决数十亿网页的存储和索
引问题。
Google
2003年、2004年谷歌发表的两篇论文为该问题提供了可行的解决方案。
GFS : Google File System
MapReduce ：数据计算的方法
Doug cutting 花费了自己的两年业余时间，将论文实现了出来
2008年1月，HADOOP成为Apache顶级项目

Hadoop Common 基础型功能
Hadoop Distributed File System 负责存放数据
Hadoop YARN 负责资源的调配
Hadoop MapReduce 大数据的计算框架
Hadoop Ozone 数据存放到仓库
Hadoop Submarine 机器学习引擎

分布式文件系统架构

FS File System

文件系统是基于硬盘之上的一个文件管理的工具
我们用户操作文件系统可以和硬盘进行解耦

DFS Distributed File System:

分布式文件系统
将我们的数据存放在多台电脑上存储
分布式文件系统有很多,
HDFS是mapreduce计算的基础

文件切分思想

文件存放在一个磁盘上效率肯定是低的

读取效率低
如果文件特别大会超出单机的存储范围

字节数组文件在磁盘真实存储文件的抽象概念
数组可以进行拆分和组装，源文件不会受到影响
切分数据

对字节数组进行切分

拼接数据

按照数组的偏移量将数据连接到一起,将字节数组链接到一起

偏移量
当前数据在数组中的相对位置，你可以理解为下标
数组都有对应的索引(下标),可以快速的定位数据
数据存储的原理：

不管文件的的大小，所有的文件都是由字节数组构成
如果我们要切分文件,就是将一个字节数组分成多份
我们将切分后的数据拼接到一起，数据可以继续使用
我们需要根据数据的偏移量将他们重新拼接到一起

Block拆分标准

拆分的数据块需要等大

数据计算的时候简化问题的复杂度
进行分布式算法设计的时候，数据不统一，算法很难设计
数据拉取的时候时间相对一致
通过偏移量就知道这个块的位置
相同文件，分成的数据块大小应该相等

数据块 Block

数据被切分后的一个整体称之为块
在H1默认大小为64M,在H2及其以后默认大小为128M
同一个文件中，每个数据块大小要一致除了最后一个节点外
不同文件中，块的大小可以不一致
文件大小不同可以设置不同的块的数量
真实情况下，会根据文件大小和集群节点的数量综合考虑块的大小
数据块的个数 =Ceil（文件大小 / 每个块的大小）

注意事项
HDFS中一旦文件被存储，数据不允许被修改

修改会影响偏移量
修改会导致数据倾斜
修改数据会导致蝴蝶效益

但是可以被追加，但是不推荐

追加设置需要手动打开
一般HDFS存储的都是历史数据。所以将来Hadoop的mr都用来进行离线数据的处理
块的大小一旦文件上传之后就不允许被修改
128m -512M

如果数据文件的切割点128M整好是一个单词的中间部分，切分数据如何保证数据的完整性？

Block数据安全

肯定要对存储数据做备份
备份的数据肯定不能存放在一个节点上
- 使用数据的时候可以就近获取数据
所以备份的数量要小于等于节点的数量
每个数据块会有3个副本，相同副本是不会存放在同一个节点上
副本的数量可以变更
- 可能近期的数据被分析的可能性跟大，副本数可以多设置几个
- 后期数据很少被分析，可以减少副本数

Block的管理效率

需要专门给节点进行分工

存储 DataNode

记录 NameNode

日志 secondaryNameNode

HDFS的特点

优点

高容错性
- 保存多个副本,且提供容错机制
- 副本丢失宕机自动回复,默认保存三份
运行在廉价的机器上
- 通过副本提高可靠性
- 提供容错回复机制
适合批处理
- 移动计算而非数据
- 数据位置暴露给计算框架,NameNode上游位置
适合大数据处理
- i.TB，甚至PB级数据
  ii.百万规模以上的文件数量
  iii.10K+节点规模
流式数据访问
- 一次写入,多次读取,高吞吐量,可以同时处理大量数据
缺点
- a)不擅长低延迟数据访问
  - 比如毫秒级
- b)不擅长小文件的分区
  - i.占用NameNode大量内存
  - ii.磁盘寻道时间超过读取时间
- c)不擅长并发写入，文件随机修改
  - i.一个文件只能有一个写入者
  - ii.仅支持append,也就是添加（有组件实现删等）

Hadoop3完全分布式搭建

准备安装环境

[root@node01 ~]# tar -zxvf hadoop-3.1.2.tar.gz
[root@node01 ~]# mv hadoop-3.1.2 /opt/lzj/
[root@node01 ~]# cd /opt/lzj/hadoop-3.1.2/etc/hadoop/

.修改集群环境

[root@node01 hadoop]# vim hadoop-env.sh

##直接在文件的最后添加
export JAVA_HOME=/usr/java/jdk1.8.0_231-amd64
export HDFS_NAMENODE_USER=root
export HDFS_DATANODE_USER=root
export HDFS_SECONDARYNAMENODE_USER=root

修改配置文件

[root@node01 hadoop]# vim core-site.xml

<property>
<name>fs.defaultFS</name>
<value>hdfs://node01:9000</value>
</property>
<property>
<name>hadoop.tmp.dir</name>
<value>/var/lzj/hadoop/full</value>
</property>

[root@node01 hadoop]# vim hdfs-site.xml

<property>
<name>dfs.namenode.secondary.http-address</name>
<value>node02:50090</value>
</property>
<property>
<name>dfs.namenode.secondary.https-address</name>
<value>node02:50091</value>
</property>
<property>
<name>dfs.replication</name>
<value>2</value>
</property>

[root@node01 hadoop]# vim workers

node01
node02
node03

拷贝分发软件

将配置好的软件分发到其他主机

[root@node01 ~]# cd /opt/lzj/
[root@node01 lzj]# scp -r hadoop-3.1.2 root@node02:`pwd`
[root@node01 lzj]# scp -r hadoop-3.1.2 root@node03:`pwd`

修改环境变量

[root@node01 hadoop]# vim /etc/profile

export HADOOP_HOME=/opt/lzj/hadoop-3.1.2
export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH

将环境变量拷贝到其他主机

[root@node01 lzj]# scp /etc/profile root@node02:/etc/profile
[root@node01 lzj]# scp /etc/profile root@node03:/etc/profile

重新加载三台服务器的环境变量

【123】# source /etc/profile

格式化NameNode

[root@node01 lzj]# hdfs namenode -format
[root@node01 lzj]# start-dfs.sh

Starting namenodes on [node01]
Last login: Fri Oct 30 21:32:11 CST 2020 from 192.168.58.1 on pts/0
node01: Warning: Permanently added 'node01,192.168.58.101' (ECDSA) to
the list of known hosts.
Starting datanodes
Last login: Fri Oct 30 22:06:12 CST 2020 on pts/0
node03: Warning: Permanently added 'node03,192.168.58.103' (ECDSA) to
the list of known hosts.
node02: Warning: Permanently added 'node02,192.168.58.102' (ECDSA) to
the list of known hosts.
node01: Warning: Permanently added 'node01,192.168.58.101' (ECDSA) to
the list of known hosts.
node03: WARNING: /opt/lzj/hadoop-3.1.2/logs does not exist. Creating.
node02: WARNING: /opt/lzj/hadoop-3.1.2/logs does not exist. Creating.
Starting secondary namenodes [node02]
Last login: Fri Oct 30 22:06:14 CST 2020 on pts/0
node02: Warning: Permanently added 'node02,192.168.58.102' (ECDSA) to
the list of known hosts.

测试集群

http://node01:9870
[root@node01 ~]# hdfs dfs -mkdir -p /lzj
[root@node01 ~]# hdfs dfs -put zookeeper-3.4.5.tar.gz /lzj/
[root@node01 ~]# hdfs dfs -D dfs.blocksize=1048576 -put zookeeper-3.4.5.tar.gz /lzj/

关闭集群

[root@node01 ~]# stop-dfs.sh

Stopping namenodes on [node01]
Last login: Fri Oct 30 22:06:20 CST 2020 on pts/0
node01: Warning: Permanently added 'node01,192.168.58.101' (ECDSA) to
the list of known hosts.
Stopping datanodes
Last login: Fri Oct 30 22:16:34 CST 2020 on pts/0
node03: Warning: Permanently added 'node03,192.168.58.103' (ECDSA) to
the list of known hosts.
node02: Warning: Permanently added 'node02,192.168.58.102' (ECDSA) to
the list of known hosts.
node01: Warning: Permanently added 'node01,192.168.58.101' (ECDSA) to
the list of known hosts.
Stopping secondary namenodes [node02]
Last login: Fri Oct 30 22:16:35 CST 2020 on pts/0
node02: Warning: Permanently added 'node02,192.168.58.102' (ECDSA) to
the list of known hosts.

关机拍摄快照

[123]# shutdown -h now

文件的数据类型

start命令

元数据信息—描述文件睡醒

文件vim命令

查看文件数据信息

分类

元数据

File 文件名
Size 文件大小（字节）
Blocks 文件使用的数据块总数
IO Block 数据块的大小
regular file：文件类型（常规文件）
Device 设备编号
Inode 文件所在的Inode
Links 硬链接次数
Access 权限
Uid 属主id/用户
Gid 属组id/组名
Access Time：简写为atime，表示文件的访问时间。当文件内容被访问时，更新这个时间
Modify Time：简写为mtime，表示文件内容的修改时间，当文件的数据内容被修改时，更新这个
时间。
Change Time：简写为ctime，表示文件的状态时间，当文件的状态被修改时，更新这个时间，例
如文件的链接数，大小，权限，Blocks数。

文件数据

真实存在文件中的数据

NameNode(NN)

功能接收客户端读写服务

NameNode存放文件与Block映射关系
DataNode存放Block与DataNode映射关系

保存文件元数据信息

文件归属
文件权限
文件大小时间
Block信息，但是block的位置信息不会持久化,需要每次开启集群的时候DN上报

收集Block的信息

系统启动时
NN关机的时候是不会存储任意的Block与DN的映射信息
DN启动的时候，会将自己节点上存储的Block信息汇报给NN
NN接受请求之后重新生成映射关系
- Block--DN3
如果某个数据块的副本数小于设置数，那么NN会将这个副本拷贝到其他节点
集群运行中
- NN与DN保持心跳机制,三秒钟发送一次

<property>
<description>Determines datanode heartbeat interval in
seconds.</description>
<name>dfs.heartbeat.interval</name>
<value>3</value>
</property>
<property>
<name>heartbeat.recheck.interval</name>
<value>300000</value>
</property>

如果客户端需要读取或者上传数据时候,NN可以知道DN健康状况

可以让客户端读取存活的DN节点

如果DN超过三秒没有心跳就认为DN出现异常

不会让新的数据读写到DataNode
- 客户访问的时候不提供异常结点的地址

如果DN超过10分钟+30秒没有心跳，那么NN会将当前DN存储的数据转存到其他节点

超时时长的计算公式为：
timeout = 2 * heartbeat.recheck.interval + 10 * dfs.heartbeat.interval。
而默认的heartbeat.recheck.interval 大小为5分钟，dfs.heartbeat.interval默认为3秒

性能

NameNode为了效率，将所有的操作都在内存中完成

NameNode不会和磁盘进行任何的数据交换
问题:
数据的持久化
数据保存在内存中，掉电易失

DataNode(DN)

功能

存放文件数据信息验证文件完整性校验信息

数据会存放到硬盘上
1m=1条元数据 1G=1条元数据
NameNode非常排斥存储小文件，一般小文件在存储之前需要进行压缩
汇报
- 启动
  - 汇报之前先验证Block文件是否被破坏
  - 向NN汇报当前DN上block信息
- 运行中
  - 向NN保持心跳机制
  - 客户可以向DN读写数据
当客户读写数据时候,首先去NN查询file与block与dn映射
- 然后客户端直接与dn建立连接诶,然后写数据

SecondaryNameNode

传统

日志机制

做任何操作之前先记录日志
当NN下次启动的时候，只需要重新按照以前的日志“重做”一遍即可
- 缺点
- edits文件大小不可控，随着时间的发展，集群启动的时间会越来越长
- 有可能日志中存在大量的无效日志
优点
- 绝对不会丢失数据

拍摄快照

可以将内存数据写到硬盘上
- 序列化
启动还可以将硬盘数据写到内存中
- 反序列化
缺点
- 关机时间过长
- 如果是异常关机，数据还在内存中，没法写入到硬盘
  如果写出频率过高，导致内存使用效率低（stop the world） JVM
优点
- 启动时间短

SNN解决方案

解决思路

让日志大小可控

定时快照保存

NameNode文件目录

查看目录

/var/lzj/hadoop/local/dfs/name/current
edits_0000000000000000001-0000000000000000019
edits_inprogress_0000000000000000020
当前已经正在执行的操作的日志信息
这些日志信息还没有被合并到镜像中
fsimage_0000000000000000000
fsimage_0000000000000000000.md5
fsimage_0000000000000000019
fsimage_0000000000000000019.md5
完整性校验规则
seen_txid -->19
VERSION

解决方案
当我们启动一个集群的时候，会产生四个文件

edits_0000000000000000001
fsimage_00000000000000000
seen_txid
VERSION

我们每次操作都会记录日志 -->edits_inprogress-000000001

随和时间的推移，日志文件会越来越大，当达到阈值的时候（64M 或 3600秒）

dfs.namenode.checkpoint.period 每隔多久做一次checkpoint ，默认3600s
dfs.namenode.checkpoint.txns 每隔多少操作次数做一次checkpoint，默
认1000000次
dfs.namenode.checkpoint.check.period 每个多久检查一次操作次数，默认60s

会生成新的日志文件
edits_inprogress-000000001 -->edits_0000001
创建新的日志文件edits_inprogress-0000000016

SNN数据恢复

强行杀死NameNode节点

kill -9 7388
清空namenode下name中的fsimage和edtis文件

[root@node01 ~]# rm -rf /var/lzj/hadoop/full/dfs/name/current/*

[root@node01 ~]# rm -rf /var/lzj/hadoop/full/dfs/name/current/*

secondary namenode下的name中的fsimage和edits复制到namenode对应文件夹中

[root@node01 ~]# scp -r
root@node02:/var/lzj/hadoop/full/dfs/namesecondary/current/*
/var/lzj/hadoop/full/dfs/name/current

启动namenode

1. hadoop-daemon.sh start namenode

访问namenode节点页面，成功
很不幸部分数据丢失

安全模式

集群启动时的一个状态

安全模式HDFS一种工作状态,处于安全模式状态下,只向客户端提供文件或只读视图,不接受对命名空间修改,同时NameNode节点也不会进行数据块的复制或删除
NameNode启动时
- 首先将镜像文件（fsimage）载入内存，并执行编辑日志（edits）中的各项操作。
  一旦在内存中成功建立文件系统元数据的映像，则创建一个新的fsimage文件和一个空的编辑日志。
  NameNode开始监听RPC和Http请求。
  此时NameNode处于安全模式，只接受客户端的读请求

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

安全模式下

安全模式下，各个DataNode会向NameNode发送自身的数据块列表
当NameNode有足够的数据块信息后，便在30秒后退出安全模式
NameNode发现数据节点过少会启动数据块复制过程

如果NN收集的Block信息没有达到最少副本数，就会将缺失的副本,从有的DN上拷贝到其他DN

dfs.replication.min=2
但是默认最低副本数为1
在拷贝的过程中系统还是处于安全模式

安全模式相关命令

hadoop dfsadmin -safemode leave 强制NameNode退出安全模式
hadoop dfsadmin -safemode enter 进入安全模式
hadoop dfsadmin -safemode get 查看安全模式状态
hadoop dfsadmin -safemode wait 等待一直到安全模式结束

HDFS的权限

HDFS对权限的控制
只能防止好人做错事
不能防止坏人做坏事
你告诉他你是谁，他就认为你是谁！

机架感知策略

distance(/D1/R1/H1,/D1/R1/H1)=0 相同的datanode
distance(/D1/R1/H1,/D1/R1/H3)=2 同一rack下的不同datanode
distance(/D1/R1/H1,/D1/R2/H4)=4 同一IDC下的不同datanode
distance(/D1/R1/H1,/D2/R3/H7)=6 不同IDC下的datanode

机架感知是为了保证副本在集群的安全性

我们需要将副本放在不同DN节点,节点也需要一定考量

可靠性,可用性,带宽消耗
第一个节点
- 集群内部(优先考虑和客户端相同节点作为第一个节点
- 集群外部(选择资源丰富且不繁忙的节点作为第一个接地那)

第二个节点

选择和第一个节点不同的机架其他节点

第三个节点

与第二个节点相同机架其他节点

第N节点,与前面节点不重复其他节点

HDFS写数据流程

写数据就是将客户端的数据上传到HDFS

宏观流程

1. 客户端向HDFS发送写数据请求
hdfs dfs -put tomcat.tar.gz /lzj/
2. filesystem通过rpc调用namenode的create方法
nn首先检查是否有足够的空间权限等条件创建这个文件,或者这个路径是否已经存在
有：NN会针对这个文件创建一个空的Entry对象,并返回成功状态给DFS
没有：直接抛出对应的异常，给予客户端错误提示信息
3. DFS如果接收到成功状态，会创建一个对象 FSDataOutputStream的对象给客户端使用
4. 客户端需要向NN询问第一个Block存放的位置
NN通过机架感知策略 (node1 node 2 node8)
5. 需要将客户端和DN节点创建连接

pipeline(管道)
客户端和node1创建连接 socket
node1和node2创建连接 socket
node2 和Node8创建连接 socket
6. 客户端将文件按照块block切分数据,但是按照packet发送数据
默认一个packet大小为64K,Block128M为2048个packet
7. 客户端通过pipeline管道开始使用FSDataOutputStream对象将数据输出
客户端首先将一个packet发送给node1,同时给予node1一个ack状态
node1接受数据后会将数据继续传递给node2,同时给予node2一个ack状态
node2接受数据后会将数据继续传递给node8,同时给予node8一个ack状态
node8将这个packet接受完成后，会响应这个ack给node2为true
node2会响应给node1 ,同理node1响应给客户端
8. 客户端接收到成功的状态,就认为某个packet发送成功了，直到当前块所有的packet都发送完成
9. 如果客户端接收到最后一个pakcet的成功状态,说明当前block传输完成
10. 客户端会将这个消息传递给NN，NN确认传输完成
NN会将block的信息记录到Entry,客户端会继续向NN询问第二个块的存储位置,依次类推
block1 (node1 node2 node8)
block2 (node1 node8 node9)
....
blockn(node1 node7 node9)
12. 当最后一个Block传输完成，管道就会被撤销
13. 当所有的block传输完成后，NN在Entry中存储所有的File与Block与DN的映射关系
14. 关闭FsDataOutPutStream

微观流程

首先客户端从自己的硬盘以流的方式读取数据文件到自己的缓存中
然后将缓存中的数据以chunk（512B）和checksum(4B)的方式放入到packet（64K）
chunk:checksum=128:1
checksum:在数据处理和数据通信领域中，用于校验目的的一组数据项的和
Packet分为两类，一类是实际数据包，另一类是heater包。
一个Packet数据包的组成结构

当packet满的时候加入到添加到 dataqueue ,
datastreamer开始从dataqueue队列上取出一个packet，通过FSDataOPS发送到Pipleline
在取出的时候，也会将packet加入到ackQueue,典型的生产者消费者模式
客户端发送一个Packet数据包以后开始接收ack，会有一个用来接收ack的ResponseProcessor进
程，如果收到成功的ack
如果某一个packet的ack为true,那么就从ackqueue删除掉这个packet
如果某一个packet的ack为false,将ackqueue中所有的packet重新挂载到发送队列,重新发送
最终DFS保存的数据格式为

HDFS读数据流程

首先客户端发送请求到DFS，申请读取某一个文件

/lzj/tomcat.tar.gz

DFS去NN查找这个文件信息
如果文件不存在,抛出指定错误
如果文件存在,返回成功状态

DFS创建FSDataInputStream对象,客户端通过这个对象读取数据

客户端获取文件第一个Block信息,返回DN1DN2 DN8

客户端直接就近原则选择DN1对应的数据

依次类推读取其他块信息,直到最后一个块,将Block合并成一个文件

关闭FSDataInputStream

Hadoop1的困境

单点故障

每个群集只有一个NameNode,NameNode存在单点故障SPOF

如果该计算机或进程不可用，则整个群集在整个NameNode重新启动或在另一台计算机上启
动之前将不可用
如果发生意外事件（例如机器崩溃），则在操作员重新启动NameNode之前，群集将不可
用。
计划内的维护事件，例如NameNode计算机上的软件或硬件升级，将导致群集停机时间的延
长。

水平扩展

将来服务器启动时候,启动速度慢

namenode随着业务的增多,内存占用也越来越多

如果namenode内存占满，将无法继续提供服务

业务隔离型差

存储:有课能需要存储不同部门的数据

计算:有可能需要不同业务计算流程

项目后期namenode吞吐量将会是集群的瓶颈

客户端所有的请求都会先访问NameNode

Hadoop2.x
NameNode节点的高可用
HA--high availability
NameNode业余的水平扩展
Federation

设计思想

hadoop2.x启用了主备节点切换模式（1主1备）

当主节点出现异常时候,集群直接将备用节点切换成主节点

要求备用节点马上就要工作

主备节点内存几乎同步

有独立的线程对主备节点进行健康监控状态

需要有一定选举机制,帮助我们确定主从关系

我们需要时间存储日志中间件

ANN

Active NameNode 的功能和原理的NN的功能是一样的

接收客户端请求,查询数据块DN信息

存储数据元数据信息

数据文件:Block:DN映射关系
工作:启动时:接收DN的block汇报
运行时:和DN保持心跳

存储介质

完全基于内存

优点:数据处理效率高

缺点:数据持久化(日志edits+快照fsimage)

SNN

Standby NameNode：NN的备用节点

他和主节点做同样的工作,但是它不会发出任何指令

存储:元数据信息

数据文件:Block：DN的映射关系
它的内存数据和主节点内存数据几乎是一致的

工作：
启动时：
接受DN的block汇报
运行时：
和DN保持心跳(3s,10m30s)

存储介质
完全基于内存
优点：数据处理效率高
缺点：数据的持久化

合并日志文件和镜像
当搭建好集群的时候，格式化主备节点的时候，ANN和SNN都会会默认创建
fsimage_000000000000000
当我们操作HDFS的时候ANN会产生日志信息
edits_inprogress_0000000000001
主节点会将日志文件中新增的数据同步到JournalNode集群上
所以只需要snn有操作的日志信息，就可以合并fsImage与edits信息,理论上是一直在合并数据
fsimage -->初始化创建
edits-->从JournalNode集群上定时同步
只要同步到edits文件，就开始于fsimage合并
当达到阈值的时候，直接拍摄快照即可
SNN将合并好的Fsimage发送给ANN,ANN验证无误后，存放到自己的目录中

DataNode(DN)

存储
- 文件的Block数据
介质
- 硬盘
启动时：
- 同时向两个NN汇报Block信息
运行中
- 同时和两个NN节点保持心跳机制

QJM

Quorum JournalNode Manager 共享存储系统，NameNode通过共享存储系统实现日志数据同
步。
JournalNode是一个独立的小集群，它的实现原理和Zookeeper的一致( Paxos)
ANN产生日志文件的时候，就会同时发送到 JournalNode的集群中每个节点上
JournalNode不要求所有的jn节点都接收到日志，只要有半数以上的（n/2+1）节点接受收到日
志，那么本条日志就生效
SNN每间隔一段时间就去QJM上面取回最新的日志
SNN上的日志有可能不是最新的
HA集群的状态正确至关重要，一次只能有一个NameNode处于活动状态。
JournalNode只允许单个NameNode成为作者。在故障转移期间，将变为活动状态的NameNode
将承担写入JournalNodes的角色，这将有效地防止另一个NameNode继续处于活动状态，从而使
新的Active节点可以安全地进行故障转移。

ZKFC

Failover Controller(故障转移控制器)

对 NameNode 的主备切换进行总体控制，能及时检测到 NameNode 的健康状况
在主 NameNode 故障时借助 Zookeeper 实现自动的主备选举和切换
为了防止因为NN的GC失败导致心跳受影响，ZKFC作为一个deamon进程从NN分离出来

启动时

当集群启动时,主备节点概念是模糊的
当zkfc检查到一个节点是健康状态,直接将其设置为主节点
当zkfc检查到两个NN节点是健康状态,发起投票机制
选出一个主节点,一个备用节点,并修改主备节点状态

运行

由 ZKFailoverController、HealthMonitor 和 ActiveStandbyElector 这 3 个组件来协同实现
主备切换

ZKFailoverController启动的时候会创建 HealthMonitor 和 ActiveStandbyElector 这两
个主要的内部组件
HealthMonitor 主要负责检测 NameNode 的健康状态
ActiveStandbyElector 主要负责完成自动的主备选举，内部封装了 Zookeeper 的处理逻
辑

主备节点正常切换

NameNode 在选举成功后，ActiveStandbyElector会在 zk 上创建一个
ActiveStandbyElectorLock 临时节点，而没有选举成功的备 NameNode 中的
ActiveStandbyElector会监控这个节点

如果 Active NameNode 对应的 HealthMonitor 检测到 NameNode 的状态异常时，
ZKFailoverController 会主动删除当前在 Zookeeper 上建立的临时节点
ActiveStandbyElectorLock
如果是 Active 状态的 NameNode 所在的机器整个宕掉的话，那么跟zookeeper连接的客户
端线程也挂了,会话结束,那么根据 Zookeepe的临时节点特性，ActiveStandbyElectorLock 节
点会自动被删除，从而也会自动进行一次主备切换

处于 Standby 状态的 NameNode 的 ActiveStandbyElector 注册的监听器就会收到这个节点
的 NodeDeleted 事件，并创建 ActiveStandbyElectorLock 临时节点，本来处于 Standby 状
态的 NameNode 就选举为Active NameNode 并随后开始切换为 Active 状态。

Zookeeper

为主备切换控制器提供主备选举支持。
辅助投票
和ZKFC保持心跳机制，确定ZKFC的存活

脑裂brain-split

定义
脑裂是Hadoop2.X版本后出现的全新问题，实际运行过程中很有可能出现两个namenode同
时服务于整个集群的情况，这种情况称之为脑裂。

原因
脑裂通常发生在主从namenode切换时，由于ActiveNameNode的网络延迟、设备故障等问
题，另一个NameNode会认为活跃的NameNode成为失效状态，此时StandbyNameNode会
转换成活跃状态，此时集群中将会出现两个活跃的namenode。因此，可能出现的因素有网
络延迟、心跳故障、设备故障等。

脑裂场景
NameNode 可能会出现这种情况，NameNode 在垃圾回收（GC）时，可能会在长时间内整
个系统无响应
zkfc客户端也就无法向 zk 写入心跳信息，这样的话可能会导致临时节点掉线，备 NameNode
会切换到 Active 状态
这种情况可能会导致整个集群会有同时有两个Active NameNode

脑裂问题解决方案是隔离

1.第三方共享存储:任意时刻,只有一个NN可以写入

2.DataNode：需要保证只有一个 NN 发出与管理数据副本有关的命令；

Client需要保证同一时刻只有一个NN 能够对 Client 的请求发出正确的响应。

每个NN改变状态时候,向DN发送自己的状态和一个序列号
DN在运行过程中维护此序列号,DN接收到这个范湖是认为该NN势新的active
如果这个时候原来的active恢复,返回给DN心跳信息包含active状态和原来的序列号,这个DN就会拒绝这个NN命令

解决方案

ActiveStandbyElector为了实现 fencing，当NN成为ANN之后创建Zookeeper临时节点
ActiveStandbyElectorLock，创建ActiveBreadCrumb 的持久节点，这个节点里面保存了这个
Active NameNode的地址信息(node-01)
Active NameNode的 ActiveStandbyElector在正常的状态下关闭 Zookeeper Session 的时
候，会一起删除这个持久节点

但如果 ActiveStandbyElector在异常的状态下关闭，那么由于 /hadoopha/${dfs.nameservices}/ActiveBreadCrumb 是持久节点，会一直保留下来，后面当另一个NameNode 选主成功之后，会注意到上一个 Active NameNode 遗留下来的这个节点，从而会回调 ZKFailoverController的方法对旧的 Active NameNode 进行 fencing。

首先尝试调用这个旧 Active NameNode 的 HAServiceProtocol RPC 接口的
transitionToStandby 方法，看能不能把它转换为 Standby 状态；

如果 transitionToStandby 方法调用失败，那么就执行 Hadoop 配置文件之中预定义的
隔离措施。

sshfence：通过 SSH 登录到目标机器上，执行命令 fuser 将对应的进程杀死
shellfence：执行一个用户自定义的 shell 脚本来将对应的进程隔离

在成功地执行完成 fencing 之后，选主成功的 ActiveStandbyElector 才会回调
ZKFailoverController 的 becomeActive 方法将对应的 NameNode 转换为 Active 状态，开始
对外提供服务。

Hadoop-Federation

HDFS Federation就是使得HDFS支持多个命名空间，并且允许在HDFS中同时存在多个Name
Node。

单NN局限性

Namespace（命名空间）的限制

NameNode所能存储的对象（文件+块）数目受到NameNode所在JVM的heap size的限制。
50G的heap能够存储20亿（200million）个对象，这20亿个对象支持4000个DataNode，
12PB的存储
DataNode从4T增长到36T，集群的尺寸增长到8000个DataNode。存储的需求从12PB增长到
大于100PB。

性能的瓶颈
- 整个HDFS文件系统的吞吐量受限于单个Namenode的吞吐量
隔离问题
- HDFS上的一个实验程序就很有可能影响整个HDFS上运行的程序
集群的可用性
Namenode的宕机无疑会导致整个集群不可用。
Namespace和Block Management的紧密耦合
纵向扩展目前的Namenode不可行
- 将Namenode的Heap空间扩大到512GB启动花费的时间太长
- Namenode在Full GC时，如果发生错误将会导致整个集群宕机

Federation(联邦)

块池Block Pool

Block pool(块池)就是属于单个命名空间一组block管理区域
每一个datanode为所有的block pool存储
Datanode是一个物理概念，而block pool是一个重新将block划分的逻辑概念
一个Namenode失效不会影响其下的datanode为其他服务Namenode的服务
datanode与Namenode建立联系并开始会话后自动建立Block pool

块池Block Pool
- Block pool(块池)就是属于单个命名空间的一组block(块)管理区域
- 每一个datanode为所有的block pool存储
- Datanode是一个物理概念，而block pool是一个重新将block划分的逻辑概念
- 一个Namenode失效不会影响其下的datanode为其他Namenode的服务
- datanode与Namenode建立联系并开始会话后自动建立Block pool
Namespace Volume（命名空间卷）
- 一个Namespace和它的Block Pool合在一起称作Namespace Volume
- Namespace Volume是一个独立完整的管理单元。当一个Namenode/Namespace被删除，
- 与之相对应的Block Pool也也被删除。
- 通过多个namenode/namespace把元数据的存储和管理分散到多个节点中
- 降低单个节点数据压力,计算压力
namenode/namespace可以通过增加机器来进行水平扩展
- 可以让更多的节点参与到运算
- namespace命名空间，通过这种方式确定要处理数据的路径
- 我们可以通过namenode和namespace组合使用
所有的nn共享dn
- 但是每一个namespace会单独管理自己的块
- 会创建一个管理块的机制：blocks pool

搭建高可用集群

准备安装环境

[root@node01 ~]# tar -zxvf hadoop-3.1.2.tar.gz
[root@node01 ~]# mv hadoop-3.1.2 /opt/lzj/
[root@node01 ~]# cd /opt/lzj/hadoop-3.1.2/etc/hadoop/

修改集群环境

[root@node01 hadoop]# vim hadoop-env.sh

##直接在文件的最后添加
export JAVA_HOME=/usr/java/jdk1.8.0_231-amd64
export HDFS_NAMENODE_USER=root
export HDFS_DATANODE_USER=root
export HDFS_ZKFC_USER=root
export HDFS_JOURNALNODE_USER=root
export YARN_RESOURCEMANAGER_USER=root
export YARN_NODEMANAGER_USER=root

修改配置文件

[root@node01 hadoop]# vim core-site.xml

<property>
<name>fs.defaultFS</name>
<value>hdfs://hdfs-lzj</value>
</property>
<property>
<name>hadoop.tmp.dir</name>
<value>/var/lzj/hadoop/ha</value>
</property>
<property>
<name>hadoop.http.staticuser.user</name>
<value>root</value>
</property>
<property>
<name>ha.zookeeper.quorum</name>
<value>node01:2181,node02:2181,node03:2181</value>
</property>

[root@node01 hadoop]# vim hdfs-site.xml

<property>
<name>dfs.nameservices</name>
<value>hdfs-lzj</value>
</property>
<property>
<name>dfs.ha.namenodes.hdfs-lzj</name>
<value>nn1,nn2</value>
</property>
<property>
<name>dfs.namenode.rpc-address.hdfs-lzj.nn1</name>
<value>node01:8020</value>
</property>
<property>
<name>dfs.namenode.rpc-address.hdfs-lzj.nn2</name>
<value>node02:8020</value>
</property>
<property>
<name>dfs.namenode.http-address.hdfs-lzj.nn1</name>
<value>node01:9870</value>
</property>
<property>
<name>dfs.namenode.http-address.hdfs-lzj.nn2</name>
<value>node02:9870</value>
</property>
<property>
<name>dfs.namenode.shared.edits.dir</name>
<value>qjournal://node01:8485;node02:8485;node03:8485/hdfs-lzj</value>
</property>
<property>
<name>dfs.journalnode.edits.dir</name>
<value>/var/lzj/hadoop/ha/qjm</value>
</property>
<property>
<name>dfs.client.failover.proxy.provider.hdfs-lzj</name>
<value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxy
Provider</value>
</property>
<property>
<name>dfs.ha.fencing.methods</name>
<value>sshfence</value>
<value>shell(true)</value>
</property>
<property>
<name>dfs.ha.fencing.ssh.private-key-files</name>
<value>/root/.ssh/id_rsa</value>
</property>
<property>
<name>dfs.ha.automatic-failover.enabled</name>
<value>true</value>
</property>
<property>
<name>dfs.replication</name>
<value>2</value>
</property>

[root@node01 hadoop]# vim workers

node01
node02
node03

拷贝分发软件

将配置好的软件分发到其他主机

[root@node02 ~]# scp -r root@node01:/opt/lzj/hadoop-3.1.2 /opt/lzj/
[root@node03 ~]# scp -r root@node01:/opt/lzj/hadoop-3.1.2 /opt/lzj/

修改环境变量

[root@node01 hadoop]# vim /etc/profile

export HADOOP_HOME=/opt/lzj/hadoop-3.1.2
export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH

将环境变量拷贝到其他主机

[root@node01 lzj]# scp /etc/profile root@node02:/etc/profile
[root@node01 lzj]# scp /etc/profile root@node03:/etc/profile

重新加载三台服务器的环境变量

【123】# source /etc/profile

首先启动Zookeeper

【123】zkServer.sh start
【123】zkServer.sh status

启动JournalNode

【123】 hdfs --daemon start journalnode

格式化NameNode

[root@node01 lzj]# hdfs namenode -format
[root@node01 lzj]# hdfs --daemon start namenode
[root@node02 lzj]# hdfs namenode -bootstrapStandby
[root@node01 lzj]# hdfs zkfc -formatZK
[root@node01 lzj]# start-dfs.sh

测试集群

http://node01:9870
http://node02:9870
[root@node01 ~]# hdfs dfs -mkdir -p /lzj
[root@node01 ~]# hdfs dfs -put zookeeper-3.4.5.tar.gz /lzj/
[root@node01 ~]# hdfs dfs -D dfs.blocksize=1048576 -put zookeeper-3.4.5.tar.gz /lzj/

关闭集群

[root@node01 ~]# stop-dfs.sh

Stopping namenodes on [node01]
Last login: Fri Oct 30 22:06:20 CST 2020 on pts/0
node01: Warning: Permanently added 'node01,192.168.58.101' (ECDSA) to
the list of known hosts.
Stopping datanodes
Last login: Fri Oct 30 22:16:34 CST 2020 on pts/0
node03: Warning: Permanently added 'node03,192.168.58.103' (ECDSA) to
the list of known hosts.
node02: Warning: Permanently added 'node02,192.168.58.102' (ECDSA) to
the list of known hosts.
node01: Warning: Permanently added 'node01,192.168.58.101' (ECDSA) to
the list of known hosts.
Stopping secondary namenodes [node02]
Last login: Fri Oct 30 22:16:35 CST 2020 on pts/0
node02: Warning: Permanently added 'node02,192.168.58.102' (ECDSA) to
the list of known hosts.

关机拍摄快照
[123]# shutdown -h now

Hadoop3.x新特性

Erasure Encoding

简介
HDFS默认情况下，Block的备份系数是3，一个原始数据块和其他2个副本。
其中2个副本所需要的存储开销各站100%，这样使得200%的存储开销
正常操作中很少访问具有低IO活动的冷数据集的副本，但是仍然消耗与原始数据集相同的资
源量。
EC技术
EC(擦除编码)和HDFS的整合可以保持与提供存储效率相同的容错。
HDFS:一个副本系数为3，要复制文件的6个块，需要消耗6*3=18个块的磁盘空间
EC:6个数据块，3个奇偶校验块
擦除编码需要在执行远程读取时，对数据重建带来额外的开销，因此他通常用于存储不太频繁
访问的数据

NameNode

在Hadoop3中允许用户运行多个备用的NameNode。
例如，通过配置三个NameNode（1个Active NameNode和2个Standby NameNode）和5个
JournalNodes节点
集群可以容忍2个NameNode节点故障。

服务器端口

早些时候，多个Hadoop服务的默认端口位于Linux端口范围以内。
因此，具有临时范围冲突端口已经被移除该范围

DataNode

单个数据节点配置多个数据磁盘，在正常写入操作期间，数据被均匀的划分，因此，磁盘被均匀填
充。
在维护磁盘时，添加或者替换磁盘会导致DataNode节点存储出现偏移
这种情况在早期的HDFS文件系统中，是没有被处理的。

Hadoop3通过新的内部DataNode平衡功能来处理这种情况，这是通过hdfs diskbalancer CLI来进行调用的。执行之后，DataNode会进行均衡处理

蚊子腿

JDK：
Hadoop3中，最低版本要求是JDK8，所以低于JDK8的版本需要对JDK进行升级，方可安装使
用Hadoop3

Yarn:
提供YARN的时间轴服务V.2，以便用户和开发人员可以对其进行测试，并提供反馈意见
优化Hadoop Shell脚本
重构Hadoop Client Jar包
支持随机Container
MapReduce任务级本地优化
支持文件系统连接器

访问Hadoop集群

HDFS Java API

一、简介

想要使用 HDFS API，需要导入依赖 hadoop-client。如果是 CDH 版本的 Hadoop，还需要额外指明其仓库地址：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.heibaiying</groupId>
    <artifactId>hdfs-java-api</artifactId>
    <version>1.0</version>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <hadoop.version>2.6.0-cdh5.15.2</hadoop.version>
    </properties>
    <!---配置 CDH 仓库地址-->
    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>
    <dependencies>
        <!--Hadoop-client-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

二、API的使用

2.1 FileSystem

FileSystem 是所有 HDFS 操作的主入口。由于之后的每个单元测试都需要用到它，这里使用 @Before 注解进行标注。

private static final String HDFS_PATH = "hdfs://192.168.0.106:8020";
private static final String HDFS_USER = "root";
private static FileSystem fileSystem;
@Before
public void prepare() {
    try {
        Configuration configuration = new Configuration();
        // 这里我启动的是单节点的 Hadoop,所以副本系数设置为 1,默认值为 3
        configuration.set("dfs.replication", "1");
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, HDFS_USER);
    } catch (IOException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (URISyntaxException e) {
        e.printStackTrace();
    }
}
@After
public void destroy() {
    fileSystem = null;
}

2.2 创建目录

支持递归创建目录：

@Test
public void mkDir() throws Exception {
    fileSystem.mkdirs(new Path("/hdfs-api/test0/"));
}

2.3 创建指定权限的目录

FsPermission(FsAction u, FsAction g, FsAction o) 的三个参数分别对应：创建者权限，同组其他用户权限，其他用户权限，权限值定义在 FsAction 枚举类中。

@Test
public void mkDirWithPermission() throws Exception {
    fileSystem.mkdirs(new Path("/hdfs-api/test1/"),
            new FsPermission(FsAction.READ_WRITE, FsAction.READ, FsAction.READ));
}

2.4 创建文件，并写入内容

@Test
public void create() throws Exception {
    // 如果文件存在，默认会覆盖, 可以通过第二个参数进行控制。第三个参数可以控制使用缓冲区的大小
    FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/a.txt"),
                                               true, 4096);
    out.write("hello hadoop!".getBytes());
    out.write("hello spark!".getBytes());
    out.write("hello flink!".getBytes());
    // 强制将缓冲区中内容刷出
    out.flush();
    out.close();
}

2.5 判断文件是否存在

@Test
public void exist() throws Exception {
    boolean exists = fileSystem.exists(new Path("/hdfs-api/test/a.txt"));
    System.out.println(exists);
}

2.6 查看文件内容

查看小文本文件的内容，直接转换成字符串后输出：

@Test
public void readToString() throws Exception {
    FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs-api/test/a.txt"));
    String context = inputStreamToString(inputStream, "utf-8");
    System.out.println(context);
}

inputStreamToString 是一个自定义方法，代码如下：

/**
 * 把输入流转换为指定编码的字符
 *
 * @param inputStream 输入流
 * @param encode      指定编码类型
 */
private static String inputStreamToString(InputStream inputStream, String encode) {
    try {
        if (encode == null || ("".equals(encode))) {
            encode = "utf-8";
        }
        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode));
        StringBuilder builder = new StringBuilder();
        String str = "";
        while ((str = reader.readLine()) != null) {
            builder.append(str).append("\n");
        }
        return builder.toString();
    } catch (IOException e) {
        e.printStackTrace();
    }
    return null;
}

2.7 文件重命名

@Test
public void rename() throws Exception {
    Path oldPath = new Path("/hdfs-api/test/a.txt");
    Path newPath = new Path("/hdfs-api/test/b.txt");
    boolean result = fileSystem.rename(oldPath, newPath);
    System.out.println(result);
}

2.8 删除目录或文件

public void delete() throws Exception {
    /*
     *  第二个参数代表是否递归删除
     *    +  如果 path 是一个目录且递归删除为 true, 则删除该目录及其中所有文件;
     *    +  如果 path 是一个目录但递归删除为 false,则会则抛出异常。
     */
    boolean result = fileSystem.delete(new Path("/hdfs-api/test/b.txt"), true);
    System.out.println(result);
}

2.9 上传文件到HDFS

@Test
public void copyFromLocalFile() throws Exception {
    // 如果指定的是目录，则会把目录及其中的文件都复制到指定目录下
    Path src = new Path("D:\\BigData-Notes\\notes\\installation");
    Path dst = new Path("/hdfs-api/test/");
    fileSystem.copyFromLocalFile(src, dst);
}

2.10 上传大文件并显示上传进度

@Test
    public void copyFromLocalBigFile() throws Exception {
        File file = new File("D:\\kafka.tgz");
        final float fileSize = file.length();
        InputStream in = new BufferedInputStream(new FileInputStream(file));
        FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/kafka5.tgz"),
                new Progressable() {
                  long fileCount = 0;
                  public void progress() {
                     fileCount++;
                     // progress 方法每上传大约 64KB 的数据后就会被调用一次
                     System.out.println("上传进度：" + (fileCount * 64 * 1024 / fileSize) * 100 + " %");
                   }
                });
        IOUtils.copyBytes(in, out, 4096);
    }

2.11 从HDFS上下载文件

@Test
public void copyToLocalFile() throws Exception {
    Path src = new Path("/hdfs-api/test/kafka.tgz");
    Path dst = new Path("D:\\app\\");
    /*
     * 第一个参数控制下载完成后是否删除源文件,默认是 true,即删除;
     * 最后一个参数表示是否将 RawLocalFileSystem 用作本地文件系统;
     * RawLocalFileSystem 默认为 false,通常情况下可以不设置,
     * 但如果你在执行时候抛出 NullPointerException 异常,则代表你的文件系统与程序可能存在不兼容的情况 (window 下常见),
     * 此时可以将 RawLocalFileSystem 设置为 true
     */
    fileSystem.copyToLocalFile(false, src, dst, true);
}

2.12 查看指定目录下所有文件的信息

public void listFiles() throws Exception {
    FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfs-api"));
    for (FileStatus fileStatus : statuses) {
        //fileStatus 的 toString 方法被重写过，直接打印可以看到所有信息
        System.out.println(fileStatus.toString());
    }
}

FileStatus 中包含了文件的基本信息，比如文件路径，是否是文件夹，修改时间，访问时间，所有者，所属组，文件权限，是否是符号链接等，输出内容示例如下：

FileStatus{
path=hdfs://192.168.0.106:8020/hdfs-api/test; 
isDirectory=true; 
modification_time=1556680796191; 
access_time=0; 
owner=root; 
group=supergroup; 
permission=rwxr-xr-x; 
isSymlink=false
}

2.13 递归查看指定目录下所有文件的信息

@Test
public void listFilesRecursive() throws Exception {
    RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/hbase"), true);
    while (files.hasNext()) {
        System.out.println(files.next());
    }
}

和上面输出类似，只是多了文本大小，副本系数，块大小信息。

LocatedFileStatus{
path=hdfs://192.168.0.106:8020/hbase/hbase.version; 
isDirectory=false; 
length=7; 
replication=1; 
blocksize=134217728; 
modification_time=1554129052916; 
access_time=1554902661455; 
owner=root; group=supergroup;
permission=rw-r--r--; 
isSymlink=false}

2.14 查看文件的块信息

@Test
public void getFileBlockLocations() throws Exception {
    FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfs-api/test/kafka.tgz"));
    BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
    for (BlockLocation block : blocks) {
        System.out.println(block);
    }
}

块输出信息有三个值，分别是文件的起始偏移量 (offset)，文件大小 (length)，块所在的主机名 (hosts)。

0,57028557,hadoop001

这里我上传的文件只有 57M(小于 128M)，且程序中设置了副本系数为 1，所有只有一个块信息。

标签：HDFS,node01,文件,--,数据,Hadoop,NameNode,root,节点
From： https://www.cnblogs.com/abldh12/p/16990983.html

Hadoop--HDFS

Hadoop 3.1.2

算法复杂度

空间复杂度

时间复杂度

十大排序算法

归并排序

大数据思维

文件的数据类型

DataNode(DN)

SecondaryNameNode

安全模式

Hadoop1的困境

设计思想

Hadoop-Federation

单NN局限性

Federation(联邦)

搭建高可用集群

HDFS Java API

一、简介

二、API的使用

2.1 FileSystem

2.2 创建目录

2.3 创建指定权限的目录

2.4 创建文件，并写入内容

2.5 判断文件是否存在

2.6 查看文件内容

2.7 文件重命名

2.8 删除目录或文件

2.9 上传文件到HDFS

2.10 上传大文件并显示上传进度

2.11 从HDFS上下载文件

2.12 查看指定目录下所有文件的信息

2.13 递归查看指定目录下所有文件的信息

2.14 查看文件的块信息

相关文章

赞助商

阅读排行

Hadoop--HDFS

Hadoop 3.1.2

算法复杂度

空间复杂度

时间复杂度

十大排序算法

归并排序

大数据思维

文件的数据类型

DataNode(DN)

SecondaryNameNode

安全模式

Hadoop1的困境

设计思想

Hadoop-Federation

单NN局限性

Federation(联邦)

搭建高可用集群

HDFS Java API

一、 简介

二、API的使用

2.1 FileSystem

2.2 创建目录

2.3 创建指定权限的目录

2.4 创建文件，并写入内容

2.5 判断文件是否存在

2.6 查看文件内容

2.7 文件重命名

2.8 删除目录或文件

2.9 上传文件到HDFS

2.10 上传大文件并显示上传进度

2.11 从HDFS上下载文件

2.12 查看指定目录下所有文件的信息

2.13 递归查看指定目录下所有文件的信息

2.14 查看文件的块信息

相关文章

赞助商

阅读排行

一、简介