Hadoop优雅架构原理剖析

Posted 2021-04-13 程序猿的修身养性

1、Hadoop高可用架构介绍

Hadoop是目前大数据领域最主流的一套技术体系，包含了多种技术，

包括HDFS（分布式文件系统），YARN（分布式资源调度系统），MapReduce（分布式计算系统）等等。

有些朋友可能听说过Hadoop，但是却不太清楚它到底是个什么东西，这篇文章就用大白话给各位阐述一下。假如你现在公司里的数据都放在mysql里，那么就全部放在一台数据库服务器上，我们就假设这台服务器的磁盘空间有2T，如下图。

现在问题来了，不停的往这台服务器的MySQL里放数据，结果数据量越来越大，超过了2T的大小，现在咋办？你可能会说，我可以搞多台MySQL数据库服务器，分库分表啊！每台服务器放一部分数据不就得了。如上图所示！

OK，那咱们搞3台数据库服务器，3个MySQL实例，然后每台服务器都可以存2T的数据。现在问你一个问题，所谓的大数据是在干什么？

说一下大数据最初级的一个使用场景。假设你有一个电商网站，现在要把这个电商网站里所有的用户在页面和APP上的点击、购买、浏览的行为日志都存放起来进行分析。现在把这些数据全都放在了3台MySQL服务器，数据量很大，但还是勉强可以放的下。

某天早上，boss来了，要看一张报表，比如要看每天网站的X指标、Y指标、Z指标等等，二三十个数据指标。好了，现在你尝试去从那些点击、购买、浏览的日志里，通过写一个SQL来分析出那二三十个指标试试看？绝对会写出来一个几百行起步，甚至上千行的超级复杂大SQL，这个SQL，你觉得它能运行在分库分表后的3台MySQL服务器上么？

如果觉得可以，那你一定是不太了解MySQL分库分表后有多坑，几百行的大SQL跨库join，各种复杂的计算，根本不现实。所以说，大数据的存储和计算压根儿不是靠MySQL来搞的，因此，Hadoop、Spark等大数据技术体系才应运而生。本质上，Hadoop、Spark等大数据技术，其实就是一系列的分布式系统。

比如hadoop中的HDFS，就是大数据技术体系中的核心基石，负责分布式存储数据，这是啥意思？别急，继续往下看。HDFS全称是Hadoop Distributed File System，是Hadoop的分布式文件系统。它由很多机器组成，每台机器上运行一个DataNode进程，负责管理一部分数据。然后有一台机器上运行了NameNode进程，NameNode大致可以认为是负责管理整个HDFS集群的这么一个进程，它里面存储了HDFS集群的所有元数据。然后有很多台机器，每台机器存储一部分数据！好，HDFS现在可以很好的存储和管理大量的数据了。

这时候你肯定会有疑问：MySQL服务器也不是这样的吗？你要是这样想，那就大错特错了。这个事情不是你想的那么简单的，HDFS天然就是分布式的技术，所以你上传大量数据，存储数据，管理数据，天然就可以用HDFS来做。如果你硬要基于MySQL分库分表这个事儿，会痛苦很多倍，因为MySQL并不是设计为分布式系统架构的，他在分布式数据存储这块缺乏很多数据保障的机制。

好，现在用HDFS分布式文件系统存储了数据，接着不就是要分布式来计算这些数据了吗？对于分布式计算：很多公司用Hive写几百行的大SQL（底层基于MapReduce），也有很多公司开始慢慢的用Spark写几百行的大SQL（底层是Spark Core引擎）。总之就是写一个大SQL，人家会拆分为很多的计算任务，放到各个机器上去，每个计算任务就负责计算一小部分数据，这就是所谓的分布式计算。

这个，绝对比你针对分库分表的MySQL来跑几百行大SQL要靠谱的多。对于上述所说的分布式存储与分布式计算，如下图。

 

HDFS的NameNode架构原理

