HDFS特性之RaidNode与EC
众所周知, RAID(磁盘阵列)是广泛运用在服务器端的一种提升存储性能的技术, 在HDFS中的高版本(HDFS3)中也有类似的重要特性EC(纠错码技术), 但是高版的EC是从内部设计的, 对HDFS的存储结构, 读写影响都很大, 无法兼容旧数据, 所以日后再单独体系的讲
先来看看早在HDFS低版本(0.2)时期已经出现了外挂RAID降低存储成本的方案: 也就是今天的主人公”RaidNode”, 可以说它是EC的前身.
0x00. 背景
HDFS中有个核心的点就是它依靠默认的3副本来保证数据可靠性, 换言之就是说在HDFS上存储1份数据要占3份的实际空间, 这对磁盘存储的消耗是很大的, 特别是放在SSD这种比较昂贵的存储介质上. 那么很自然的, 大家想到将磁盘RAID的技术运用到HDFS上, 以达到降低实际存储占用的效果, 如下图所示:

最早在2009年, Facebook就开源了自己的实现, 并很快贡献给了HDFS社区引入了HDFS-0.21的版本中(后续被移除), 一般我们现在提到的各种版本RaidNode都是基于Facebook这一版做了改动而来的, 所以主要是学习它的设计和理念. (其中EC相关的概念可参考此篇或官方的这篇文章, 或者国内版)
那么Facebook-Raidnode的工作原理是什么呢? 简单说它有两种校验模式, 一种是XOR(类似RAID5), 另一种是RS(里所码), 一般采用的是RS模式, 二者对比图如下: (参考官方blog)

RS模式, 是指通过对N个数据块进行运算. 算出M个校验块, 最多可以容忍M个块的丢失. 丢失的块可通过现存的N个块重新算出恢复, 这种方式存储成本最低, 但计算和恢复成本较大.
然后我们遵循Facebook官方的定义, 把数据块称作”source data“, 把校验块又称作”parity(奇偶) data“, 采用它推荐的“10+4” 的组合, 参见官方图:

