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

1. 背景与挑战PB级Shuffle的混部需求在vivo大数据平台的演进过程中我们面临着一个典型的基础设施矛盾在线业务如互联网服务和算法服务呈现出明显的昼高夜低流量特征而离线大数据任务则集中在夜间执行。这种资源使用的潮汐现象导致服务器在白天高峰期资源紧张而在夜间利用率不足。2023年启动的混部项目正是为了解决这一矛盾——通过在K8S在线集群中运行大数据离线任务实现资源的错峰利用。然而这种混部架构对Shuffle服务提出了前所未有的挑战。传统Embedded Shuffle ServiceESS存在几个致命缺陷资源隔离差Shuffle过程占用Executor内存容易引发OOM导致任务失败磁盘I/O不可控混部环境下难以保证本地磁盘的独占使用扩展性瓶颈单节点Shuffle数据量超过TB级时性能急剧下降经过对主流方案的深度评估技术选型细节见2.1节我们最终选择Apache Celeborn作为Shuffle服务解决方案。这个决策基于三个关键判断架构匹配度Celeborn的远程服务化设计完美契合混部场景的资源隔离需求性能天花板其Off-Heap内存管理和多磁盘并发机制可支撑PB级数据传输运维友好性完善的Prometheus指标监控体系满足生产环境要求2. 技术选型为什么是Celeborn2.1 主流Shuffle方案横评在Spark生态中Shuffle服务的演进经历了三个阶段原始阶段基于磁盘文件的HashShuffleSpark 1.x优化阶段引入SortShuffle和ESSSpark 2.x服务化阶段Remote Shuffle ServiceSpark 3.x我们对三种主流RSS方案进行了为期两个月的对比测试评估维度CelebornUber RSSFirestormSpark版本支持全版本≤3.0≤3.1动态资源分配原生支持部分支持不支持监控体系Prometheus集成基础指标无内存管理Off-Heap 堆外缓存纯堆内混合模式数据可靠性多副本校验和单副本三副本最大单节点吞吐12GB/s8GB/s10GB/s测试环境配置集群规模10 Worker节点256G内存/24核/12×4T HDD测试工具Spark TPC-DS 10TB数据集网络25Gbps RDMA2.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); } }这种设计带来两个关键收益避免GC停顿测试显示Full GC频率从ESS的每小时5-10次降至零内存效率提升相同数据量下内存占用减少40%磁盘I/O优化Celeborn采用多磁盘轮询小文件合并策略每个Worker管理12-24块物理磁盘通过celeborn.worker.flusher.threads控制每磁盘并发数默认256MB的FlushBuffer减少随机写实测表明该策略在混部环境下仍能保持稳定的磁盘吞吐场景平均IOPS吞吐量延迟(ms)纯Celeborn85002.1GB/s1.2与在线业务混部62001.6GB/s1.83. 性能优化实战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 prio5 os_prio0 tid0x00007f8e5c0e5000 nid0x1e3e 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请求的异步化处理客户端改造// 修改前的同步调用 val stream transportClient.openStream(shuffleId, reduceId) // 修改后的异步调用 val future transportClient.asyncOpenStream(shuffleId, reduceId) future.onComplete { case Success(stream) // 处理流 case Failure(e) // 错误处理 }服务端优化将文件排序操作移至后台线程采用非阻塞IO模型处理请求效果验证优化前后关键指标对比指标优化前优化后提升幅度P99读取延迟4.2s320ms13xWorker CPU利用率85%62%27%↓吞吐量波动率±45%±12%73%↓3.2 小文件内存缓存问题定位通过监控发现小于1MB的文件读取耗时与文件大小不成正比文件大小平均读取耗时主要耗时占比1KB1.8s99% IO等待1MB2.1s95% IO等待10MB120ms30% IO等待实现方案在Commit阶段增加缓存判断逻辑public class FileMemCacheManager { private final ConcurrentHashMapFile, 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.threshold1048576 # 最大缓存比例默认10%堆外内存 celeborn.worker.filecache.maxMemoryRatio0.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.discardThreshold0.8 # 惩罚系数每0.1负载减少的slot数 celeborn.slots.assign.loadAware.penaltyFactor24.2 异常流量控制识别算法采用改进的Z-Score算法检测异常应用对于应用A的Push流量序列X 1. 计算移动窗口(5min)均值μ和标准差σ 2. 计算当前值x的z-scorez (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.4GHz16核/3.0GHz避免低频多核如64核/1.8GHz内存256GB DDR4128GB需调小缓存不要配置NUMA磁盘12×4T HDDJBOD6×8T HDDRAID0绝对避免RAID5网络25Gbps以太网10Gbps需RDMA禁用TCP Offload关键发现磁盘数量比容量更重要12块HDD的聚合带宽~6GB/s优于4块SSD~4GB/s内存带宽是瓶颈在PB级Shuffle场景DDR4-3200比DDR4-2400提升15%吞吐网络协议栈优化启用net.ipv4.tcp_rmem4096 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 done2. 服务指标层核心Grafana看板Push吞吐量分AppFetch延迟百分位P50/P90/P99各Worker的Slot使用率关键告警规则# PromQL示例 rate(celeborn_worker_fetch_failures_total[5m]) 5 and celeborn_worker_disk_usage_ratio 0.93. 业务影响层Shuffle阶段耗时占比失败任务中的Shuffle相关错误码各业务线的SLA达标率5.3 升级与维护滚动升级步骤分批标记Worker为DECOMMISSIONING状态等待存量Shuffle数据被读取完成通过celeborn_worker_active_shuffles指标逐台执行升级保留数据目录验证新版本Worker后加入集群配置变更黄金法则每次只变更一个参数变更后观察至少30分钟重要参数采用渐进式调整# 分阶段调整示例 celeborn.worker.flusher.threads16 - 24 - 32 每次间隔不低于4小时经过两年多的生产验证我们的Celeborn集群已达到以下水平日均处理Shuffle数据6PB单集群最大节点数824台平均Push延迟1.5msP995ms数据可靠性99.9999%年丢数10MB这套体系不仅支撑了混部项目的成功落地还为公司节省了超过30%的计算资源成本。未来我们将继续在云原生支持如K8S Operator和智能调度基于机器学习预测负载方向进行探索。