好了，前奏铺垫完之后，进入正题。本小节其实主要就是讨论一下HDFS集群中的NameNode的核心架构原理。NameNode有一个很核心的功能：管理整个HDFS集群的元数据，比如说文件目录树、权限的设置、副本数的设置，等等。下面就用最典型的文件目录树的维护，来举例说明，如下图。现在有一个客户端系统要上传一个1TB的大文件到HDFS集群里。

此时它会先跟NameNode通信，说：大哥，我想创建一个新的文件，它的名字叫“/usr/hive/warehouse/access_20180101.log”，大小是1TB，你看行不？然后NameNode就会在自己内存的文件目录树里，在指定的目录下搞一个新的文件对象，名字就是“access_20180101.log”。这个文件目录树不就是HDFS非常核心的一块元数据，维护了HDFS这个分布式文件系统中，有哪些目录，有哪些文件，对不对？

但是有个问题，这个文件目录树是在NameNode的内存里的啊！这可坑爹了，你把重要的元数据都放在内存里，万一NameNode不小心宕机了可咋整？元数据不就全部丢失了？可你要是每次都频繁的修改磁盘文件里的元数据，性能肯定是极低的啊！毕竟这是大量的磁盘随机读写！没关系，我们来看看HDFS优雅的解决方案。

每次内存里改完了，写一条edits log，即元数据修改的操作日志到磁盘文件里，不修改磁盘文件内容，就是顺序追加，这个性能就高多了。每次NameNode重启的时候，把edits log里的操作日志读到内存里回放一下，不就可以恢复元数据了？如下图。

但是问题又来了，那edits log如果越来越大的话，岂不是每次重启都会很慢？因为要读取大量的edits log进行回放以恢复元数据！所以HDFS说，我可以这样子啊，我引入一个新的磁盘文件叫做fsimage，然后呢，再引入一个JournalNodes集群，以及一个Standby NameNode（备节点）。每次Active NameNode（主节点）修改一次元数据都会生成一条edits log，除了写入本地磁盘文件，还会写入JournalNodes集群。

然后Standby NameNode就可以从JournalNodes集群拉取edits log，应用到自己内存的文件目录树里，跟Active NameNode保持一致。然后每隔一段时间，Standby NameNode都把自己内存里的文件目录树写一份到磁盘上的fsimage，这可不是日志，这是完整的一份元数据。这个操作就是所谓的checkpoint检查点操作。然后把这个fsimage上传到Active NameNode，接着清空掉Active NameNode中旧的edits log文件，这里可能都有100万行修改日志了！然后Active NameNode继续接收修改元数据的请求，再写入edits log，写了一小会儿，这里可能就几十行修改日志而已！

如果说此时，Active NameNode重启了，bingo！没关系，只要把Standby NameNode传过来的fsimage直接读到内存里，这个fsimage直接就是元数据，不需要做任何额外操作，纯读取，效率很高！然后把新的edits log里少量的几十行的修改日志回放到内存里就ok了！这个过程的启动速度就快得多了！因为不需要回放大量（上百万行）的edits log来恢复元数据！如下图。

此外，在上图中有俩NameNode，一个是主节点对外提供服务接收请求；另外一个纯就是接收和同步主节点的edits log以及执行定期checkpoint的备节点。发现没有！它们俩内存里的元数据几乎是一模一样的！所以呢，如果Active NameNode挂了，是不是可以立马切换成Standby NameNode对外提供服务？这不就是所谓的NameNode主备高可用故障转移机制么！

接下来再想想，HDFS客户端在NameNode内存里的文件目录树中，新加了一个文件。但是这个时候，人家要把数据上传到多台DataNode机器上去啊，这可是一个1TB的大文件！咋传呢？很简单，把1TB的大文件拆成N个block，每个block是128MB。1TB = 1024GB = 1048576MB，一个block是128MB，那么就是对应着8192个block。这些block会分布在不同的机器上管理着，比如说一共有100台机器组成的集群，那么每台机器上放80个左右的block就ok了。

