阅读 1091

带你入坑大数据(二) --- HDFS的读写流程和一些重要策略

前言

前情回顾

如果说上一篇是在阐述HDFS最基础的理论知识,这一篇就是HDFS的主要工作流程,和一些较为有用的策略

补充一个问题,就是当我们 NameNode 挂掉,SecondaryNameNode作为新的NameNode上位时,它确实可以根据fsimage.ckpt把一部分元数据加载到内存,可是如果这时还有一部分操作日志在edits new中没有执行怎么办?

这时候有一个解决方案就是利用一个network fileSystem来解决,比如说集群中有一个服务器安装了一个nfs server,而在NameNode上再安装一个nfs client,此时客户端向HDFS写数据时,同时把向edits new中写的数据写一份到nfs server中,SecondaryNamenode就可以通过这个nfs server来获取此时断层的数据了

其他似乎也没啥可多说的,让我们直奔主题吧

以往链接

从零开始的大数据(一) --- HDFS的知识概述(上)

一、HDFS的读流程

之后的内容会围绕下图开始

1.认识角色

简单过一下图里面的角色,最大块的是一个client node,也就是说,这个节点上运行着客户端,如果实在是没搞清楚哪个是客户端,那也很简单,平时没事就执行

hadoop fs -ls /
复制代码

这个命令的机器,那就是客户端了,其他就是NameNode和DataNode,在client node上运行着一个JVM虚拟机,让HDFS client跑起来

2.步骤分析

① HDFS client调用文件系统的open方法

Distributed FileSystem顾名思义是一个分布式文件系统,它会通过RPC的方式远程过程调用NameNode里的open方法,这个open方法有什么作用呢,就是获取要读的文件的file block locations,也就是文件的block的位置,在上一讲我们也已经提到了,一个文件是会分割成128M一块的大小分别存储在各个数据节点的。

同时在执行open方法时,客户端会产生一个FSData InputStream的一个输入流对象(客户端读数据是从外部读回来的)

② FSData InputStream读数据

HDFS client调用FSData InputStream的read方法,同上也是远程过程调用DataNode的read方法,此时的读取顺序是由近到远,就是DataNode和client node的距离,这里所指的距离是一种物理距离,判定可以参考上一篇文章中机架的概念。

在联系上DataNode并成功读取后,关闭流就走完了一个正常的流程。

而且补充一下就是,上面Distributed FileSystem所调用的get block locations的方法只会返回部分数据块,get block locations会分批次地返回block块的位置信息。读block块理论上来说是依次读,当然也可以通过多线程的方式实现同步读。

③ 容错机制

1.如果client从DataNode上读取block时网络中断了如何解决?

此时我们会找到block另外的副本(一个block块有3个副本,上一篇已经说过了),并且通过FSData InputStream进行记录,以后就不再从中断的副本上读了。

2.如果一个DataNode挂掉了怎么办?

在上一篇中我们提到了一个HDFS的心跳机制,DataNode会隔一小时向NameNode汇报blockReport,比如现在的情况是,block1的三个副本分别存储在DataNode1,2,3上,此时DataNode1挂掉了。NameNode得知某个block还剩2个副本,此时携带这block的其余两个副本的DataNode2,3在向NameNode报告时,NameNode就会对它们中的某一个返回一个指令,把block1复制一份给其他正常的节点。让block1恢复成原本的3个副本。

3.client如何保证读取数据的完整性

因为从DataNode上读数据是通过网络来读取的,这说明会存在读取过来的数据是不完整的或者是错误的情况。

DataNode上存储的不仅仅是数据,数据还附带着一个叫做checkSum检验和(CRC32算法)的概念,针对于任何大小的数据块计算CRC32的值都是32位4个字节大小。此时我们的FSData InputStream向DataNode读数据时,会将与这份数据对应的checkSum也一并读取过来,此时FSData InputStream再对它读过来的数据做一个checkSum,把它与读过来的checkSum做一个对比,如果不一致,就重新从另外的DataNode上再次读取。

4.上一个问题完成后工作

FSData InputStream会告诉NameNode,这个DataNode上的这个block有问题了,NameNode收到消息后就会再通过心跳机制通知这个DataNode删除它的block块,然后再用类似2的做法,让正常的DataNode去copy一份正常的block数据给其它节点,保证副本数为3

