BIGO大数据HDFS集群慢节点的优化实践
随着HDFS集群规模的不断增长、服务器使用寿命的缩减，在大规模集群中性能退化节点的出现是必然的，我们将这样的节点称为慢节点。慢节点问题是大规模集群中的常见问题，其影响范围可以波及整个集群，会对集群的吞吐能力造成严重的影响。
HDFS作为大数据平台的存储底座，慢节点问题将对上层各个应用产生频繁的影响，可能造成数据无法按时产出，上游任务的延迟将波及下游任务，发生链式反应。Hadoop社区原生版本中对慢节点的防治有初步的实践，在此基础上我们进行了系统性的设计与开发，尽最大可能优化慢节点问题。本文主要介绍BIGO大数据存储团队在HDFS慢节点治理方面所做的主要工作和实践经验。
一、典型案例
HDFS集群部署在普通服务器之上，集群规模越大，出现的不稳定因素越多。所谓慢节点粗略来说是指读写吞吐能力较差、延迟时间较高的服务器节点。个别慢节点可能拖慢整个集群，使得集群的吞吐能力大幅度下降，进而导致关键计算任务无法按时完成，慢节点成为影响服务质量的重点问题。以下案例是一种典型的慢节点导致读数据慢的场景：
上图中截取了Spark任务读取ORC文件慢的场景，文件大小约24MB，定位到慢节点并重启慢节点HDFS DataNode进程后，Client与慢节点断开链接，Client从其他DataNode继续读数据，任务才恢复正常，Spark任务被阻塞约7个小时。收集Spark任务的strace记录，能够发现每次读数据512字节，耗时约1秒左右，24MB的文件约需要读10个小时左右。
' fill='%23FFFFFF'%3E%3Crect x='249' y='126' width='1' height='1'%3E%3C/rect%3E%3C/g%3E%3C/g%3E%3C/svg%3E)
进一步分析单次read为什么耗时这么久，最终发现问题发生在Datanode内部。FoledTreeSet是HDFS自己实现的一个数据结构，用来按顺序存放单个DN的所有数据块，当DN块个数增加到一定程度时，会存在较明显的性能瓶颈。
以上仅为长期工作中发现的一个案例。造成慢节点的原因有很多，任何一个环节出现延迟持续较高的情况，将对整体造成较大的影响，需要制定行之有效的方案来应对不同的场景，并且在长期实践过程中结合出现的新问题，不断优化和补充解决方案。
二、HDFS集群简介
在介绍慢节点的一系列优化方案之前，先介绍HDFS的整体架构，以下以社区版3.X为例，单个HDFS集群的核心组件如下图。目前较大集群规模已达千台。
Active NameNode: 提供可读可写的元数据服务。
ZooKeeperFailoverController：提供NameNode的高可用保障机制。
EditLog：每次的元数据修改都会产生一条log，保证元数据操作的原子性。
JournalNode：基于PAXOS算法实现的EditLog存储节点，保证EditLog的一致性和高可靠。
Standby NameNode：在Active NameNode故障时可以随时切为Active NameNode。
Backup NameNode：提供非一致的元数据访问服务，主要用来做元数据统计。
Observer NameNode：提供一致的元数据访问服务，可以横向扩展提升集群的读元数据处理能力。
Storage Policy Satisfier：热数据写入SSD，转温之后调整Storage Policy，自动转储到机械硬盘。
DataNode：HDFS文件以数据块为单位拆分，DN存储数据块，管理存储空间。
由于单个HDFS集群存在元数据的扩展性问题，所以在文件数量较多时需要拆分集群，一般是按照不同的业务线来划分集群，如下图所示。通过HDFS Router将不同的集群集合到统一的命名空间，Hive表多以天分区来分割数据，因为在Router层开发了元数据路由转发策略，做数据的冷热分离，热数据存放到带有SSD的集群，冷数据采用Erasure Coding RS 12+3的方式存储，数据的冗余度为125%，相比三副本300%的冗余度，在成本上可以大幅度减低。GlacialDataManager是自主开发的冷数据管理服务，主要以hive表为单位按照天分区来拷贝冷数据到冷备集群。
单个HDFS集群可以承载的元数据数量有限，因此按照不同的业务线将较大的拆分为多个集群，再由HDFS Router构建统一的命名空间。各个计算引擎已全面接入HDFS Router。同时为了节省存储成本，90天前的数据经由自研的管理工具GlacialDataManager系统转储到冷备集群采用纠删码（Erasure Coding）格式存储，约可以降低50%以上的存储成本。
三、慢节点解决方案
造成慢节点的原因有很多，常见的有慢盘、CPU过于繁忙导致I/O得不到及时处理、网卡打满、网络错误、TCP重传率高、DataNode JVM长时间频繁GC、DataNode内部的锁竞争等。造成慢盘的因素也有很多，常见的有磁盘故障、RAID卡或者HBA卡故障、磁盘压力不均衡导致的个别磁盘持续繁忙等。不太常见的慢节点问题在线上环境可能随机出现，有较大的定位难度，无法预估找到问题原因并解决慢节点的时间。下图中记录了HBA卡故障导致的所有磁盘I/O卡顿。
慢节点一旦出现，主要是影响Client和DN的数据传输效率，而且慢节点在短时间内很难恢复，所以一般的处理方式是及时发现慢节点、及时断开与慢节点的数据链接、从其他节点恢复数据传输。结合HDFS的实现机制和I/O模型，我们制定了四种慢节点的优化治理方案。
方案1：慢节点检测与自动剔除。Client与DataNode的所有交互过程中都有可能因为各种原因导致交互过程出现延迟或者阻塞，在Client侧建立慢节点模型，统计与DataNode所有的交互过程的耗时，如果触发了慢节点模型，则在Client侧主动断开与该DataNode的链接，进入错误恢复流程，与其他正常的DN建立链接，恢复数据传输。
方案2：优化DataNode内部的锁机制。将全局锁改为Volume级别的读写锁，增加并发处理能力，避免慢盘对正常磁盘的影响。
方案3：优化DataNode在写数据时的选盘策略，充分考虑负载均衡和数据分布均衡，尽可能避免部分磁盘空闲、部分磁盘持续繁忙的情况。
方案4：DataNode统计每次与其他DataNode数据传输的耗时，如果两个DataNode之间数据传输比较慢，触发了慢节点的模型，则向NameNode汇报SlowPeer；DataNode统计每次操作本地文件系统的耗时，如果触发了慢盘模型，则向NameNode汇报SlowDisk。NameNode在接收到来自DataNode的SlowPeer后，分析出哪些DataNode在多个SlowPeer中多次出现，出现次数越多的DataNode越有可能是慢节点。NameNode在接收到来自DataNode的SlowDisk汇报后，将此DataNode标记为慢节点。在NameNode分配数据块时主动排除掉慢节点，在Client查询数据块的位置时，将慢节点排在靠后的位置，Client尽可能避免与慢节点进行数据交互。
方案一：客户端慢节点检测与自动剔除
在客户端写流程中，我们加入了kickout slownode策略。这里需要考虑两个问题。一个问题是如何判断出慢节点，另一个问题时在读写流程中如何无感知踢出或替换慢节点，并且不影响读写最终的一致性。这两个问题在后续流程中会逐一解答。
在写流程中，HDFS是基于Pipline模型实现的整个写请求。每个DN节点在处理每一个packet接受和处理的时候会记录下downstream的完整时延和写入disk的完整时延，并且记录到ack中返回给客户端，同时结合HHDFS-16348中DN支持携带SLOW的flag。客户端在处理ack时，会统计对应节点的每次downstream时间和disk时间到对应的SlowNodeMetrics中。通过整体记录的统计，如果在一定时间窗口内出现了大于时间阈值慢请求超过了限制次数，则认为出现了慢节点，会将对应的DataNode标记为慢节点。当客户端发现慢节点或坏节点后，Client会进入Pipline Recovery流程，重新向Namenode申请新的Datanode，执行原Datanode到新Datanode的DataTransfer，来保证已写入的数据的持久性和一致性。接着会重建pipline，恢复写入流。
在客户端读流程中，我们同样加入了kickout slownode策略。此处与写场景不同的是，由于大部分场景我们采用三副本进行存储，某一数据块只能在三份副本保存的三个Datanode进行数据读取。首先，Client从Namenode获取blockLocation时获取到的是基于离客户端最近且最不可能是慢节点的Datanode排序（即如果Namenode记录了该DN为慢节点，则DN会被往后排）。其次Client在每次与DN建立连接，和每次对数据流触发读操作时，会统计时延到SlowNodeMetrics中。如果在一定时间窗口内，出现了超过时间阈值的读操作超过了指定的次数的情况，则认为该节点是一个慢节点。Client会重新挑选blockLocation中下一位的Datanode进行建联，并从上次读取位置继续进行数据读取。如果所有节点都尝试过了，但是都判断为慢节点，会重新选择慢节点中延迟最低的节点进行数据读取（即使此时它仍然被认为是慢节点）。
通过kickout slownode的策略，能够有效单一慢节点或慢盘对计算任务读写场景的影响。
方案二：datanode拆锁
HDFS社区版本中DN内部使用了一把全局锁，Volume之间的并发能力受到制约。
基于HDFS-15382对datanode将全局锁拆分成BlockPool + Volume为粒度的读写锁。通过这种二级锁的方式，使得datanode中不同BlockPool和Volume之间可以并发操作，同时避免了慢磁盘对其他正常磁盘的影响。
datanode进行细粒度锁改造可以在很多场景避免慢盘造成的任务延迟，但是还是有一些场景处理不了。比如，在datanode内部有很多线程任务为了获取某个磁盘的副本信息，需要对ReplicaMap进行扫描，而ReplicaMap是BlockPool级别的，需要加BlockPool锁，扫描BlockPool下所有磁盘的副本信息，导致任务效率低下，且会阻塞其他磁盘任务。
我们将ReplicaMap改造成二级map，将BlockPool map划分为Volume map，并用同步锁维护两级map的数据一致性。这样，要获取磁盘副本信息只需要扫描对应的Volume map，无需扫描整个BlockPool map。
全量块汇报优化
在HDFS早期版本中，datanode的全量块汇报（FBR）和增量块汇报（IBR）都在心跳线程上，当datanode上面的副本数比较多时，全量块汇报耗时比较久，阻塞增量块汇报，导致任务延迟。
为了减少全量块汇报对增量块汇报的影响，一个简单的方法是将增量块汇报放到一个独立线程处理，但是这样没有保证全量块汇报和增量块汇报在处理副本时的顺序性，无法保证副本的状态一致性。所以，我们将全量块汇报和增量块汇报从心跳线程独立出来在的线程上运行并且在全量块汇报期间，基于磁盘个数多线程处理全量块汇报和增量块汇报。
基于ReplicaMap的改造，在全量块汇报期间，每个磁盘线程可以快速扫描对应磁盘的副本信息，线程内维持全量块汇报和增量块汇报的顺序性，保证副本状态一致性。不同磁盘线程之间的全量块汇报间隔发送到namenode处理，减少在namenode侧排队时间。这样，一个磁盘线程在处理全量块汇报时，其他磁盘线程可以处理对应磁盘上的增量块汇报，减少增量块汇报被阻塞时间。
方案三：选盘策略优化
目前社区有两种选盘策略：RoundRobinVolumeChoosingPolicy和AvailableSpaceVolumeChoosingPolicy。由配置项dfs.datanode.fsdataset.volume.choosing.policy指定。RoundRobinVolumeChoosingPolicy策略就是用轮询方式依次选择每个volume。这个策略的优点就是实现起来比较简单，缺点每次写block的大小可能是不一样的，容易造成数据分布不均匀。
AvailableSpaceVolumeChoosingPolicy策略考虑了磁盘的剩余可用空间这个因素，在可选磁盘列表之中所有的磁盘剩余可用空间相差在配置的某个阈值（默认10GB）内时，说明此时各个磁盘之间数据量比较均匀，于是退化成RoundRobinVolumeChoosingPolicy。如果相差大于配置的阈值时，则按照一定的概率(实际中一般是大于0.5，小于1)，优先选择剩余可用空间大的volume。这个策略的优点就是会让磁盘数据分布比较均衡，缺点就是可能写请求会一直发到剩余容量较大的磁盘，容易造成写入热点，最终成为慢节点。基于两种策略的缺点，我们实现了基于ioutil的选盘策略，此策略同时兼顾磁盘负载以及剩余空间等因素。
以下图为例：一个写请求过来，这是一个写storage type是DISK的请求，所以我们会排除掉非DISK类型的磁盘，比如下图中的/dev/sdy。在DISK类型的磁盘中，根据我们的选盘策略最终选出的磁盘将会是/dev/sdc。因为他的ioutil是最小的80，同时剩余可用空间又是最大的。
下图是上线新的选盘策略后，磁盘ioutil的变化情况，可以看到，磁盘ioutil 100%的情况大幅减少。
接下来介绍基于磁盘ioutil的选盘策略的实现思路。核心要点是怎么计算得到磁盘的ioutil，这个计算方法是参考了linux下iostat这个命令计算ioutil的实现。/proc/diskstats这个文件的第十三列是磁盘花在I/O上的时间，单位为毫秒。那假设我们每隔1s中读取/proc/diskstats文件的第13列的值，然后使用如下公式计算就可以得到ioutil了：io util(%) = (t2 - t1) * 100 / 1000。
在DataNode启动的时候开启一个后台线程，每隔1秒进行一次磁盘ioutil的计算，保存在一个Map类型的数据结构中。然后在选盘策略的doChooseVolume方法中使用磁盘的ioutil进行比较，选出ioutil较小的磁盘进行写入数据操作，如果两个磁盘ioutil相等，那就选择剩余可用空间大的磁盘进行写入。
方案四：客户端慢节点收集与自动剔除
HDFS集群内部对慢节点的识别和剔除，需要DN和NN的共同协作才能完成。DN根据数据传输耗时，将自己认为较慢的节点通过SlowPeerReport的方式汇报到NameNode，Namenode汇总所有DN汇报的慢节点信息，进行聚合，选出最慢的节点，并将这些节点的UUID保存在名为SlowNodesUUIDSet的Set集合。在读和写流程中，可以利用该集合完成对慢节点的剔除。
1. DN收集监控数据
HDFS在写数据时，采用的是Pipline模型。DN选用Pipline的倒数第二个节点作为基准节点，统计其发送packet到最后一个节点的耗时，以此作为判断慢节点的依据。只采用倒数第二个节点进行统计，可以降低统计的复杂度。DN内部为慢节点统计维护了一个Map<String, LinkedBlockingDeque>，key是下游DN的ip，value是以SumAndCount为元素的队列。DN每次向下游DN发送packet时，会将发送的packet数量和耗时记录下来，用SampleStat保存。然后每5min对采集的数据做聚合，保存到SumAndCount，该对象记录了最近5min内发送的packet个数和总耗时。最后，将SumAndCount存放到LinkedBlockingDeque。LinkedBlockingDeque默认长度是36，如果队列满了，踢掉队首元素，在队尾追加。因此，该模型统计的时间窗口为最近3小时，这一方面可以保证数据的时效性，又可以保证数据的准确性。
2. DN汇报慢节点
DN在心跳汇报时，会将根据最近3小时数据，计算与每个下游节点传输packet的平均耗时，并计算对应的中位数和平均绝对偏差（MAD）。采用median + 3*MAD作为阈值，如果和某个下游DN传输packt的平均耗时超过了该阈值，则认为是慢节点。那么，就把改下游DN汇报给Namenode。
3. NameNode汇总慢节点集合
Namenode将所有DN的SlowPeerReport保存在ConcurrentMap<String, ConcurrentMap>中，命名为allReports。allReports的key是慢节点的ip，value是所有ReportNodes组成的一个集合。每个慢节点的ReportNodes也保存在ConcurrentMap中，其中key为ReportNode的ip，value是汇报时间。
我们在Namenode中启动一个SlowPeerCollector线程，定期将allReports里面的慢节点根据reportNodes的数量进行排序，根据配置取出TopN的节点，组成集群中最慢的节点集合slowNodesUuidSet。在读写流程中，将可以使用slowNodesUuidSet对慢节做相应处理。Namenode根据reportNodes数量排序，认为被汇报最多的节点是最慢的节点，这保证了慢节点识别的准确性。
4. 写流程自动剔除慢节点
HDFS在写数据时，会调用BlockPlacementPolicy的chooseTarget方法，为block的副本选择合适的DN节点。我们会在chooseTarget方法中，根据SlowPeerCollector收集的slowNodesUuidSet，将慢节点剔除掉。这样，chooseTarget将返回不包含慢节点的DN集合，给客户端写数据，以此优化写数据性能。
5. 读流程自动剔除慢节点
HDFS在读数据时，会调用FSNamesystem的getBlockLocations方法，选择读数据的DN列表。然后默认选择列表中的第一个DN读数据，如果改DN读数据失败则选取下一个DN，以此类推。在getBlockLocations方法中，有对DN列表进行排序的功能，会将非正常节点放在正常节点之后。我们修改了Comparator的对比规则，如果要读取的block所在的DN在slowNodesUuidSet中，则将该节点放在正常节点之后。这样，可以帮助客户端尽可能避开从慢节点读数据，以此提升读数据的性能。
四、总结
以上功能已经全部上线到生产环境，在长期运行中，根据用户反馈，优化效果比较明显，慢节点导致的任务慢的问题大幅度减少。以天级任务为例，下图中红框内的表示任务结束时间，在优化前任务运行总时间较长且不稳定，经常无法在预期时间内完成。在慢节点优化方案上线后，基本可以在预期时间完成。
以小时级任务为例，下图中的任务优化前的运行时间约几个小时，优化后基本上可以在半个小时内完成。
▼版权声明