Apache Celeborn在PB级Shuffle混部场景的实践与优化

发布时间：2026/10/31 13:20:47
Apache Celeborn在PB级Shuffle混部场景的实践与优化

1. 背景与挑战：PB级Shuffle的混部需求

在vivo大数据平台的演进过程中，我们面临着一个典型的基础设施矛盾：在线业务（如互联网服务和算法服务）呈现出明显的"昼高夜低"流量特征，而离线大数据任务则集中在夜间执行。这种资源使用的潮汐现象导致服务器在白天高峰期资源紧张，而在夜间利用率不足。2023年启动的混部项目正是为了解决这一矛盾——通过在K8S在线集群中运行大数据离线任务，实现资源的错峰利用。

然而，这种混部架构对Shuffle服务提出了前所未有的挑战。传统Embedded Shuffle Service（ESS）存在几个致命缺陷：

  • 资源隔离差：Shuffle过程占用Executor内存，容易引发OOM导致任务失败
  • 磁盘I/O不可控：混部环境下难以保证本地磁盘的独占使用
  • 扩展性瓶颈：单节点Shuffle数据量超过TB级时性能急剧下降

经过对主流方案的深度评估（技术选型细节见2.1节），我们最终选择Apache Celeborn作为Shuffle服务解决方案。这个决策基于三个关键判断：

  1. 架构匹配度：Celeborn的远程服务化设计完美契合混部场景的资源隔离需求
  2. 性能天花板：其Off-Heap内存管理和多磁盘并发机制可支撑PB级数据传输
  3. 运维友好性：完善的Prometheus指标监控体系满足生产环境要求

2. 技术选型：为什么是Celeborn？

2.1 主流Shuffle方案横评

在Spark生态中，Shuffle服务的演进经历了三个阶段：

  1. 原始阶段：基于磁盘文件的HashShuffle（Spark 1.x）
  2. 优化阶段：引入SortShuffle和ESS（Spark 2.x）
  3. 服务化阶段：Remote Shuffle Service（Spark 3.x）

我们对三种主流RSS方案进行了为期两个月的对比测试：

评估维度CelebornUber RSSFirestorm
Spark版本支持全版本≤3.0≤3.1
动态资源分配原生支持部分支持不支持
监控体系Prometheus集成基础指标无
内存管理Off-Heap + 堆外缓存纯堆内混合模式
数据可靠性多副本+校验和单副本三副本
最大单节点吞吐12GB/s8GB/s10GB/s

测试环境配置：

  • 集群规模：10 Worker节点（256G内存/24核/12×4T HDD）
  • 测试工具：Spark TPC-DS 10TB数据集
  • 网络：25Gbps RDMA

2.2 Celeborn的核心优势

经过实测验证，Celeborn在以下方面展现出显著优势：

内存管理创新

// Celeborn的PushBuffer实现（简化版） public class PushBuffer { private final ByteBuffer buffer; // 堆外内存 private final int maxRetries; public void push(ByteBuffer data) { int retry = 0; while (retry < maxRetries) { try { if (buffer.remaining() >= data.remaining()) { buffer.put(data); return; } flushToDisk(); } catch (IOException e) { retry++; } } throw new PushException("Push failed after retries"); } }

这种设计带来两个关键收益：

  1. 避免GC停顿：测试显示Full GC频率从ESS的每小时5-10次降至零
  2. 内存效率提升：相同数据量下内存占用减少40%

磁盘I/O优化Celeborn采用"多磁盘轮询+小文件合并"策略：

  1. 每个Worker管理12-24块物理磁盘
  2. 通过celeborn.worker.flusher.threads控制每磁盘并发数
  3. 默认256MB的FlushBuffer减少随机写

实测表明，该策略在混部环境下仍能保持稳定的磁盘吞吐：

场景平均IOPS吞吐量延迟(ms)
纯Celeborn85002.1GB/s1.2
与在线业务混部62001.6GB/s1.8

3. 性能优化实战

3.1 异步OpenStream机制

问题现象在Shuffle Read阶段，我们观察到周期性出现的"阶梯式"延迟：

[Stage3] Read 2.4GB: 56ms (正常) [Stage3] Read 128MB: 12s (异常) [Stage3] Read 512MB: 43s (严重恶化)

根因分析通过arthas抓取Worker线程堆栈，发现阻塞点在：

"transport-io-3" #23 daemon prio=5 os_prio=0 tid=0x00007f8e5c0e5000 nid=0x1e3e runnable [0x00007f8e34efc000] java.lang.Thread.State: RUNNABLE at java.io.FileInputStream.readBytes(Native Method) at java.io.FileInputStream.read(FileInputStream.java:255) at org.apache.celeborn.common.network.util.LimitedInputStream.read(LimitedInputStream.java:72) - 阻塞在文件排序的IO等待

解决方案实现OpenStream请求的异步化处理：

  1. 客户端改造：
// 修改前的同步调用 val stream = transportClient.openStream(shuffleId, reduceId) // 修改后的异步调用 val future = transportClient.asyncOpenStream(shuffleId, reduceId) future.onComplete { case Success(stream) => // 处理流 case Failure(e) => // 错误处理 }
  1. 服务端优化：
  • 将文件排序操作移至后台线程
  • 采用非阻塞IO模型处理请求

效果验证优化前后关键指标对比：

指标优化前优化后提升幅度
P99读取延迟4.2s320ms13x
Worker CPU利用率85%62%27%↓
吞吐量波动率±45%±12%73%↓

3.2 小文件内存缓存

问题定位通过监控发现，小于1MB的文件读取耗时与文件大小不成正比：

