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方案进行了为期两个月的对比测试：

测试环境配置：

• 集群规模：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减少随机写

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

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) => // 错误处理 }

2. 服务端优化：

• 将文件排序操作移至后台线程
• 采用非阻塞IO模型处理请求

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

3.2 小文件内存缓存

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

实现方案在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的硬件黄金组合：

关键发现：

• 磁盘数量比容量更重要：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）和智能调度（基于机器学习预测负载）方向进行探索。