携程 Hadoop 跨机房架构实践

本文将分享携程 Hadoop 跨机房架构实践，包含 Hadoop 在携程的发展情况，整个跨机房项目的背景，我们跨机房的架构选型思路和落地实践，相关的改造和对未来的展望，希望给大家一些启迪。

一、Hadoop 在携程的落地及发展情况

携程 Hadoop 是从 2014 年引进的，基本上每年较前一年以两倍的速度在增长，我们对 Hadoop 集群做了大量性能方面的改造和优化。

1)目前，HDFS 存储层面拥有数百 PB 的数据，数千的节点，分为 4 个 namespace 做 Federation，自研了 namenode proxy 来路由 rpc 到对应的 namespace，2019 年初上了一套基于 Hadoop 3 的 Erasure Code 集群来做对用户透明的冷热存储分离，目前已迁移几十 PB 数据到 EC 集群，节省一半的存储资源。

2)计算层面，搭建了两套离线和一套在线 Yarn 集群做 Federation，总量 15 万 +core，每天 30 万 + Hadoop 作业，其中 90% 为 spark。所有节点分布在四个机房，其中离线集群部署在其中两个机房，在线集群部署在三个机房。

二、跨机房项目背景

来看下整个项目的背景。之前我们的 Hadoop 机器部署在金和福两个机房，95% 的机器在福。去年底，携程自建了日机房，同时福机房的机架数达到了物理上限，没办法继续扩容。另外按照目前计算和存储的增速来看，预计 2024 年底集群规模会达到万台，新采购的机器只能加在日机房，我们需要多机房架构和部署的能力。

这其中的难点在于，两个机房的带宽仅 200Gbps，正常情况下网络延迟在 1ms，当带宽打满情况下，延迟会达到 10ms，同时会有 10% 的丢包率。我们需要尽可能减少跨机房的网络使用带宽。

2.1 原生 Hadoop 架构问题

看下原生 Hadoop 的问题。网络 IO 开销主要来自两方面，Shuffle 读写和 HDFS 读写。

1)先看 shuffle，MR 和 Spark 作业的前一个 stage 会将中间临时文件刷到磁盘，下一个 stage 通过网络来 Fetch。如果分配的 map task 在机房 1，reducetask 在机房 2，就会产生跨机房流量开销。

2)其次 HDFS 层面，对于读场景，三个副本存放在不同的节点，客户端会从 namenode 拿到按照距离排好序的副本信息，优先从最近的副本所在的节点读取。但是当三个副本都和客户端不在一个机房的情况下，就会产生跨机房读网络 IO 开销。写场景的话，HDFS 采用 Pipeline 写，选择副本时只考虑到机架层面的存放策略，会将三个副本放在两个机架，如果选择的两个机架跨机房了，也会有跨机房网络写开销。

2.2 可选的方案

当时我们讨论下来有两种架构解决方案，多机房多集群和多机房单集群，两种各有利弊。

2.2.1 多机房多集群

多机房多集群方案的优势是不需要修改源代码，可以直接部署。缺点是：

1)对用户不透明，用户需要修改配置，指定提交到某个集群;

2)运维成本较高，每个机房有独立的集群，配置管理麻烦;

3)最重要的是第三点，数据一致性难以保证。有些公共数据需要被多个事业部访问的话，只能跨机房读取，这个 IO 无法省掉，而如果用 distcp，在本机房也放一些副本以省掉这部分流量开销的话，又会由于副本是通过不同的 namenode 管理的，导致数据可能会有不一致的问题。

2.2.2 多机房单集群

再来看多机房单集群架构，劣势是需要改 Hadoop 源代码，因为动了 BlockManager 的核心代码逻辑，会有风险，需要做好完备的测试和验证。但是好处也很明显。

1)对用户透明，用户不需要关心作业提交到了哪个机房，副本存放在哪里，无感知;

2)运维部署简单;

3)因为是由一个 namenode 来管理副本状态，所以可以保证多机房副本的一致性。

主要由于第一和第三点优势，我们希望保证用户使用时的透明性和一致性，最终选择了多机房单集群方案。

三、先期尝试——在线离线混部跨机房

其实对于第一种多机房多集群方案，我们之前在在线离线混部项目中采用过。当时的场景是，离线集群的资源在凌晨高峰打满，白天低峰较空。而在线 k8s 集群恰恰相反，我们希望利用 k8s 凌晨的计算资源帮我们减轻负担。而 k8s 集群部署在金和欧机房，数据没有本地性。所以我们希望将一些 cpu 密集，但是对 IO 压力又不大的作业，能分配到在线集群。