代码简单示例(可跳过)
try {
        //
        String srcFile = "hdfs://node-01:9000/data/hdfs01.mp4";
        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(URI.create(srcFile),conf);
        FSDataInputStream hdfsInStream = fs.open(new Path(srcFile));

        BufferedOutputStream outputStream = new BufferedOutputStream(new FileOutputStream("/home/node1/hdfs02.mp4"));

        IOUtils.copyBytes(hdfsInStream, outputStream, 4096, true);
    } catch (IOException e) {
        e.printStackTrace();
    }
复制代码

二、HDFS写流程

写流程我们会按照下图来进行讲解,比读数据更加复杂一丢丢,角色基本没有改变所以就不详细介绍了

客户端向HDFS写数据的时候是把文件分块存储在HDFS的各个节点上,而规定了存储位置的是NameNode,所以Client node在存储文件时需要先和NameNode进行联系让它进行分配。

步骤分析

① 客户端调用分布式文件系统的create方法

和上面读的方法类似,不过这次调用的是Distributed FileSystem的create方法,此时也是通过远程调用NameNode的create方法

此时NameNode会进行的举措

  • 1.检测自己是否正常运行

  • 2.判断要创建的文件是否存在

  • 3.client是否有创建文件的权限

  • 4.对HDFS做状态的更改需要在edits log写日志记录

② 客户端调用输出流的write方法

create方法的返回值是一个OutputStream对象,为什么是output,因为是由HDFS去往DataNode去写数据,此时HDFS会调用这个OutputStream的write方法

但是有个问题,此时我们还不知道我们的这些block块要分别存放于哪些节点上,所以此时FSData OutputStream就要再和NameNode交互一下,远程过程调用NameNode的addBlock方法,这个方法返回的是各个block块分别需要写在哪3个DataNode上面。

此时OutputStream就完整得知了数据和数据该往哪里去写了

③ 具体的写流程分析

请看流程4.1,chunk是一个512字节大小的数据块,写数据的过程中数据是一字节一字节往chunk那里写的,当写满一个chunk后,会计算一个checkSum,这个checkSum是4个字节大小,计算完成后一并放入chunk,所以整一个chunk大小其实是512字节+4字节=516字节

上述步骤结束后,一个chunk就会往package里面放,package是一个64kb大小的数据包,我们知道64kb = 64 * 1024字节,所以这个package可以放非常多的chunk。

此时一个package满了之后,会把这个packjage放到一个data queue队列里面,之后会陆续有源源不断的package传输过来,图中用p1,p2···等表示

这时候开始真正的写数据过程

  1. data queue中的package往数据节点DataNode上传输,传输的顺序按照NameNode的addBlock()方法返回的列表依次传输

(ps:传输的类为一个叫做dataStreamer的类,而且其实addBlock方法返回的列表基本是按照离客户端物理距离由近到远的顺序的)

  1. 往DataNode上传输的同时也往确认队列ack queue上传输

  2. 针对DataNode中传输完成的数据做一个checkSum,并与原本打包前的checkSum做一个比较

  3. 校验成功,就从确认队列ack queue中删除该package,否则该package重新置入data queue重传

补充:1.以上逻辑归属于FSData OutputStream的逻辑

2.虽然本身一个block为128M,而package为64Kb,128M对于网络传输过程来说算是比较大,拆分为小包是为了可靠传输

3.网络中断时的举措:HDFS会先把整个pineline关闭,然后获取一个已存在的完整的文件的version,发送给NameNode后,由NameNode通过心跳机制对未正确传输的数据下达删除命令

4.如果是某个DataNode不可用,在1中我们也提到过了,通过心跳机制会通知其余的可用DataNode的其中一个进行copy到一个可用节点上

④ 写入结束后的行动

完成后通过心跳机制NameNode就可以得知副本已经创建完成,再调用addBlock()方法写之后的文件。

⑤ 流程总结

1.client端调用Distributed FileSystem的create,此时是远程调用了NameNode的create,此时NameNode进行4个操作,检测自己是否正常,文件是否存在,客户端的权限和写日志

2.create的返回值为一个FSData OutputStream对象,此时client调用流的write方法,和NameNode进行连接,NameNode的addBlock方法返回块分配的DataNode列表

3.开始写数据,先写在chunk,后package,置入data queue,此时两个操作,向DataNode传输,和放入ack queue,DataNode传输结束会检测checkSum,成功就删除ack queue的package,否则放回data queue重传

4.结束后关闭流,告诉NameNode,调用complete方法结束

简单代码示例(可跳过)

