大数据学习——HDFS

1.HDFS概述

HDFS是一种分布式文件管理系统，适合一次写入，多次读出的场景。

1.1HDFS 架构

HDFS为主/从架构，主要由以下几个部分组成：

(1)NameNode：管理者。管理HDFS的名称空间；配置副本策略；管理数据块(Block)映射信息；处理客户端读写请求。

(2)DataNode：NameNode下达命令，DataNode执行实际操作。存储实际的数据块；执行数据块的读/写操作。

(3)Client：就是客户端。文件切分。文件上传HDFS的时候，Client将文件切分成一个一个的Block，然后进行上传；与NameNode交互，获取文件的位置信息；与DataNode交互，读取或者写入数据；Client可以管理和访问HDFS，比如对HDFS增删查改操作。

(4)Secondary NameNode：用于辅助NameNode，分担其工作量，比如定期合并Fsimage和Edits，并推送给NameNode ；紧急情况下，可辅助恢复NameNode。当NameNode挂掉的时候，它并不能马上替换NameNode并提供服务。

1.2HDFS 文件块大小

HDFS块的大小设置主要取决于磁盘传输速率, 寻址时间为传输时间的1%时，则为最佳状态。HDFS中的文件在物理上是分块存储(Block )， 块的大小可以通过配置参数( dfs.blocksize)来规定，默认大小是128M。如果寻址时间约为10ms，即查找到目标block的时间为10ms。而目前磁盘的传输速率普遍为100MB/s。因此，传输时间=10ms/0.01=1000ms=1s。

(1)HDFS的块设置太小，会增加寻址时间，程序一直在找块的开始位置；

(2)如果块设置的太大，从磁盘传输数据的时间会明显大于定位这个块开始位置所需的时间。导致程序在处理这块数据时，会非常慢。

1.3 HDFS优缺点

优点：

（1）可靠性：HDFS使用数据冗余和故障恢复机制来保证数据的高可靠性。它通过数据块复制和副本策略来处理DataNode的故障，从而提供了数据的容错能力。

（2）扩展性：HDFS能够处理大规模数据存储和处理需求。它支持水平扩展，可以轻松地添加更多的节点来增加存储容量和吞吐量。

（3）高吞吐量：HDFS设计用于处理大量数据的批量读写操作，具有高吞吐量的特点。它通过数据切块、数据本地性和并行操作等方式来实现高效的数据处理

（4）简化管理：HDFS提供了简化的数据管理和命名空间管理机制。它通过NameNode来管理文件系统的元数据，并提供了简单的命令行和API来管理文件和目录。

（5）易于集成：HDFS与其他Hadoop生态系统组件（如MapReduce、Hive、Spark等）紧密集成，可以无缝地与它们配合使用。

缺点：

（1）不适合低延迟访问：HDFS主要针对大数据批处理场景进行优化，对于低延迟和小文件访问的需求不太适合。由于数据切块和副本机制，以及通过网络进行数据传输，HDFS在低延迟场景下可能存在一定的性能损失。

（2）不支持并发写入：HDFS的设计目标是支持大规模数据的批量读写操作，而不是频繁的并发写入操作。在并发写入的场景下，可能会出现性能瓶颈或数据一致性的问题。

（3）存储冗余：HDFS通过数据块复制来提供数据的容错能力，这会导致存储冗余。默认情况下，每个数据块会有三个副本，这会占用更多的存储空间。

（4）复杂性：HDFS的架构相对复杂，需要管理多个组件（如NameNode、DataNode、Secondary NameNode等），以及进行适当的配置和调优。这对于一些非专业的用户来说可能会有一定的学习和管理成本。

2 HDFS 的读写流程

2.1 HDFS 写数据流程

2.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会放入一个应答队列等待应答。

(8)当一个 Block传输完成之后，客户端再次请求 NameNode上传第二个 Block的服务器。(重复执行3-7 步)。

2.2 HDFS 读数据流程

(1)客户端通过DistributedFileSystem 向NameNode 请求下载文件，NameNode 通过查询元数据，找到文件块所在的DataNode 地址。

(2)挑选一台DataNode(就近原则，然后随机)服务器，请求读取数据。

(3)DataNode 开始传输数据给客户端(从磁盘里面读取数据输入流，以文件块为单位来做校验)。

(4)客户端以文件块为单位接收，先在本地缓存，然后写入目标文件。

2.3副本存储