但是问题又来了，那如果这个时候1台机器宕机了，不就导致80个block丢失了？也就是说上传上去的1TB的大文件，会丢失一小部分数据啊。没关系！HDFS都考虑好了！它会默认给每个block搞3个副本，一模一样的副本，分放在不同的机器上，如果一台机器宕机了，同一个block还有另外两个副本在其他机器上呢！如下图。每个block都在不同的机器上有3个副本，任何一台机器宕机都没事！还可以从其他的机器上拿到那个block。

这下子，你往HDFS上传一个1TB的大文件，可以高枕无忧了！

2、Hadoop的NameNode如何承载高并发

先来分析一下，高并发请求下的NameNode会遇到什么样的问题。我们知道，每次请求NameNode修改一条元数据（比如说申请上传一个文件，那么就需要在内存目录树中加入一个文件），都要写一条edits log，包括两个步骤：写入本地磁盘；通过网络传输给JournalNodes集群。但是如果对Java有一定了解的朋友都该知道多线程并发安全的问题。

NameNode在写edits log时的第一条原则：必须保证每条edits log都有一个全局顺序递增的transactionId（简称为txid），这样才可以标识出来一条一条的edits log的先后顺序。那么如果要保证每条edits log的txid都是递增的，就必须得加锁。每个线程修改了元数据，要写一条edits log的时候，都必须按顺序排队获取锁后，才能生成一个递增的txid，代表这次要写的edits log的序号。

好的，那么问题来了，如下图。

  如果每次都是在一个加锁的代码块里，生成txid，然后写磁盘文件edits log，网络请求写入journalnodes集群一条edits log，会咋样？

不用说，这个绝对完蛋了！NameNode本身用多线程接收多个客户端发送过来的并发的请求，结果多个线程居然修改完内存中的元数据之后，还要排着队写edits log！而且我们知道，写本地磁盘 + 网络传输给journalnodes集群，都是很耗时的啊！性能两大杀手：磁盘写 + 网络写！

如果HDFS的架构真要是这么设计的话，基本上NameNode能承载的每秒的并发数量就很少了，可能就每秒处理几十个并发请求处理撑死了！

HDFS优雅的解决方案

首先想一下，既然咱们不希望每个线程写edits log的时候，串行化排队生成txid + 写磁盘 + 写JournalNode，那么是不是可以搞一个内存缓冲？也就是说，多个线程可以快速的获取锁，生成txid，然后快速的将edits log写入内存缓冲。接着就快速的释放锁，让下一个线程继续获取锁后，生成txid + 写edits log进入内存缓冲。

然后接下来有一个线程可以将内存中的edits log刷入磁盘，但是在这个过程中，还是继续允许其他线程将edits log写入内存缓冲中。但是这里又有一个问题了，如果针对同一块内存缓冲，同时允许有人写入，还同时允许有人读取后写磁盘，那也有问题，因为不能并发读写一块共享内存数据！所以HDFS在这里采取了double-buffer双缓冲机制来处理！将一块内存缓冲分成两个部分：其中一个部分可以写入；另外一个部分用于读取后写入磁盘和JournalNodes集群。如下图。

分段加锁机制+内存双缓冲机制

首先各个线程依次第一次获取锁，生成顺序递增的txid，然后将edits log写入内存双缓冲的区域1，接着就立马第一次释放锁了。趁着这个空隙，后面的线程就可以再次立马第一次获取锁，然后立即写自己的edits log到内存缓冲。写内存那么快，可能才耗时几十微妙，接着就立马第一次释放锁了。所以这个并发优化绝对是有效果的，有没有感受到？

接着各个线程竞争第二次获取锁，有线程获取到锁之后，就看看，有没有谁在写磁盘和网络？如果没有，好，那么这个线程是个幸运儿！直接交换双缓冲的区域1和区域2，接着第二次释放锁。这个过程相当快速，内存里判断几个条件，耗时不了几微秒。好，到这一步为止，内存缓冲已经被交换了，后面的线程可以立马快速的依次获取锁，然后将edits log写入内存缓冲的区域2，区域1中的数据被锁定了，不能写。

多线程并发吞吐量的百倍优化

