一、大数据架构变迁

伴随着业务的拓展和数据量的积累，公司早在 2016 年便开始大数据体系的构建。彼时 Hadoop 生态正大行其道，我司也基于此搭建了第一代大数据集群。该集群以自建 IDC 机房为主要载体，使用 CDH 发行版进行管理，承接着来自多个业务线的数据处理任务。该集群中以 HDFS 为数据存储载体，主要使用 Apache Spark 进行 ETL 处理，依靠 YARN 进行任务资源规划。为了应对偶发性的 Spark 任务出现大量 Shuffle 数据的情况，每个 YARN 管理的节点均挂载 8T 大磁盘承接临时数据，保障集群任务顺利完成。

虽然集群稳定运行多年，但随着数据量的增加，其短板也日渐凸显。

部署运维成本高

IDC 机房的建设会带来巨大的投入，更不用提后期人工运维、机房管理、迁移等一系列支出更是海量。每次为客户进行新集群搭建和平台部署时，都需要运维同学和开发同学全力参与，手工部署处理。即使是这样全力投入，也还是常常出现因 CDH 某参数遗漏导致的停滞和返工。这些都给集群的运维带来了沉重的负担。

资源规划粗放，精细化管理困难

Hadoop 生态的很多组件都没法做到精细化资源管理，为了更大限度充分利用资源，多个组件会混合部署在一台机器上。比如为了利用 HDFS 的短路读（Short-circuit Read）优化机制，会将 YARN 的 NodeManager 和 HDFS 的 DataNode 节点部署在一起。但突然出现的超大 SQL 查询就可能直接占满某台机器的本地磁盘，进而引发集群报警，对集群的稳定性产生了一定打击。

组件繁多且复杂，版本耦合性太强

CDH 发行版本的组件繁多，架构复杂（共包括11类服务（存储、计算、运维、监控、安全等），22种角色类型）。并且随着时间推移，很多新技术引入异常麻烦，需要考虑非常多兼容性问题。

后来，随着业务的迅猛发展，传统多租户的方式已经无法满足客户对数据安全和资源成本管理的需求，CDH 相关组件版本的滞后也导致客户无法体验到最新的功能和特性。为了平稳支撑多点DMALL 全面 To B 业务的转型，我们迎来了一个契机，可以彻底甩开历史包袱，设计搭建存算分离、轻量级、可扩展、云中立大数据集群架构。

二、引入历程

2.1 Shuffle 处理的初次设想

在逐步设计搭建云原生存算分离架构时，我们曾考虑过使用与 on YARN 时一样的处理方案，即使用 Kubernetes 机器挂载超大磁盘解决 Shuffle 问题。但这一想法很快就被我们自己否掉了，毕竟这明显违背了 on Kubernetes 的初衷。

在调研使用 JuiceFS + 对象存储替代传统 HDFS 的方案时，我们认为 JuiceFS Kubernetes CSI 驱动的 writeback 模式理论上很适用于 Shuffle 数据处理。这种模式下，数据会优先写入本地缓存目录，超过阈值后异步推入远程对象存储。可惜在性能测试时发现，任务的效率明显无法与 on YARN 相抗衡，甚至部分 SQL 产生了近 12 倍的差距。究其根本原因是 Spark 的 Shuffle 场景本身会存在大量小文件及随机读操作，但是 JuiceFS 的 writeback 模式并不适用于这种情况。

30f13b298d23eeccb924dde5b61b7a80_640_wx_fmt=png&from=appmsg&tp=webp&wxfrom=5&wx_lazy=1&wx_co=1.webp

2.2 引入 Apache Celeborn

在社区同学的建议下，我们开始调研开源的 Remote Shuffle Service（简称 RSS）。经过调研参数对比和实际数据测试，最终我们选用了 Apache Celeborn（后简称 Celeborn）。

2b0ae4ffaeab9c16df270a4baa522a7d_640_wx_fmt=png&from=appmsg&tp=webp&wxfrom=5&wx_lazy=1&wx_co=1.webp

上图粗略描述了加入 Celeborn 后，一个 Spark 任务的 Shuffle 数据流转情况。当 Spark 任务执行到需要进行 Shuffle 数据处理时，在 Executor 的 Celeborn Client SDK 会自动读取 Shuffle 数据，通过与 Celeborn Master 通信后获取可使用的 Celeborn Worker 信息，并与相关 Celeborn Worker 产生连接，将数据推到 Worker 端。根据配置，Celeborn Worker 默认会先写入本地 SSD 磁盘进行临时存储。待本地 SSD 磁盘临时存储超过阈值后，Worker 会自动降级，写入 HDFS 中。在我们的存算分离集群中，这一步便是经由 JuiceFS SDK 写入对象存储中。