我们在 k8s 上部署了一套 Yarn 集群，并开发了一套作业资源画像系统，主要是采集作业的 vcore/memory 使用，shuffle，hdfs 读写等 metrics。由于 zeus 调度系统提交的作业一般不怎么修改，每个作业的历史执行时间和所消耗资源都有趋同性，我们按照 zeus jobid 聚合，根据历史多次执行情况分析出每个作业的资源使用趋势。下次作业启动时 zeus 会将 shuffle 量和 hdfs 读写量较低的作业分配到在线集群跑。

另外由于在线集群也跨了两个机房，我们在 FairScheduler 上开发了基于 label 的调度，一个 label 对应一个机房，会根据每个 label 的负载，动态分配作业到所属的 label，一个 app 所有的 task 只会固定在一个 label 内执行，这样机房间不会产生 shuffle 流量。该方案上线后，可以缓解离线集群 8% 的计算压力。

四、多机房单集群方案

我们规划一个事业部对应的一个默认机房，数据尽可能在同机房内流动。由此对于多机房单集群架构改造主要包括四个方面：多机房单 HDFS 集群，多机房多 Yarn 集群，自动化数据和作业迁移工具，跨机房带宽监控和限流。

4.1 多机房单 HDFS 架构

先来看 HDFS 改造，我们改造了 namenode 源码，在机架感知之上，增加了机房感知，NetworkTopology 形成了 < 机房，机架，Datanode> 三元组。这样客户端读 block 时，计算出来和副本所在节点的距离，本地机房肯定小于跨机房，会优先读本地机房数据。

另外我们在 namenode 中增加了跨机房多副本管理能力，可以设置目录的多机房副本数，比如只在机房 1 设置 3 个副本，或者机房 1 和机房 2 各设置三个副本，对于没有设置跨机房副本的路径，我们会在 zookeeper 和内存中维护一个用户对应默认机房的 mapping 关系，写文件 addBlock 的时候，根据 ugi 找到对应的机房，在该机房内选择节点。

Decommission 或者掉节点时候会有大量的副本复制操作，极易容易导致跨机房带宽被打爆。对此，我们修改了 ReplicationMonitor 线程的逻辑，在副本复制的时候，会优先选择和目标节点相同机房的源节点来进行复制，降低跨机房带宽。

为了持久化跨机房路径副本信息，我们增加 Editlog Op 来保存每一次跨机房副本设置变更记录，fsimage 中新增了跨机房副本 Section，这样 namenode 只会保存一份元数据，failover 切换到 standby 的时候也能加载出来，没有其他外部依赖。

4.2 改造 Balancer&Mover&EC

HDFS 层面还有其他一些改造，比如 Balancer，我们支持了多实例部署，每个 Balancer 增加 IP 范围列表，每个机房会起一个，只 balance 本机房 IP 的 datanode 的数据。对于 Mover，我们也支持了多机房多实例部署，因为 mover 是在客户端选择目标副本节点的，所以需要改造按照目录的跨机房副本放置策略在客户端来选择合适的节点。

这边要注意一点的是，尽量保证 proxy 节点和 target 节点在同一个机房，因为真正迁移的网络 IO 是在这两个节点发生的。另外我们在新的日机房部署了一套基于 Hadoop 3 的 Erasure Code 集群，会将一部分历史冷数据迁移过去，目前这块没有做跨机房的代码改造，我们的 EC 迁移程序只会迁移那些已经被迁移到日机房的 BU 的冷数据到 EC 集群。

4.3 副本修正工具 -Cross FSCK

由于我们有多个 namespace，跨机房版本的 HDFS 是一个一个 ns 灰度上线的，灰度过程中，其他 ns 的副本放置还没有考虑机房维度，所以我们开发了 Cross IDC Fsck 工具，可以感知跨机房配置策略，来修正不正确放置的副本。

因为需要不停的读取副本信息，会产生大量的 getBlockLocations rpc 请求，我们将请求改成从 standby namenode 读，一旦发现不匹配会调用 reportBadBlocks rpc 给 active namenode，BlockManager 会删除错误的副本，重新选择新的副本。由于这个操作比较重，高峰时间对 HDFS 会有影响，所以我们在客户端加了 rpc 限流，控制调用次数。

4.4 多机房多 Yarn 集群

下面来看下 Yarn 的改造，我们在每个机房独立部署一套 Yarn 集群，自研了 ResourceManager Proxy，它维护了用户和机房的 mapping 关系，这个信息是和 namenode 共用的，都是内存和 zookeper 各一份。

修改了 Yarn Client，用户提交的 Yarn 作业会首先经过 rmproxy，然后再提交到对应 Yarn 集群。这样一个 app 所有的 Task 只会在一个机房内调度，不会产生跨机房 Shuffle。如果要切换用户账号对应的机房和集群也很方便，会立马通过 zookeeper 通知到所有 rmproxy，修改内存中的 mapping 关系。

rmproxy 可以多实例部署，互相独立，同时在 Yarn Client 做了降级策略，在本地定期缓存一份完整的 mapping 关系，一旦所有 rmproxy 都挂了，client 也能在这段时间做本地路由提交到对应集群。