2.3.1存储节点选择

2.3.2节点距离计算

在 HDFS写数据的过程中，NameNode会选择距离待上传数据最近距离的 DataNode接收数据。这个最近距离的计算方式为：两个节点到达最近的共同祖先的距离总和。

3 NameNode ，SecondaryNameNode和DataNode

3.1 NN 和2NN 工作机制

第一阶段：NameNode 启动

(1)第一次启动NameNode 格式化后，创建Fsimage 和Edits 文件。如果不是第一次启动，直接加载编辑日志和镜像文件到内存。

(2)客户端对元数据进行增删改的请求。

(3)NameNode 记录操作日志，更新滚动日志。

(4)NameNode 在内存中对元数据进行增删改。

第二阶段：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。

3.2 DataNode 工作机制

(1)一个数据块在DataNode 上以文件形式存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据包括数据块的长度，块数据的校验和，以及时间戳。

(2)DataNode 启动后向NameNode 注册，通过后，周期性(6 小时)的向NameNode 上报所有的块信息。DN 向NN 汇报当前解读信息的时间间隔，默认6 小时；DN 扫描自己节点块信息列表的时间，默认6 小时；

(3)心跳是每3 秒一次，心跳返回结果带有NameNode 给该DataNode 的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10 分钟+30秒没有收到某个DataNode 的心跳，则认为该节点不可用。

(4)集群运行中可以安全加入和退出一些机器。

3.3数据完整性保证

DataNode 节点保证数据完整性的方法：当DataNode 读取Block 的时候，它会计算CheckSum。如果计算后的CheckSum，与Block 创建时值不一样，说明Block 已经损坏。Client 读取其他DataNode 上的Block。常见的校验算法crc(32)，md5(128)，sha1(160)。DataNode 在其文件创建后周期验证CheckSum。

4 HDFS 的 Shell操作

4.1基本语法

hadoop fs

具体命令或者

hdfs dfs

具体命令

4.2常用命令

4.2.1启停 Hadoop集群

sbin/start-dfs.sh

sbin/start-yarn.sh

sbin/stop-dfs.sh

sbin/stopt-yarn.sh

4.2.2文件上传下载

(1)

-moveFromLocal/-moveToLocal

：从本地剪切到HDFS/反之

hadoop fs -moveFromLocal

文件位置 目标路径

(2)

-copyFromLocal/-copyToLocal

：从本地文件系统中拷贝文件到 HDFS路径去/反之

(3)

-put/-get

：同(2)

(4)

-appendToFile

：追加一个文件到已经存在的文件末尾

4.2.2文件查看

(1)

-ls

： 显示目录信息

(2)

-cat

：显示文件内容

4.2.3文件操作

(1)

-chgrp

、

 -chmod

、

 -chown

：与Linux文件系统中的用法一样，修改文件所属权限第一阶段：NameNode 启动

(2)

-mkdir

：创建路径

(3)

-cp

：从 HDFS的一个路径拷贝到 HDFS的另一个路径

(4)

-mv

：在 HDFS目录中移动文件

(7)

-tail

：显示一个文件的末尾数据

(8)

-rm/-rm -r

：删除/递归删除文件或文件夹

5. HDFS 的 API操作

此部分参照46_尚硅谷_Hadoop_HDFS_API环境准备_哔哩哔哩_bilibili

5.1准备Hadoop的 Windows依赖文件。

配置HADOOP_HOME环境变量，在 IDEA中创建一个 Maven工程HdfsClientDemo，并导入相应的依赖项 +日志添加

<dependencies>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-client</artifactId>

            <version>3.1.3</version>

        </dependency>

        <dependency>

            <groupId>junit</groupId>

            <artifactId>junit</artifactId>

            <version>4.12</version>

        </dependency>

        <dependency>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-log4j12</artifactId>

            <version>1.7.30</version>

        </dependency>

    </dependencies>

在项目的src/main/resources目录下，新建一个文件，命名为“ log4j.properties”，在文件中填入

创建包名 com.hdfs_test.hdfs，创建 HdfsClient类

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] -%m%n

log4j.appender.logfile=org.apache.log4j.FileAppender

log4j.appender.logfile.File=target/spring.log

log4j.appender.logfile.layout=org.apache.log4j.PatternLayout

log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] -%m%n

5.2API操作

package com. hdfs_test.hdfs;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.*;

import org.junit.After;