考虑到成本控制，云原生的大数据集群并非完全使用云厂商提供的弹性节点，而是会有一部分固定节点部署常驻服务和引擎。这些固定节点上会挂载较小的 SSD 硬盘用作缓存和临时文件存储等。Celeborn 通过手动部署的方式部署在这些固定节点上，以 HA 模式部署 3 个 Master 节点，以及与固定云节点等量的 Worker 节点。每个 Worker 节点被设置允许最多使用 100G 左右磁盘，超出时则写入对象存储。通过对历史任务的分析发现，大部分任务 Shuffle 数据量适中，通过磁盘完全可以快速处理完成，而有对象存储在背后“托底”，我们也不用担心偶尔也会出现超大的 Shuffle 量的任务。虽然写入对象存储的速度相比本地磁盘要慢一些，但是借用组内小伙伴的话，“这么大 Shuffle 数据量的任务，就要有长时间才能结束的觉悟”。这样的设计也算是尽可能在成本和效率中间找到一个平衡点。

加入 Celeborn 后，我们的性能测试就顺利通过了。除了因为组件的不同，在 Kubernetes 上申请资源相比 YARN 上速度慢 10s 左右外，运行效率与 YARN 齐平，部分任务甚至速度还有提升。

三、升级演化

我们于 2022 年末正式引入 Celeborn，至今已两年有余。在这段过程中，我们的存算分离平台从初上线到如今几十个集群的规模，在保障Spark任务稳定运行中 Celeborn 一直扮演着重要角色。同样在这段历程中，我们也在跟踪社区一起不断讨论、解决问题，升级优化，推动 Celeborn 组件在集群中更稳定可靠的运转。

3.1 版本跟踪升级

我们从 v0.1.4 版本开始引入 Celeborn 并进行测试，历经多次测试后，初次上线时已经开始部署 v0.2.1 版本，到如今大部分集群都升级到了 v0.5.1 版本。在这个过程中，我们遇到过一些问题。例如：

仅支持 HDFS 协议

最初我们试图引入时发现 Celeborn 在代码中仅支持 HDFS 协议（例如hdfs://），为了支持使用 JuiceFS 协议（例如jfs://）对接对象存储，我们把相关代码做了调整，并提交给了社区

（https://github.com/apache/celeborn/pull/1360；https://github.com/apache/celeborn/pull/1367）

Revive for shuffle 0 partition xxx failed.

典型报错日志：

Shell
org.apache.celeborn.common.exception.CelebornlOException: Revive for shuffle 0 partition 182 failed.

与社区沟通后，把下面的 patch 打入代码中：

https://issues.apache.org/jira/browse/CELEBORN-656

Uncaught exception in thread ThreadlDatafusher

典型报错日志：

Shell
Uncaught exception in thread ThreadlDatafusher-3453,5,main at org.apache.celeborn.client.ShuffleClientImpl.revive(ShuffleClientImpl.java:697)

与社区沟通确认后，找到了对应的 JIRA

（https://github.com/apache/celeborn/pull/1819），最终通过升级版本修复。

当本地 SSD 磁盘满后降级写 HDFS 时，原有 HDFS 目录清理失败

这个问题社区同学与我们一起研究后发现，是由于我们是利用 JuiceFS SDK 写对象存储，而 Celeborn master 的依赖目录下并未加入 JuiceFS SDK 导致的。之后我们加入依赖，问题便完美解决。

3.2 从 on Premises 到 on Kubernetes

多点 DMALL 的整个数据中台为了可以做到通过 Helm charts 小时级部署，几乎所有的组件都重构并做了镜像。但是最初 Celeborn 的部署依旧需要人工手动处理，这给部署运维工作带来了一定麻烦。后来社区开源了镜像和 Helm chart 的部署方式，我们也随之将 Celeborn 的部署修改为了 on Kubernetes 的方式。为了成本管控，Celeborn 与其它平台和引擎混部在一起，只是考虑到稳定性，在节点亲和性上做了处理，依旧会限制分发到大数据 Kubernetes 集群的固定节点上。