接着，之前那个幸运儿线程，将内存缓冲的区域1中的数据读取出来（此时没人写区域1了，都在写区域2），将里面的edtis log都写入磁盘文件，以及通过网络写入JournalNodes集群。这个过程可是很耗时的！但是没关系啊，人家做过优化了，在写磁盘和网络的过程中，是不持有锁的！因此后面的线程可以噼里啪啦的快速的第一次获取锁后，立马写入内存缓冲的区域2，然后释放锁。这个时候大量的线程都可以快速的写入内存，没有阻塞和卡顿！

缓冲数据批量刷磁盘+网络的优化

那么在幸运儿线程吭哧吭哧把数据写磁盘和网络的过程中，排在后面的大量线程，快速的第一次获取锁，写内存缓冲区域2，释放锁，之后，这些线程第二次获取到锁后会干嘛？他们会发现有人在写磁盘啊，兄弟们！所以会立即休眠1秒，释放锁。此时大量的线程并发过来的话，都会在这里快速的第二次获取锁，然后发现有人在写磁盘和网络，快速的释放锁，休眠。怎么样，这个过程没有人长时间的阻塞其他人吧！因为都会快速的释放锁，所以后面的线程还是可以迅速的第一次获取锁后写内存缓冲！而且这时，一定会有很多线程发现，好像之前那个幸运儿线程的txid是排在自己之后的，那么肯定就把自己的edits log从缓冲里写入磁盘和网络了。这些线程甚至都不会休眠等待，直接就会返回后去干别的事情了，压根儿不会卡在这里。然后那个幸运儿线程写完磁盘和网络之后，就会唤醒之前休眠的那些线程。那些线程会依次排队再第二次获取锁后进入判断，咦！发现没有人在写磁盘和网络了！然后就会再判断，有没有排在自己之后的线程已经将自己的edtis log写入磁盘和网络了。如果有的话，就直接返回了。没有的话，那么就成为第二个幸运儿线程，交换两块缓冲区，区域1和区域2交换一下，然后释放锁，自己开始吭哧吭哧的将区域2的数据写入磁盘和网络。但是这个时候没有关系，后面的线程如果要写edits log，还是可以第一次获取锁后立马写内存缓冲再释放锁。以此类推。

总结

这套机制还是挺复杂的，涉及到了分段加锁以及内存双缓冲两个机制。通过这套机制，NameNode保证了多个线程在高并发的修改元数据之后写edits log的时候，不会说一个线程一个线程的写磁盘和网络，那样性能实在太差，并发能力太弱了！所以通过上述那套复杂的机制，尽最大的努力保证，一个线程可以批量的将一个缓冲中的多条edits log刷入磁盘和网络。在这个漫长的吭哧吭哧的过程中，其他的线程可以快速的高并发写入edits log到内存缓冲里，不会阻塞其他的线程写edits log。所以，正是依靠以上机制，最大限度优化了NameNode处理高并发访问修改元数据的能力！

3、HDFS如何提高文件上传效率

首先，通过下图来回顾一下文件上传的大概原理。

  由上图所示，文件上传的原理，其实说出来也简单。比如有个TB级的大文件，太大了，HDFS客户端会给拆成很多block，一个block就是128MB。这个HDFS客户端你可以理解为是云盘系统、日志采集系统之类的东西。比如有人上传一个1TB的大文件到网盘，或者是上传个1TB的大日志文件。然后，HDFS客户端把一个一个的block上传到第一个DataNode，第一个DataNode会把这个block复制一份，做一个副本发送给第二个DataNode。第二个DataNode发送一个block副本到第三个DataNode。所以你会发现，一个block有3个副本，分布在三台机器上。任何一台机器宕机，数据是不会丢失的。最后，一个TB级大文件就被拆散成了N多个MB级的小文件存放在很多台机器上，这不就是分布式存储么？

原始的文件上传方案

这里要讨论的问题，就是那个HDFS客户端上传TB级大文件的时候，到底是怎么上传的？先来考虑一下，如果用一个比较原始的方式来上传，应该怎么做？大概能想到的是如下图的样子。

 