String source="/home/node1/hdfs01.mp4"; //linux中的文件路徑,demo存在一定数据

    //先确保/data目录存在
    String destination="hdfs://node-01:9000/data/hdfs01.mp4";//HDFS的路徑

    InputStream in = null;
    try {
        in = new BufferedInputStream(new FileInputStream(source));
        //HDFS读写的配置文件
        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(URI.create(destination),conf);

        //调用Filesystem的create方法返回的是FSDataOutputStream对象
        //该对象不允许在文件中定位,因为HDFS只允许一个已打开的文件顺序写入或追加
        OutputStream out = fs.create(new Path(destination));

        IOUtils.copyBytes(in, out, 4096, true);
    } catch (FileNotFoundException e) {
        System.out.println("exception");
        e.printStackTrace();
    } catch (IOException e) {
        System.out.println("exception1");
        e.printStackTrace();
    }
复制代码

3.Hadoop HA高可用

之前已经提到过,元数据是放在NameNode的内存中的,当元数据丢失,造成服务不可用,这时候就需要时间来恢复。HA就可以让用户感知不到这种问题。这需要yarn,MapReduce,zookeeper的支持

仅有一个NameNode时,当NameNode挂掉就需要把fsimage读到内存,然后把edits log所记录的日志重新执行一遍,元数据才能恢复,而这种做法需要大量的时间

所以解决方案就在于我们要花大量时间来恢复元数据metaData,所以解决的方案就是让集群瞬间变回可用状态即可。

通过设置一个stand by的NameNode,并和主NameNode的元数据保持一致,图中绿色的区域表示一个共享存储,主NameNode的元数据会传输至共享存储里面,让stand by的NameNode进行同步。

下面的DataNode会同时往两个NameNode发送blockReport,因为读取DataNode的块信息并不会很快,所以为了保证在active挂掉的时候,standby能立刻顶上位置,所以要事先读取块信息,同时这也是方便standby来构建它的元数据信息的途径。

active挂掉后让stand by立刻生效的机制是上面的FailoverControllerActive实现的,简称zkfc,它会定时ping主NameNode,如果发现NameNode挂掉,就会通知我们的zookeeper集群,然后集群的另一个FailoverControllerActive就会通知stand by。

4.Hadoop联邦

集群中的元数据会保存在NameNode的内存中,而这些元数据每份占用约150字节,对于一个拥有大量文件的集群来说,因为NameNode的metaData被占满,DataNode就无法写入了,联邦就可以帮助系统突破文件数上限

其实就是布置了多个NameNode来共同维护集群,来增加namespace,而且分散了NameNode的访问压力,而且客户端的读写互不影响。就是扩展性,高吞吐和隔离性

5.HDFS存储大量小文件

和刚刚的联邦的介绍时的情况一样,文件数量(每个文件元数据150byte)会影响到NameNode的内存

方案1:HAR文件方案

其实就是通过一个MR程序把许多小文件合并成一个大文件,需要启动Yarn

# 创建archive文件
hadoop archive -archiveName test.har -p /testhar -r 3 th1 th2 /outhar # 原文件还存在,需手动删除
# 查看archive文件
hdfs dfs -ls -R har:///outhar/test.har
# 解压archive文件
hdfs dfs -cp har:///outhar/test.har/th1 hdfs:/unarchivef
hadoop fs -ls /unarchivef	# 顺序
hadoop distcp har:///outhar/test.har/th1 hdfs:/unarchivef2 # 并行,启动MR
复制代码

方案2:Sequence File方案

其核心是以文件名为key,文件内容为value组织小文件。10000个100KB的小文件,可以编写程序将这些文件放到一个SequenceFile文件,然后就以数据流的方式处理这些文件,也可以使用MapReduce进行处理。一个SequenceFile是可分割的,所以MapReduce可将文件切分成块,每一块独立操作。不像HAR,SequenceFile支持压缩。在大多数情况下,以block为单位进行压缩是最好的选择,因为一个block包含多条记录,压缩作用在block之上,比reduce压缩方式(一条一条记录进行压缩)的压缩比高.把已有的数据转存为SequenceFile比较慢。比起先写小文件,再将小文件写入SequenceFile,一个更好的选择是直接将数据写入一个SequenceFile文件,省去小文件作为中间媒介.

此方案的代码不是很重要,所以就直接省略了,实在是想看看长啥样的可以艾特我

finally

到此HDFS的内容就差不多了,希望对你会有所帮助。之后会继续往下分享MapReduce,带你走完整个大数据的流程,感兴趣的朋友可以持续关注下,谢谢。

关注下面的标签,发现更多相似文章
评论