我们的存算分离集群依旧使用 JuiceFS + 对象存储的方式替代原有的 HDFS 存储，因此我们在开源 Helm 包和镜像都做了些调整和修改，包括添加 JuiceFS 依赖包，添加需默认挂载的 Volume 和 ConfigMap ，以及自动注册 Prometheus 进行 Merics 采集等问题。后者我们也整理贡献给了社区。

（https://github.com/apache/celeborn/pull/2673）。

3.3 从 Hostpath 到 PVC

随着客户业务数据量的增长，通过固定节点 Hostpath 方式进行 shuffle 数据暂存已经无法满足需求，很多任务也开始更频繁的报警。

经过分析后我们发现，与其它应用抢占本地磁盘成为 Celeborn 处理任务的阻碍。于是经过调研，我们将目光锁定在 Kubernetes 的 PersistentVolumeClaim（简称 PVC）上。相较于固定容量的 Hostpath，使用 PVC 的方式在容量的申请上更加的容易，对于支持可扩展的 StorageClass 而言还能够在使用中按需扩展容量，相关 Pod 无需删除重启。

通过在 Helm 包的相关 Yaml 文件中修改，支持通过配置在部署时选择 shuffle 临时数据存储于原生的 Hostpath 方式抑或是 VolumeClaimTemplates 的方式。云厂商均提供了支持临时数据快速读写的块存储 StorageClass，而且大部分支持按使用量付费，相较固定磁盘而言，成本依旧可控。

当前已经在部分集群压力较大的线上环境使用云厂商块存储进行 shuffle 数据承接，极大地提高了任务执行的稳定性。当然为了成本考虑，对于其它压力可控的集群我们依旧采用 Hostpath 的方式，并进行持续监控，择机进行切换。

57ee556df28e34394bfb3f4f402425a9_640_wx_fmt=png&from=appmsg&tp=webp&wxfrom=5&wx_lazy=1&wx_co=1.png

3.4 BestEffort 带来的不稳定问题

某天突然收到业务的求助，发现在凌晨集群压力最大的一个时间段内 Spark 任务出现大量失败。通过查看监控我们发现在那个时间段内，有一个 Celeborn worker Pod 重启，导致了问题的发生。

深入分析后发现，Celeborn 是通过 JVM 配置限定 Container 的资源使用（例如 v0.5.1 版本中 Worker 节点默认堆内内存 2G，堆外内存 12G），而不是通过 Pod 的 Resource 声明进行限制。这种情况下 Kubernetes 会判定 Pod 的 Quality of Service（简称 Qos）为 BestEffort。简单来说，就是集群资源充足时不受影响，但当 Pod 所属节点资源非常紧张时，会先驱逐（Evict）这部分没有明确声明资源使用的 Pod，以释放资源保证节点健康。

这就使问题陷入了一个难以抉择的状态。经过观察，Celeborn 的 Master 和 Worker 组件在“静息”状态下，单 Pod 仅占用内存 600M 左右。如果直接按照 JVM 配置设置 Pod 的 Resource 资源，大数据架构中的组件 Vocano 会根据申请和限制设置预占用，很明显，这会造成一定资源浪费。尤其是对于我们大部分集群所选的机型而言，每个机器都占用一部分固定内存用于 Celeborn 无疑会造成资源闲置和成本消耗。

非常幸运的是，Celeborn 提供了 Replica 的功能，支持 Client 端在连接 Worker 写数据时，同时再往另一个 Worker 中写入，以防止主 Worker 失败后数据丢失任务失败。我们在 UAT 环境进行了更细致的测试，测试发现开启该功能后，本地磁盘 shuffle 数据写入量确实达到了双倍，处理时间略有延长，但在可接受范围内。

四、总结与展望

当前该套存算分离方案已经为几十个客户搭建专属大数据集群并稳定运行至今，部分集群中的 Apache Celeborn 也已升级到 v0.5.1 版本。因为 To B 客户对成本的极致把控，多点 DMALL 的大数据集群搭建也是在不断探寻成本与效率的平衡，提供极致性价比的大数据服务。多点 DMALL 的 To B 业务相比其它 To C 企业而言数据体量不大，但对于数据的准确性和稳定性要求极高。又要保障集群的稳定可靠，又要平衡成本和效率，Apache Celeborn 容器化部署无疑是最好的选择。

本文来自公众号： Apache Celeborn