adhoc 和分析报表大量使用了 Sparkthrift service，presto，hive service 来做计算。对这块常驻服务也做了改造，每个机房各部署一套，客户端之前都是通过 jdbc 直连对应的 thrift service，改造后接入 rmproxy，会先从 rmproxy 中拿到用户对应机房的服务 jdbc url，再连接，这块同样对用户透明。

五、自动化迁移工具

由于日机房的节点会按采购到货情况逐步往上加，所以需要按照计算和存储的容量来规划该迁移哪些账号，这是一个漫长的过程，希望能尽量做到自动化迁移，以 BU-> 账号的粒度进行迁移，我们梳理了迁移流程，分为如下四步：

1)批量设置 BU 对应 Hive 账号开始迁移(初始为 3：0，即福机房 3 份，日机房 0 份)

2)按照 Hive 账号下的 DB 和用户 Home 目录依次设置 3：3，数据复制到日机房

3)账号和队列迁移到日机房

4)观察跨机房流量，回收福机房的计算和存储资源(设置 0：3)

迁移时间过程中有些注意点：

1)迁移过程会耗费大量跨机房网络带宽，需要在集群低峰时间执行，我们是放在早上 10 点到晚上 11 点之间，其他时间会自动暂停迁移，否则会影响线上报表和 ETL 作业的 SLA。

2)即使在白天迁移，也需要控制迁移的速率，一方面是减少 namenode 本身的处理压力，另一方面也是降低带宽，白天也会有一些 ETL 和 adhoc 查询需要跨机房访问数据，若打满的话也会有性能影响。迁移中我们会实时监控 namenode 的 UnderReplicatedBlocks 和跨机房流量 metrics，根据这些值动态调整迁移速率。

3)实时监控被迁移机房的 hdfs 可用容量，包括不同的 StorageType 的，防止磁盘打爆。还有有些 hive DB 库目录设置了 hdfs quota，也会由于迁移设置 3：3 超过 quota 而报错，我们会自动暂时调高 quota，等迁移整体完成后再把 quota 调回去。

4)公共库表由于被多个 BU 都有访问依赖，需要提前设置多机房的副本，我们有个白名单功能，可以手动设置，一般设为 2：2，每个机房各放两份。

六、跨机房带宽监控 & 限流

实践中有些 BU 的表，会被当做公共表来使用，我们需要识别出来，设置跨机房多副本策略。目前的 hdfs audit log 中，没有 dfsclient 访问 datanode，datanode 和 datanode 传输数据的实际流量 audit 信息，而我们需要这部分信息来看实际的路径和 block 访问情况，做进一步数据分析，另外当流量打爆的况下，需要有一个限流服务按照作业优先级提供一定的 SLA 保障，优先让高优先级作业获取到带宽资源。

对此我们开发了限流服务，在 dfsclient 和 datanode 代码中埋点实时向限流服务汇报跨机房读写路径，block 读写大小，zeus 作业 id 等信息，限流服务一方面会记录流量信息并吐到 ES 和 HDFS 做数据分析，另一方面会根据作业的优先级和当前容量决定是否放行，客户端只有获得限流服务的 Permit，才能继续执行跨机房读写操作，否则 sleep 一段时间后再次尝试申请。

有了实际的流量信息后，通过离线数据分析，就很容易知道哪些表会被其他 BU 大量读，通过自动和手动结合方式设置这部分表的跨机房副本数 2：2。设置后跨机房 Block 读请求量下降到原来的 20%。跨机房带宽原来是打满的，现在下降到原来的 10%。

七、总结与未来规划

总结一下，本文主要介绍了携程 Hadoop 跨机房实践，主要做了如下改造：

1)实现单 hdfs 集群机房感知功能，跨机房副本设置

2)实现基于 rm proxy 和 yarn federation 的计算调度

3)实时自动化存储和计算迁移工具

4)实现跨机房流量监控和限流服务

目前整套系统已在线上稳定运行了半年，迁移了 40% 的计算作业和 50% 的存储数据到新机房，跨机房带宽流量也在可控范围之内，迁移常态化，用户完全不需要感知。

未来我们希望能智能决定该迁移哪些账号，大多数公共路径设置为 2：2 四个副本，比通常会多加一个副本的物理存储量，现在是设置在表层面，希望能进一步细化到分区层面，因为分析出来大多数下游作业都是只依赖最近一天或者一周的分区。所以一旦过了时间，完全可以将历史分区设置回三副本来减少存储开销。最后是将跨机房的改造也应用到基于 Hadoop 3 的 EC 集群，也支持跨机房的能力。

作者：昱康，携程架构师，对分布式计算和存储、调度、查询引擎、在线离线混部、高并发等方面有浓厚兴趣。