它采用的RS(Reed-Solomon)-EC编码方式原理可以先放一下, 最直观的理解就是, 以前10个block, 我需要耗费3倍的存储空间, 而现在我只需要1.4倍的空间, 可以说节省了100%的存储占用, 而且这总共14个block中, 可以容忍最多4个块的丢失, 还能从剩余的块把它们计算恢复, 优点是显而易见的.
听起来很神奇, 但随之就有一系列的疑问, 比如:
- 这是如何做到的, 使用场景是什么?
- 原有的块副本去哪了? 是直接把副本数设置为1了么?
- Raidnode又是如何使用外挂的结构呢?
- 使用Raidnode方案, 有什么代价, 目前又有哪些不足?
下面会一一说明, 先来看看Raidnode的整体架构.
PS: Raidnode大部分参考资料质量都不太好, 官方Wiki也被移除了, 已有的里面推荐两个 (若访问失败可以通过Google快照)
0x01. 整体架构
首先, Raidnode整体也是C/S结构的设计, 有一个Server也就是Raidnode进程, 然后可由Client端发起请求触发操作, server也定期自动触发操作, 如下所示:
- Server端
- RaidNode
- 块替换策略
- Client端
- DistributedRaidFileSystem
- Raid Shell
整体如下图所示:
不过这里需要注意的是, 实际生产环境里, 为了简化RaidNode使用, 可能会弃用它的Client, 只需要RaidNode自行定期检查–>工作.
RaidNode涉及与NameNode、DataNode、Corona/Yarn集群通信
- 其中与NameNode通信的主要目的是获取需要归档的文件、获取待修复文件、获取降幅本文件以及删除过期的校验数据等
- 与DataNode通信是为了避免多个raid相关的block同时落在同一个DataNode中 (这个是直接发RPC给DN?)
- 与Corona/Yarn集群通信则是为了提交任务进行生成校验块, 修复丢失块,归档等 (降副本实际是NN做的)
0x02. 具体结构
A. RaidNode
1. 功能
那么可以推测出, 之所以低版本的Raid实现是比较独立的, 就是因为主要的组件是单独的一个进程, 那么先来看看核心进程RaidNode做哪些事:
- 定时读取配置文件, 得知哪些目录需要被Raid
- 定时的与NN通信, 对配置Raid的目录扫描, 然后主要做以下3个事:
- 生成校验文件 (MR)
- 降低数据文件副本
- 定时清理过期的校验文件 (比如数据文件被删后)
- 提交归档(合并成HAR)任务, 以减少校验文件的INode占用
2. 模块
RaidNode主要由几个重要线程类组成:
- ConfigManager:定时加载由
raid.config.file
配置 - TriggerMonitor(主要):定期检查配置的raid目录,然后提交MR作业, 是主要模块
- BlockIntegrityMonitor:定时检查raid后文件的完整性,有异常则提交修块的MR任务
- PurgeMonitor:定时检查是否有源文件NN中被删除了, 有则将其对应校验文件删除
- PlacementMonitor: 定时检查, 确保raid后的块尽可能分散在不同的DataNode上 (DN间移动block)
- HarMonitor: 定时扫描校验文件, 提交归档任务以节省NN的INode占用.
其中1~6个组件都对应RaidNode中的不同后台线程, 在RaidNode进程启动后就会依次初始化它们, 以实现在定期检测工作的效果.
PS: 除此之外, 还有StatsCollectorThread和HTTP Server模块, 但是并非重点, 暂时略过, 同时原版中3的名字似乎是BlockFixer
, 后续调整了, 说明一下
B. JobTracker
这里说的JobTracker
不是一个Raidnode中具体的类, 它是往调度平台提交Map-Reduce任务的任务调度器, 简单理解即可, 我们关注以下几个问题:
- 为什么任务要以MR方式提交?
- 如何提交并执行的? (关键)
- MR任务执行成功/失败后续?
C. DRFS & RaidShell
DRFS和RaidShell的关系差不多就跟我们之前DFS和FShell关系一样, Shell提供了一层封装方便使用, 本质是一体的, 但是因为这部分不一定必要, 所以生产环境中你可能会发现Raidnode-Client部分被弃用了, 也没多少影响, 就是说不能用户自己去提交命令给Raidnode
指定修复某个块了.
1. DRFS
DRFS(DistributedRaidFileSystem)其实就是在客户端Raid的具体FileSystem实现, 它主要功能是:
- 允许客户端读”坏的数据块”
- 可捕获
BlockMissing
和ChecksumError
的异常 - 可重新生成缺失的块
- 可捕获
- 但读数据时不会立刻去修复丢失的块, 仅仅是让读操作成功完成. (TODO)
2. RaidShell
一个快捷的命令行工具, 主要功能是:
- 发送块恢复指令
- 重建丢失的块
- 将重建的块发给Datanode
- Raid的FSCK命令
- 管理员相关命令
0x03. 部署使用
了解了前面的大体铺垫后, 先来具体用一下, 看看它到底是怎么运作的, 熟悉环境 (记得提前准备好HDFS+Yarn的环境, 最好是3个至少节点)
1. 配置相关
首先是配置相关, 它确定了当前的RAID策略, 以及具体的校验文件存放等, 先在hdfs-site.xml
添加配置项(其他可参考)
1 | <configuration> |
然后就是配置raid.xml
, 举个具体例子 (多个目录需要配多个policy项)
1 | <configuration> |
可以看到它就是配置了几个关键的参数项, 注释也写的很清楚, 从而确定了Raidnode运行的具体模式
2. 启动运行
这里有两种方式可以启动, 一种是传统(原版0.20)方式通过自带的脚本, 然后你可以用raid-shell去操作, 还有一种是升级后高版中去掉了raid-shell, 直接通过hadoop jar raid.jar xxx.RaidNode
的方式启动, 都说一下.
A. 使用默认脚本启动
关键的当然就是启动/停止RaidNode进程, 但是要注意Raidnode可能还会间接启动一个单独的Archive
进程
1 | # hadoop根目录下, 注意此脚本仅限原版0.2X才有 |
B. 使用直接提交Jar包方式启动
这个方式适用于改进版本, 以及高版适配后不能使用脚本启动的方式,
1 | # 先自己编译raidnode修改后的项目打个jar包 |
C. 检验结果
那如何判raidnode运行正常了呢? 有Client端/Raidshell, 最简单的方法就是用raidshell
测试一下, 此外还可以有两点通用的检测方式:
观察你在hdfs-site中配置的
hdfs.raidrs.locations
对应的路径是否有生成文件1
2
3
4
5
6
7
8
9
10
11# 这里具体的raid路径,和raid后路径以实际为准
bin/hadoop fs -ls -R /archive* /tmp /raid
# 正在raid时候, 应该会在/tmp下生成临时校验文件, 完成后移动到/archive-raid中
# 假设我们/archive/raid目录下有两个大文件1.big和2.big, 那么执行raid时
/tmp/raidrs/1.big-3308087870674951347
/tmp/raidrs/2.big8835042559380365685
# 执行完后, 应该存在
/archive-raid/meta/rs/archive/raid/1.big
/archive-raid/meta/rs/archive/raid/2.big你配置需要降副本的目录, fsck查看副本数是否降为1了. 当然其实高版本
HDFS-shell
使用ls
时就会在第二列显示副本数, 更加直观.
0x04. 代码实现
代码主要就说Server端的Raidnode
了, Client的实现暂略. 由于Raidnode的本地实现基本不具有实际可用性, 所以先只看分布式的DistRaidNode
实现
1.ConfigManager
它只做一件事, 就是定时判断是否重新加载配置文件, 线程的run
方法的如下:
1 | public void run() { |
2.TriggerMonitor实现
上面reload配置文件后, 具体做事的线程是TriggerMonitor, 它会定期扫描配置目录, 并调用DistRaid
以MR方式提交raid任务:
Trigger线程主要做的如上图左侧所示, 它扫描目录和读配置的地方不是重点, 关键就两点:
- 初始化任务
- 执行任务
下面先看看整体的代码结构, 然后后续围绕这两点再细说:
1 | // TriggerMonitor循环调用doProcess()方法 |
上面selectFiles()
的实质是调用RaidState.check()
来根据多个条件筛选待raid文件, 因为不是重点, 暂不细说, DistRaid的具体调用下面单独说.
备注: 在Facebook原版代码中, 准备任务前还有个切分文件并封装为EncodingCandidate
的操作, 详见源码, 但是后续改良的版本里把这部分去掉了, 原因暂不确定, 所以引的图片我也做了对应调整.
3. DistRaid (核心)
DistRaid简单来说是一个封装了MR作业具体操作的类, 基本所有作业相关都是在这里定义和实现的, 它有三个比较关键的点:
setup()
方法预处理文件DistRaidInputFormat
内部类DistRaidMapper
内部类
1. 预处理文件
预处理做除了一些基本job参数设置, 主要做的是, 把一个目录下待raid的多个文件, 转换为一个SequenceFile格式的特殊文件, 类似的做法也在distcp
实现中采用了, 不过我们这里待raid的文件都是大文件, 合并它们应该不是存储占用上的考虑, 且它也没使用压缩, 具体是什么原因还不太清楚..
1 | private boolean setup() throws IOException { |
这里只看代码会显得过于抽象, 也不理解为什么实际应用里会把SYNC_FILE_MAX
改为0(每写一个文件就写一个SYNC
标记), 其实这是一个原版的BUG, 会导致严重的map读取倾斜, 如下图所示: (基于原图修改)
解释一下上图的意思: 理想情况下每个map读取的文件应尽可能均匀, 但由于采用了SequenceFile的格式设计, 它对应使用的读取器SequenceFileRecordReader
是按SYNC
标志为每次读取”开始-结束”的, 所以当设置为10个文件写一个SYNC标记, 就可能造成map1
读取了两个SYNC内10个文件, 而其他的map读不到数据的情况, 也就是图中间的例子. 图最右部分就是改成每写入一个文件写一个SYNC标记. (这里还有疑问, 包括seq文件为啥没有看到? seq文件一定合并为1个么?)
2. DistRaidInputFormat实现
类似distcp
的MR实现一样, DistRaid也是只有map阶段, 没有reduce阶段的, map的实现就是DistRaidMapper
, 之后再看,
简单说InputFormat接口规定了MR阶段输入文件的规范, 实现类需要通过两个方法来确定读什么, 怎么读:
- getSplits(): 按给定规则切分源文件, 切成一条条split数组 (就好比把一张纸切成N条, 每条有M个记录)
- getRecordReader(): 对split后的数据条再拆分, 转为map具体读取的一条条K-V数据, 核心
next()
方法
然后Client提交任务后, JobTracker会把数据条分给对应的mapper处理, 下面来看看getSplits()
的实现: (代码重构过, 原始版本参考)
1 | /* 切分核心关注两点: |
刚才的getSplits()
先把SeqFile从一个大文件切成若干个数据条(FileSplit
对象), 切分方式类似于:
split1:
SYNC1 --> file1 --> file2 ...-->file10
(起始位置A)split2:
SYNC2 --> file11 --> file12 ...-->file20
(起始位置A+B)split3…..同上直至写完
接下来把每个数据条传入SequenceFileRecordReader
的读取具体K-V
对的方法中,这样map才能直接读取一个个K-V
对. 我们需要溯源一下底层实现:
1 |
|
这里简单看SequenceFileRecordReader
的构造方法还不是很清楚如何读的每一个K-V, 其中V在raidnode
中应该就是一个个的文件.
未完待续
3. DistRaidMapper实现
这是map的具体实现, 因为raid的过程只执行map阶段, 所以它的实现等于就是raid的MR实现, 是怎么做的关键, 也就是如何生成校验文件.
- 获取目前raid所需的所有信息
- 对文件进行一系列检查, 确认满足raid条件
- 生成校验(parity)文件
下面是map的实现源码, 核心其实就是去调用doRaid()
方法, 然后统计一下执行结果.
1 |
|
关键在于第三点, 如何生成校验文件, 如图所示:
- 创建一个临时的目录和临时文件(tmp)
- 通过
ParallelStreamReader
来并行读取多个源数据block, 用于生成校验文件, 对应4个校验块 - 第一个块先直接写到HDFS上
- 后续的块先写到本地的临时文件中
- 再调用RS算法将这些block追加到之前存储第一个block的HDFS文件中
- 清理收尾, 之后将源文件副本设置为1, 实际是委派给了NN执行 (默认),
图中EC编码过程也就是代码的Encoder.encodeFile()
, 是生成校验文件的核心, 进入方法后的核心是encodeFileToStream()
, 再进入后最后落到encodeStripeImpl()
方法, 里面使用多线程对应多个输入流, 最后对每一小块字节进行具体编码是encodeStripeParallel()
, 这个之后补充个时序图, 调用就会清晰许多.
时序图待补…
ParallelStreamReader
是并行读取源文件block的关键, 它的实现还稍微复杂一些, 又分了几个关键成员:
- ReadResult子类: 读取的数据会封装成此数据结构, 然后放在有界队列
boundedBuffer
中, 防止一次读取过多数据 - MainThread子线程类: 观察队列是否有元素, 有就取出进行编码
- ReadOperation子线程类: 由MainThread的线程调度, 可以理解为MainThread的具体读取的子线程.
这里面主要的内容其实分为两块, 一块是多线程编程相关, 用到了线程池, 有界队列和信号量, 这里不单独叙述, 等并发编程专题单独说, 另一部分是IO细节, 包括如何操作每一部分字节并进行编码, 暂时跳过, 代码后续再补, 可以先看看整体的结构图 (细节在RSEncode和RSDecode中)
如上图所示, 解释一下:
- 源文件的每1个块对应1个输入流, 对应到ReadResult就是用一个二维字节数组
readBufs[blockLength][bufferSize]
来存储读取每一个块的数据 - 默认每一次读取1M的数据, 就调用encode编码一次
- 校验文件的第一个块会通过输出文件的输出流直接输出到DN中
- 剩余的块会保留在本地后期再追加到检验文件中
- 生成校验文件后会把它与源文件的
lastUpdateTime
设为一致, 以确定降幅本成功(啥意思?) (重要)
4. BlockIntegrityMonitor
1. 整体结构
上面的TriggerMonitor
主要做的是对普通文件raid, 是编码的过程, 那么BlockIntegrityMonitor
就是做相反的事恢复丢失的块, 是解码的过程.
它主要实现类是由Work线程类的两个子类:
- CorruptionWorker: 定期与NN通信, 确定raid的源文件和校验文件是否有块损坏的, 有则提交MR任务进行解码恢复损坏的块
- DecommissioningWorker: 同HDFS里的概念, 这里是部分DN被安排下线后的块状态, 提交逻辑和上面几乎一样 (任务优先级稍低)
这里之后补个图看看整体流程, 优先级计算不是重点就不说了, 关键在于下面的2和3, 也就是提交了的修块任务如何执行的, 代码很多
2. ReconstructionInputFormat
同之前提到的, 这里主要也是getSplits()
确定如何切分输入. 篇幅还挺长..
3. ReconstructionMapper
这是修块的实际map实现类, 要注意的是, 根据文件不同类型, 修复逻辑各不一样
- 源文件
- 校验文件
- 归档后的校验文件
推测最麻烦的应该是归档文件的修复? 这里的篇幅还挺多的其实… 先暂时跳过把
未完待续…
5. PlacementMonitor与PurgeMonitor
PlacementMonitor与PurgeMonitor配合使用完成删除过时的校验文件以及将Raid文件均匀分散到不同DN的功能。
6. HarMonitor
它做的主要是, 定时扫描需要归档的校验文件, 通过archive方式把他们合并为若干个大文件(har包), 以此来减轻大量校验文件对Namenode的压力, 关注几个点:
- 合并后的校验文件, 如果要读取(恢复), 是否会麻烦?
- 合并之后, 原有的多个校验文件合成了一个大文件, 那就需要额外对的映射管理, 是否麻烦?
0x05. 高版迁移
顾名思义, 高版迁移就是把Raidnode引入到高版本的HDFS上, 虽然这里说的代码大部分已经是修改或优化过的版本, 但是还是不能直接在高版上直接跑, 这里涉及到几个问题.
- 相关RPC是否兼容
- Yarn3.X需要适配(有些方法需要修改)
- 旧的HTTP服务是抛弃还是修复让它可以运行
- DN信息是否可以正常获取
- 生成校验文件是否完整, 恢复是否正常.
那为什么要做这件事, 迁移到HDFS2/3这种可能已支持EC的高版本上呢? 主要还是因为高版本EC(新EC)存在一些关键问题:
- 开启新EC之前的旧数据都无法读取, 只能在全新的集群使用
- 新EC开启后, 因为整个存储结构变化, 有一系列操作如
append/hflush/hsync
都无法再使用, 而这些API可能被上层的Hbase等直接调用 - 新EC的线上稳定性还没有广泛验证, 特别是在巨大规模的存储量下, 大家都在摸着石头过河ing
- 旧的EC虽然有种种问题, 但是它是几乎外挂解耦的, 方便管理操作, 有问题解决起来较容易, 新版EC完全cover的人都还不多
所以在目前的情况来说, 先让旧版的EC方案能在高版本HDFS3跑起来, 然后等新EC经过充分测试和研究后, 再使用是符合生产环境要求的. 然后下一篇单独记录迁移问题中需要修改的点. (包括Yarn的部分)
0x06. 问题与展望
上面介绍了RaidNode相关内容, 主要是说的它的优点, 下面来看看它的不足, 和为什么社区需要新的EC特性, 以及未来的发展趋势
1. 硬资源成本
低版本提交的RaidNode, 存在两个明显的物理资源占用:
- 非常高的CPU编码/解码资源占用 (高版的EC通过支持Intel自带的指令集ISA-L可以显著降低, 可参考官方Benchmark)
- 数据恢复时, 大量的网络带宽消耗
软件层面可以通过检测集群的空闲时段(比如夜间), 主要在这段时间做生成/恢复block的操作. 从而提高资源的合理使用.
其他层面的话, 一般是只将其用在偏冷(访问次数少, 文件生成时间早)的数据集群, 减少其劣势出现的几率, 最典型的例子就是各种备份集群, 大文件(云盘)存储集群.
2. 软资源成本
除了直观可见的硬件上的资源占用, 它还有隐性一些的软件层面的资源占用, 同样不可小觑:
每1个满足raid条件的文件, 都会生成1个对应的校验文件(以及对应目录), 存在Namenode中, 等于说元数据INode可能会翻倍
解决方案:考虑到读校验文件很少, 所以可对校验文件合并归档(Archive), 从而大幅减少INode占用
RaidNode会定期的与NN通信, 确认哪些文件需要raid, 已被删除的raid的文件, 以及缺失的块信息等, 如果访问周期较短, 也会对NN-RPC造成一定压力. 此外Raidnode如果发现某个DN上存储文件的
source block
过多, 就会将它移到其他DN上, 但是此时NN并不会更新block-DN的映射, 就会导致读数据出现异常重试.解决方案: 根据线上NN的实际访问情况, 调整/延长RaidNode的各种周期检查时间, 比如扫描周期应大于DN的块汇报时间.
RaiNode引入后, 新增/删除一个文件的块, 在NN端分配DN的时候都会更加麻烦, 原因在于raid后的文件的块需要尽可能保证不存在同一个DN上, 以免某个DN挂掉导致多个block同时丢失, 默认的NN选择块的策略并非如此. 此外, 这个还会严重影响NN的启动耗时, 因为NN启动需要重建元信息, 对比增删block.
解决方案: 启动NN的时候先采用默认的block分配策略, 完成后再修改为考虑Raid的分配策略, 动态刷新生效 (这个如何做的? 待确认)
raid后的文件块同样还会影响到原有的balance机制, 默认的balance不会考虑raid, 从而可能把多个
source block
放在一个DN解决方案: 在NN上添加新的RPC接口, 用于查询block所属文件, 并结合raid的块选择策略, 将其均匀分散 (待确认)
Raid完成后, 原本文件的三副本实际降为1副本, 潜在的影响有本地读概率降低, 从而增加计算作业的执行时间的可能
解决方案: 本质同硬件资源, 只对冷数据采用Raid可明显降低此影响
当然功能上说, 它也有一系列的小问题, 比如因为它会归档校验文件的机制, 使得操作归档后的校验块就会很麻烦.
3. “小文件”问题
要知道上面RS模式讲的是以”10+4”为基准的, 那么需要考虑一个问题, 10个块的文件有4个校验块, 那5个块的文件呢? 2个块的文件呢?
实际情况是, 集群中会存在大量的”小文件“(block<3), 那它还能获得1.4X的效果么? 答案显然是不能的, 下图是两种校验方案不同block数的节省资源对比图:
10+4
模式下, 不管数据块多少个, 都有4个校验块- 一个文件对应的block越小, 通过RAID能节省的空间就越少
- 一个文件少于3个block, 那它基本不能节省任何空间 (<3个block的文件, 就在这被定义为”小文件”)
要知道3个block及以上的文件至少是256MB+, 大部分场景中, 小于256MB的文件肯定是不少的, 所以这个问题全局影响是很大的, 比如你的网盘中, 虽然会有电影和一些大安装包单文件占用上G, 但是大部分存储的仍然是小文件, 比如照片库可能总共十几G, 是由上万张每个几MB的图片组成. 按文件粒度, 它们都无法被Raid
那如何解决这个问题呢? 有思路是把Raidnode的操作单元从文件级别, 提高到目录级别, 这样一个图片目录下的许多小文件就可一起被Raid, 详细可参考此篇.
TODO: 关于Raidnode的有一些细节, 这篇限于篇幅还说的不够细致, 比如生成校验块使用编码的地方, 修块的细节, 之后参考此再细致补充一篇, 因为raidnode年代已久, 所以其实大家不太愿多花时间在它身上了, 不过我觉得好好学习一下设计和实现对高版EC还是很有帮助的. (包括理解和对比)
参考资料: