Hadoop(二）HDFS

海量数据处理

分而治之

核心思想：

把数据分发到多个节点
移动计算到数据附近
计算节点进行本地数据处理
优选顺序，次之随机读

一、HDFS概述

修改，先删除，再重新生成

1.架构

namenode维护着HDFS中存储的文件的元数据，以及每个文件块的列表，以及块所在datanode的信息。namenode会把元数据信息加载到内存中，管理副本数，默认副本是三个副本，每个block复制到多个datanode上存储。

通常启动两个namenode，active和standby。

Datanode真正数据块存储，执行客户端读写请求，datanode也会周期性的与namenode进行通信，汇报datanode上block的信息及其运行状态。

2.Active NameNode

主Master（只有一个）
管理HDFS文件系统的命名空间
维护文件元数据信息
管理副本策略，默认3个副本
处理客户端读写请求

HDFS文件系统和Linux文件系统非常相似，都是通过文件和目录管理的，所有目录和文件的层级关系可以看成一棵文件树。

由上图可知，namenode维护着文件系统树，及整棵树内所有的文件和目录，这些信息以命名空间镜像文件fsimage和编辑日志文件edits两种文件形式，永久保存在本地磁盘当中。

为了快速访问，在集群运行的时候，会把命名空间信息加载到内存当中。

namenode在内存中维护着文件的元数据，包括文件被切分成哪些块，每个块的副本数，生成时间，文件的权限，以及块所在的datanode位置的映射信息。这个映射信息并不保存到磁盘中，即fsimage当中不保存block块和datanode的映射关系。datanode周期性的向namenode发送心跳信息，汇报其存储的所有块的列表信息，namenode通过datanode上传列表信息，就会确保拥有最新的块映射信息。如果namenode重启，datanode向namenode汇报自己存储的block块的信息，namenode可以汇总datanode上存储的块信息，在内存中重建block和datanode的映射。

在HDFS中，某个datanode心跳超时，namenode就认为这个datanode不可用，就把该datanode标记为死亡，并且不会向这个标记为死亡的datanode转发任何新的读写请求，如果datanode这台机器被标记为死亡，存储的block块不可用，导致块的副本数低于正常水平，namenode会在适当的时候拷贝部分，使副本保持正常的水平。

namenode还负责处理客户端的读写请求，客户端从namenode获取元数据信息，再根据这些信息与datanode进行联系，进行数据块真正的读写操作。

3.NameNode元数据文件

NameNode启动的时候，会先将fsimage文件中的元数据加载到内存中，并执行编辑日志中的各项操作。一旦在内存中建立文件系统元数据的映像，则创建一个新的fsimage文件和一个空的edits文件，在这个过程中namenode运行在安全模式下。
系统中数据块的位置并不是由namenode维护的，而是以块列表的形势存储在datanode中，在安全模式中，datanode会向namenode发送最新的块列表信息，namenode在内存中建立一块和datanode的映射关系， namenode了解到足够多的块位置信息之后，namenode会对外提供服务。

在NameNode运行期间，客户端对HDFS的写操作都保存到edits文件中，久而久之edits文件会变得很大，虽然这对NameNode运行的时候是没有影响的，但是在NameNode重启的时候，NameNode先将fsimage中的内容映射到内存中，然后再一条一条执行edits编辑日志中的操作。当edits文件非常大的时候会导致namenode启动的时间非常漫长，而在这段时间中HDFS处于安全模式，所以需要在Namenode运行的时候将edits和fsimage定时进行合并，减小edits文件的大小。

fsimage文件：是HDFS文件系统存于硬盘中的元数据检查点（即全量），里面记录了自最后一次检查点之前HDFS文件系统中所有目录和文件的序列化信息；

edits log文件：保存了自最后一次检查点之后所有针对HDFS文件系统的操作（即增量），比如：增加文件、重命名文件、删除目录等等

NameNode定期将内存中的新增的edits与fsimage合并保存到磁盘。操作被记录到edits中，fsimage并不会同步记录操作，而是在namenode定期地设置检查点，到点将edits与fsimage进行合并，保存到磁盘当中，使fsimage定期的与内存中的元数据信息保持同步，这就保证了NameNode内存中保存一份最新的镜像信息，新的镜像内容=fsimage（旧的镜像内容）+edits（新增的信息）

4.DataNode

Slave工作节点，可以启动多个
存储数据块
执行客户端的读写请求操作
通过心跳机制定期向NameNode汇报运行状态和所有块列表信息
在集群启动时DataNode向NameNode提供存储的Block块列表信息

5.Block数据块

文件写入到HDFS会被切分成若干个Block块
数据块大小固定，默认大小128MB，可自定义修改
HDFS最小存储单元
若一个块的大小小于设置的数据块大小，则不会占用整个块的空间
默认情况下每个Block有三个副本

6.Client

文件切分
与NameNode交互获取文件元数据信息
与DataNode交互，读取或写入数据
管理HDFS

7.不适合存储小文件

8.Secondary NameNode和Standby NameNde

HDFS单NameNode，即没有配置高可用，会有Secondary NameNode

Secondary NameNode负责每隔一段时间将旧的fsimage文件和edits log文件merge成新的fsimage并替换，即为NameNode 合并编辑日志edits log，减少 NameNode 启动时间；
非实时merge，一旦NameNode挂了，可能会导致元数据丢失；

高可用下会有Standby NameNode

实时merge，一旦前者挂了，后者能够马上顶上，不会出现元数据丢失；

同步edits编辑日志，定期合并fsimage与edits到本地磁盘
Active NameNode故障快速切换为新的Active

二、高可用原理

基于QJM的高可用机制

搭建Hadoop集群时，在三台及以上，一般奇数个机器上启动JournalNode，组成QJM共享存储系统，该系统非常轻量级，一般不会出现问题

一般会在两个namenode节点上启动两个JournalNode，在另一台节点上再启动一个JournalNode，这个系统里面存储edit log这个编辑日志

1.数据同步

通常启动两个namenode，一个active，一个standby

standbynamenode定期从QJM存储系统里同步activenamenode的元数据信息，使节点间元数据保持一致

active在处理客户端提交的创建文件，移除文件等写请求操作的时候，会首先把这些记录，记录到edit编辑日志中，同时也会同步阻塞并行的向JorunalNode集群中每一个JournalNode发送写请求，大多数Journalnode节点写成功，就认为整个集群写入edit成功了，最后修改内存中的元数据

actice会定对内存中的文件系统命名空间元数据信息创建检查点，在磁盘中生成fsimage镜像文件，持久化存储，另外一个standbynamenode定期从Journalnode集群中同步编辑日志edit，回放到其内存中，也会定期对内存中的元数据信息创建检查点，在磁盘中生成fsimage文件，持久化存储。

2.主备切换

主要使用主备切换控制器，ZKFC

当启动namenode的时候，也会在namenode所在的节点上启动ZKFC守护进程，作为主备切换的控制器，ZKFC启动时，会创建HealthMonitor和ActiveStandbyElector两个组件，前者循环检查namenode健康状况，后者使用zookeeper完成主备的选举。

在HDFS集群启动时，每个namenode对应一个ZKFC，每个ZKFC启动一个ActiveStandbyElector

每次启动时，都会尝试在zookeeper中创建临时锁节点，利用zookeeper写一致性保证最终只有一个activeelector创建锁节点成功。不管ActiveElector在zookeeper上是否创建成功临时锁节点，都会随后向zookeeper来注册监听事件，监听临时锁节点的删除事件

三、 HDFS文件写入流程

首先与namenode进行通信，创建远程的RPC请求，发起创建文件的请求，namenode接收到请求后，执行对新建文件的各种检查，以确保文件是不存在的，以及客户端有创建新文件的权限。所有检查结束后，返回运行写文件的消息。客户端接收到消息后，会把数据流式的写到客户端本地文件系统临时文件中，当临时文件大小达到block块大小（默认128M）的时候，客户端再次向namenode发送上传block块的请求，namenode根据请求在datanode信息池里检查datanode的状态，把存储block的datanode列表，包括备份节点的datanode返回给客户端

客户端接收到该列表，创建第一个datanode连接，请求将这组datanode列表构建成信息流通道

创建完数据流通道，客户端将以数据包的形式按照流式的方式写入到数据文档之中，首先将一个数据包package写入到第一个datanode，node01上，当第一个datanode写磁盘的时候，从第一datanode，通过数据管道将数据包发送到第二个datanode，第二个datanode开始写本地磁盘的时候，从第二个datanode发送数据包到第三个datanode，最后一个datanode写完之后，有确认信息，这个确认信息从保存了该数据包的节点通过管道（第9步）反馈给前一个节点，第二个节点反馈给第一个节点，第一个节点发送确认信息给客户端

等数据块传完之后，客户端发送最终的确认信息给namenode(第10步），第一数据块传输完成。

其他数据块重复上述4-10步

当整个文件写入完成被关闭时，namenode执行提交操作，从而使文件在集群中可见

四、读取流程

首先，客户端通过远程RPC调用，向namenode发送读文件请求(1)，namenode接收到请求后，检查读取文件是否存在，检查客户端是否有读该文件的权限(2.1)

namenode从文件的元数据中查询这个文件是由哪些block块组成，这些block块存储在哪些datanode上(2.2)

检查通过后，返回给客户端(3)，返回给客户端组成这个要读取文件的数据块列表以及数据块所在datanode的位置

列表中datanode的顺序是按照与客户端的由近到远的顺序来排列的，即客户端访问哪个机器更快一些，哪个排在前面

客户端接收到namenode返回的信息后，创建与存储第一个块的最近的datanode连接

比如在node01上获取block1

datanode将block这个数据传输给客户端，如果访问的datanode出现故障，就会访问备份数据块的数据节点，直到数据块传输完成，关闭与该datanode的连接

然后寻找下一个数据块

直到整个文件合并完成

客户端只要读取连续的数据流即可，其他操作都是透明的，读取完成关闭连接

五、操作

1.Hadoop服务脚本

2.HDFS文件操作命令

hadoop fs: 使用面最广，可以操作任何文件系统。

hadoop dfs和hdfs dfs: 只能操作HDFS文件系统相关(包括与Local FS间的操作)，hadoop dfs已经废弃，被hdfs dfs代替。

hadoop fs -ls 显示目录信息，递归-lsr
hadoop fs -mkdir /user/tguigu 在hdfs上创建目录
hadoop fs -moveFromlocal test.txt /user/tguigu/data 从本地剪切粘贴到hdfs
hadoop fs -appendTofile test.txt /user/tguigudata/test.txt 追加一个文件到已经存在的文件末尾
hadoop fs -cat 显示文件内容
hadoop fs -tail 显示一个文件的末尾
hadoop fs -cp /user/tguigu/../x.txt /user/tguigu/test../ 从hdfs的一个路径拷贝到hdfs的另一个路径
hadoop fs -mv /user/tguigu/../x.txt /.../ 在hdfs目录中移动文件
hadoop fs -get /user/tguigu/../x.txt ./ 等同于copyToLocal,就是从hdfs下载文件到本地
hadoop fs -getmerge /user/tguigu//test/* ./zaiyiqi.txt 合并下载多个文件
hadoop fs -put 等同于copyFromLocal上传
hadoop fs -rm 删除文件或文件夹

hadoop fs -rm -r 递归删除
hadoop fs -rmdir 删除空目录
hadoop fs -df 统计文件系统的可用空间
hadoop fs -du 统计文件的大小信息
hadoop fs -setrep 设置hdfs中文件的副本量数

3.HDFS API

public class HDFSClient {
    /**
     * 获取HDFS文件系统对象
     * @return
     * @throws IOException
     */
    private FileSystem getFileSystem() throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);//创建hdfs文件系统对象
        return fs;
    }

    /**
     * 读取hdfs中的文件内容
     * @param hdfsFilePath
     */
    public void readHDFSFile(String hdfsFilePath){
        BufferedReader reader = null;
        FSDataInputStream fsDataInputStream = null;
        //通过HDFS Java API读取HDFS中的文件
        try {
            Path path = new Path(hdfsFilePath);
            fsDataInputStream = this.getFileSystem().open(path);//根据path创建FSDataInputStream输入流对象
            reader = new BufferedReader(new InputStreamReader(fsDataInputStream));
            String line = "";
            while((line = reader.readLine()) != null){
                System.out.println(line);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (fsDataInputStream != null) {
                    fsDataInputStream.close();
                }

                if (reader != null) {
                    reader.close();
                }
            }catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 将本地文件内容写入到HDFS指定文件中
     * @param localFilePath
     * @param hdfsFilePath
     */
    public void writeHDFSFile(String localFilePath,String hdfsFilePath){
        FSDataOutputStream fsDataOutputStream = null;
        FileInputStream fileInputStream = null;
        Path path = new Path(hdfsFilePath);
        try {
            //根据path创建输出流对象
            fsDataOutputStream = this.getFileSystem().create(path);

            //创建读取本地文件的输入流对象
            fileInputStream = new FileInputStream(new File(localFilePath));

            IOUtils.copyBytes(fileInputStream,fsDataOutputStream,4096,false);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (fsDataOutputStream != null){
                    fsDataOutputStream.close();
                }
                if (fileInputStream != null){
                    fileInputStream.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] ars) {
        String hdfsFilePath = "hdfs://ns/hdfs_client/from_local_2_hdfs.txt";
        HDFSClient client = new HDFSClient();
//        client.readHDFSFile(hdfsFilePath);
        String localFilePath = "/Users/derek/hdfstest.txt";

        client.writeHDFSFile(localFilePath,hdfsFilePath);

        client.readHDFSFile(hdfsFilePath);

    }
}