文件大小平均读取耗时主要耗时占比
1KB1.8s99% IO等待
1MB2.1s95% IO等待
10MB120ms30% IO等待

实现方案在Commit阶段增加缓存判断逻辑：

public class FileMemCacheManager { private final ConcurrentHashMap<File, ByteBuffer> cache = ...; private final long maxCacheSize; public void commitFile(File file) { if (shouldCache(file)) { ByteBuffer buffer = readFileToBuffer(file); cache.put(file, buffer); } // 原处理逻辑 } private boolean shouldCache(File file) { return file.length() < 1_000_000 && currentCacheSize() < maxCacheSize; } }

缓存策略配置

# 缓存阈值（默认1MB） celeborn.worker.filecache.threshold=1048576 # 最大缓存比例（默认10%堆外内存） celeborn.worker.filecache.maxMemoryRatio=0.1

收益分析

  • 小文件读取延迟下降92%（从秒级到毫秒级）
  • 磁盘IOPS降低35%，延长HDD使用寿命
  • 副作用：增加约5%的内存开销

4. 稳定性保障体系

4.1 动态负载均衡

负载评估模型我们设计了一套基于指数加权移动平均（EWMA）的负载评分算法：

load_score = α * disk_util + β * io_wait + γ * pending_tasks 其中： - α=0.5, β=0.3, γ=0.2 - 各指标归一化到[0,1]区间

实现逻辑

class LoadAwareSlotStrategy extends SlotStrategy { override def allocateSlots(workers: Seq[WorkerInfo]): Map[WorkerInfo, Int] = { workers.map { worker => val load = calculateLoadScore(worker) val slots = if (load > threshold) 0 else baseSlots - loadPenalty(load) worker -> slots }.toMap } }

关键参数

# 负载阈值（超过则停止分配） celeborn.slots.assign.loadAware.discardThreshold=0.8 # 惩罚系数（每0.1负载减少的slot数） celeborn.slots.assign.loadAware.penaltyFactor=2

4.2 异常流量控制

识别算法采用改进的Z-Score算法检测异常应用：

对于应用A的Push流量序列X： 1. 计算移动窗口(5min)均值μ和标准差σ 2. 计算当前值x的z-score：z = (x - μ)/σ 3. 若z > 3且x > 200MB/s，判定为异常

限流实现

public class PushController { private final RateLimiter limiter; public void pushData(ByteBuf data) { if (isAbnormalApp(appId)) { if (!limiter.tryAcquire()) { throw new RateLimitException(); } } // 正常处理 } }

效果验证在流量突增场景下（模拟测试）：

  • 未启用限流：Worker OOM崩溃率100%
  • 启用限流：系统保持稳定，仅异常应用受影响

5. 运维实践与经验

5.1 硬件选型建议

基于对三种服务器配置的测试数据，我们总结出Celeborn的硬件黄金组合：

组件推荐配置替代方案避坑指南
CPU24核/2.4GHz+16核/3.0GHz避免低频多核（如64核/1.8GHz）
内存256GB DDR4128GB（需调小缓存）不要配置NUMA
磁盘12×4T HDD（JBOD）6×8T HDD（RAID0）绝对避免RAID5
网络25Gbps以太网10Gbps（需RDMA）禁用TCP Offload

关键发现：

  • 磁盘数量比容量更重要：12块HDD的聚合带宽（~6GB/s）优于4块SSD（~4GB/s）
  • 内存带宽是瓶颈：在PB级Shuffle场景，DDR4-3200比DDR4-2400提升15%吞吐
  • 网络协议栈优化：启用net.ipv4.tcp_rmem=4096 87380 2147483647可将吞吐提升20%

5.2 关键监控项

我们构建的监控体系包含三个层级：

1. 基础资源层

# 自定义的磁盘健康检查脚本 #!/bin/bash for disk in /dev/sd*; do smartctl -H $disk | grep "FAILED" && \ echo "$disk SMART FAILED" | tee -a /var/log/celeborn_disk_alert.log done

2. 服务指标层

  • 核心Grafana看板：
    • Push吞吐量（分App）
    • Fetch延迟百分位（P50/P90/P99）
    • 各Worker的Slot使用率
  • 关键告警规则：
    # PromQL示例 rate(celeborn_worker_fetch_failures_total[5m]) > 5 and celeborn_worker_disk_usage_ratio > 0.9

3. 业务影响层

  • Shuffle阶段耗时占比
  • 失败任务中的Shuffle相关错误码
  • 各业务线的SLA达标率

5.3 升级与维护

滚动升级步骤

  1. 分批标记Worker为DECOMMISSIONING状态
  2. 等待存量Shuffle数据被读取完成（通过celeborn_worker_active_shuffles指标）
  3. 逐台执行升级（保留数据目录）
  4. 验证新版本Worker后加入集群

配置变更黄金法则

  • 每次只变更一个参数
  • 变更后观察至少30分钟
  • 重要参数采用渐进式调整：
    # 分阶段调整示例 celeborn.worker.flusher.threads=16 -> 24 -> 32 每次间隔不低于4小时

经过两年多的生产验证，我们的Celeborn集群已达到以下水平：

  • 日均处理Shuffle数据：6PB
  • 单集群最大节点数：824台
  • 平均Push延迟：1.5ms（P99<5ms）
  • 数据可靠性：99.9999%（年丢数<10MB）

这套体系不仅支撑了混部项目的成功落地，还为公司节省了超过30%的计算资源成本。未来我们将继续在云原生支持（如K8S Operator）和智能调度（基于机器学习预测负载）方向进行探索。