很多Java的初学者，估计都知道这样来上传文件，其实无非就是不停的从本地磁盘文件用输入流读取数据，读到一点，就立马通过网络的输出流写到DataNode里去。上面这种流程图的代码，估计刚毕业的同学都可以立马写出来。因为对文件的输入流最多就是个FileInputStream。而对DataNode的输出流，最多就是个Socket返回的OutputStream。然后中间找一个小的内存byte[]数组，进行流对拷就行了，从本地文件读一点数据，就给DataNode发一点数据。

但是如果要这么弄，那性能真是极其的低下，网络通信讲究的是适当频率，每次batch批量发送，你得读一大批数据，通过网络通信发一批数据。不能说读一点点数据，就立马来一次网络通信，就发出去这一点点的数据。如果按照上面这种原始的方式，绝对会导致网络通信效率极其低下，大文件上传性能很差。为什么这么说呢？

相当于你可能刚读出来几百个字节的数据，立马就写网络，卡顿个比如几百毫秒。然后再读下一批几百个字节的数据，再写网络卡顿个几百毫秒，这个性能很差，在工业级的大规模分布式系统中，是无法容忍的。

HDFS对大文件上传的性能优化

讲完了原始的文件上传，我们来看看，Hadoop中的大文件上传是如何优化性能的？如下图。

首先需要自己创建一个针对本地TB级磁盘文件的输入流。然后读到数据之后立马写入HDFS提供的FSDataOutputStream输出流。这个FSDataOutputStream输出流在干啥？大家觉得他会天真的立马把数据通过网络传输写给DataNode吗？答案当然是否定的！这么干的话，不就跟之前的那种方式一样了！

Chunk缓冲机制

首先，数据会被写入一个chunk缓冲数组，这个chunk是一个512字节大小的数据片段，可以先这么来理解。然后这个缓冲数组可以容纳多个chunk大小的数据在里面进行缓冲。光是这个缓冲，首先就可以让客户端快速的写入数据了，不至于说几百字节就要进行一次网络传输，想一想，是不是这样？

Packet数据包机制

接着，当chunk缓冲数组都写满了之后，就会把这个chunk缓冲数组进行一下chunk切割，切割为一个一个的chunk，一个chunk是一个数据片段。然后多个chunk会直接一次性写入另外一个内存缓冲数据结构，就是Packet数据包。一个Packet数据包，设计为可以容纳127个chunk，大小大致为64mb。所以说大量的chunk会不断的写入Packet数据包的内存缓冲中。通过这个Packet数据包机制的设计，又可以在内存中容纳大量的数据，进一步避免了频繁的网络传输影响性能。

内存队列异步发送机制

当一个Packet被塞满了chunk之后，就会将这个Packet放入一个内存队列来进行排队。然后有一个DataStreamer线程会不断的获取队列中的Packet数据包，通过网络传输直接写一个Packet数据包给DataNode。如果一个Block默认是128mb的话，那么一个Block默认会对应两个Packet数据包，每个Packet数据包是64MB。也就是说，传送两个Packet数据包给DataNode之后，就会发一个通知说，一个Block的数据都传输完毕。这样DataNode就知道自己收到一个Block了，里面包含了发送过来的两个Packet数据包。

总结

看完上面的那个图以及Hadoop采取的大文件上传机制，是不是感觉设计的很巧妙？说白了，工业级的大规模分布式系统，都不会采取特别简单的代码和模式，那样性能很低下。这里都有大量的并发优化、网络IO优化、内存优化、磁盘读写优化的架构设计、生产方案在里面。所以大家观察上面那个图，HDFS客户端可以快速的将TB级大文件的数据读出来，然后快速的交给HDFS的输出流写入内存。基于内存里的chunk缓冲机制、packet数据包机制、内存队列异步发送机制，绝对不会有任何网络传输的卡顿，导致大文件的上传速度变慢。反而通过上述几种机制，可以上百倍的提升一个TB级大文件的上传性能。

4、Hadoop如何提升大规模集群的性能