import org.junit.Before;

import org.junit.Test;

import java.io.IOException;

import java.net.URI;

import java.net.URISyntaxException;

import java.util.Arrays;

/**

 * 客户端代码常用套路

 * 1. 获取一个客户端对象

 * 2. 执行相关的操作命令

 * 3. 关闭资源

 * HDFS zookeeper

 */

public class HdfsClient {

    private FileSystem fs;

    @Before

    public void init() throws URISyntaxException, IOException, InterruptedException {

        // 连接的集群nn地址

        URI uri = new URI("hdfs://hadoop102:8020");

        // 创建一个配置文件

        Configuration configuration = new Configuration();

        configuration.set("dfs.replication", "2");

        // 创建一个用户

        String user = "test";

        // 1. 获取到了一个客户端对象

        fs = FileSystem.get(uri, configuration, user);

    }

    @After

    public void close() throws IOException {

        // 3. 关闭资源

        fs.close();

    }

    // 创建目录

    @Test

    public void testmkdir() throws URISyntaxException, IOException, InterruptedException {
        // 2. 创建一个文件夹

        fs.mkdirs(new Path("/xiyou/huaguoshan"));

    }

    // 上传

    /**

     * 参数优先级

     * 代码里的配置 > 在项目资源目录下的配置文件 > hdfs-site.xml > hdfs-default.xml

     * @throws IOException

     */

    @Test

    public void testsPut() throws IOException {

        // 参数解读：参数一：表示删除原数据；参数二：是否允许覆盖；参数三：原数据路径；参数四：目的地路径

        fs.copyFromLocalFile(false, true, new Path("E:\\LayerData.vtk"), new Path("hdfs://hadoop102/xiyou/huaguoshan"));

    }

    // 文件下载

    @Test

    public void testGet() throws IOException {

        // 参数的解读：参数一：原文件是否删除；参数二：原文件路径HDFS；参数三：目标地址路径：Win；参数四

        // fs.copyToLocalFile(true, new Path("hdfs://hadoop104/xiyou/huaguoshan"), new Path("E:\\test.vtk"), true);

        fs.copyToLocalFile(false, new Path("hdfs://hadoop102/a.txt"), new Path("E:\\"), false);

    }

    // 删除

    @Test

    public void testRm() throws IOException {

        // 参数解读：参数1；要删除的路径：参数2：是否递归删除

        // 删除文件

        // fs.delete(new Path("/jdk-8u212-linux-x64.tar.gz"), false);

        // 删除空目录

        fs.delete(new Path("/xiyou"), false);

        // 删除非空目录

        // fs.delete(new Path("/jingguo"),true);

    }

    // 文件的移动和更名

    @Test

    public void testmv() throws IOException {

        // 参数解读：参数一：原文件路径；参数二：目标文件路径

        // 对文件名称的更改

        // fs.rename(new Path("/wcinput/word.txt"), new Path("/wcinput/words.txt"));

        // 文件的移动和更名

        // fs.rename(new Path("/wcinput/words.txt"), new Path("/word.txt"));

        // 目录更名

        fs.rename(new Path("/wcoutput"), new Path("/wcintput"));

    }

    // 获取文件详细信息

    @Test

    public void fileDetail() throws IOException {

        // 获取文件所有信息

        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

        // 遍历文件

        while (listFiles.hasNext()) {

            LocatedFileStatus fileStatus = listFiles.next();

            System.out.println("==========" + fileStatus.getPath() + "==========");

            System.out.println(fileStatus.getPermission());

            System.out.println(fileStatus.getOwner());

            System.out.println(fileStatus.getGroup());

            System.out.println(fileStatus.getLen());

            System.out.println(fileStatus.getModificationTime());

            System.out.println(fileStatus.getReplication());

            System.out.println(fileStatus.getBlockSize());

            System.out.println(fileStatus.getPath().getName());

            // 获取块信息

            BlockLocation[] blockLocations = fileStatus.getBlockLocations();

            System.out.println(Arrays.toString(blockLocations));

        }

    }

    // 判断是文件还是文件夹

    @Test

    public void testFile() throws IOException {

        FileStatus[] listStatus = fs.listStatus(new Path("/"));

        for (FileStatus status : listStatus) {

            if (status.isFile()) {

                System.out.println("文件：" + status.getPath().getName());

            } else {

                System.out.println("目录：" + status.getPath().getName());

            }

        }

    }

}