先引入一个小的背景，假如多个客户端同时要并发的写Hadoop HDFS上的一个文件，大家觉得这个事儿能成吗？明显是不可以接受的，HDFS上的文件是不允许并发写的，比如并发的追加一些数据什么的。所以说，HDFS里有一个机制，叫做文件契约机制。

也就是说，同一时间只能有一个客户端获取NameNode上面一个文件的契约，然后才可以写入数据。此时如果其他客户端尝试获取文件契约，就获取不到，只能干等着。通过这个机制，就可以保证同一时间只有一个客户端在写一个文件。在获取到了文件契约之后，在写文件的过程期间，那个客户端需要开启一个线程，不停的发送请求给NameNode进行文件续约，告诉NameNode：NameNode大哥，我还在写文件啊，你给我一直保留那个契约好吗？而NameNode内部有一个专门的后台线程，负责监控各个契约的续约时间。如果某个契约很长时间没续约了，此时就自动过期掉这个契约，让别的客户端来写。如下图。

 

那么现在问题来了，假如我们有一个大规模部署的Hadoop集群，同时存在的客户端可能多达成千上万个。此时NameNode内部维护的那个文件契约列表会非常非常的大，而监控契约的后台线程又需要频繁的每隔一段时间就检查一下所有的契约是否过期。比如，每隔几秒钟就遍历大量的契约，那么势必造成性能不佳，所以说这种契约监控机制明显是不适合大规模部署的hadoop集群的。

Hadoop的优化方案

那么Hadoop是如何对文件契约监控算法进行优化的呢？如下图。

其实奥秘十分的简单，每次一个客户端发送续约请求之后，就设置这个契约的最近一次续约时间。然后，基于一个TreeSet数据结构来根据最近一次续约时间对契约进行排序，每次都把续约时间最老的契约排在最前头，这个排序后的契约数据结构十分的重要。TreeSet是一种可排序的数据结构，他底层基于TreeMap来实现。TreeMap底层则基于红黑树来实现，可以保证元素没有重复，同时还能按照我们自己定义的排序规则在每次插入一个元素的时候来进行自定义的排序。

所以这里的排序规则：就是按照契约的最近一次续约时间来排序。其实这个优化就是如此的简单，就是维护这么一个排序数据结构而已。现在来看一下Hadoop中的契约监控的源码实现：

每次检查契约是否过期的时候，不要遍历成千上万的契约，那样遍历效率当然会很低下。完全可以就从TreeSet中获取续约时间最老的那个契约，假如说连最近一次续约时间最老的那个契约都还没过期，那么就不用继续检查了！这说明续约时间更近的那些契约绝对不会过期！

举个例子：续约时间最老的那个契约，最近一次续约的时间是10分钟以前，但是我们判断契约过期的限制是超过15分钟不续约就过期那个契约。这个时候，连10分钟以前续约的契约都没有过期，那么那些8分钟以前，5分钟以前续约的契约，肯定也不会过期！

这个机制的优化对性能的提升是相当有帮助的，因为正常来说，过期的契约肯定还是占少数，所以压根儿不用每次都遍历所有的契约来检查是否过期。只需要检查续约时间最旧的那几个契约就可以了，如果一个契约过期了，那么就删掉那个契约，然后再检查第二旧的契约好了。以此类推。

通过这个TreeSet排序+优先检查最旧契约的机制，有效的将大规模集群下的契约监控机制的性能提升至少10倍以上，这种思想是非常值得我们学习和借鉴的。优秀的开源项目，蕴含着很多优秀的设计思想。多看各种优秀开源项目的源码，是短时间内快速、大幅度提升一个人的技术功底和技术水平的方式，不妨尝试一下！

文章内容摘自：石杉的架构笔记

《兄弟，用大白话告诉你小白都能看懂的Hadoop架构原理》

《大规模集群下Hadoop NameNode如何承载每秒上千次的高并发访问》

《【性能优化的秘密】Hadoop如何将TB级大文件的上传性能优化上百倍？》

《【眼前一亮！】看Hadoop底层算法如何优雅的将大规模集群性能提升10倍以上？》