掘金 后端 ( ) • 2024-03-07 17:44
货拉拉基于Apache Uniffle的Tez Remote Shuffle实践
Tez Uniffle 货拉拉

背景介绍

Apache Tez是一种开源计算框架，其独特的优势在于能够将多个相互依赖的作业整合为一个单一作业，从而显著提升作业的性能。在货拉拉，Tez被广泛应用作为主要的离线计算引擎，支撑着我们的离线计算场景和业务需求。

然而随着公司业务规模的持续扩大，我们面临着数据量激增和集群成本急剧上升的挑战。为实现公司的降本增效战略，以及满足大数据云原生与存算分离的发展趋势，我们引入了Remote Shuffle Service（简称RSS）服务。

在接下来的部分，将详细阐述我们在货拉拉实践Tez Remote Shuffle的过程，并分享在这个过程中积累的经验和教训。

业界调研

从2020年Uber开源Remote Shuffle Service方案，业界已经出现了许多成熟的设计和解决方案。为给Tez引擎选择合适的RSS解决方案，我们进行了充分的调研。

业界方案调研

经过对RSS方案的全面调研和比较，考虑到通用性、性能、稳定性和社区活跃度，我们最终选择了Apache Uniffle的RSS架构，并基于此开发Tez客户端。

1. 对Tez源码无侵入：Apache Uniffle架构支持Map端分区Block排序，使Tez的reduce端获取局部有序数据，简化排序过程。而阿里的RSS框架不支持此功能，需对Tez源代码有较多改动，稳定性风险增大。

2. 社区活跃和支持客户端多：Apache Uniffle社区活跃度高，功能迭代快，组织双周会。除了Spark Client，还支持MR Client，对实现Tez Client有参考作用。

Apache Uniffle更详细的介绍：Uniffle: 云原生时代Shuffle新篇章

开源厂商 通用性(支持的客户端) 性能 稳定性 社区活跃 腾讯 incubator-uniffle Spark/MR 1. 在小数据量Shuffle可以和Spark原生Shuffle持平。 2. 在大数据量Shuffle可以比Spark原生Shuffle快30% 3. Benchmark结果链接 1. Server稳定性能力：负载均衡、限流、多副本、Stage失败重试、Quota机制、Coordinator HA等 2. Client改造对Tez侵入性：极低。无需改动Tez源码 Map端同一个Partition的Block内部的数据有序，Block之间无序。可复用Tez原有的Reduce端归并排序。 1. 属于Apache社区：开源地址 2. 用户包括：腾讯，滴滴，爱奇艺，顺丰，唯品会等多家都有日均PB级数据的实践 3. Star数：336；contributors数：66；社区双周例会 阿里 incubator-celeborn Spark Celeborn 在不额外消耗机器资源的情况下，单副本比 External Shuffle Service 性能提升 20%，双副本有 13% 的提升 1. Server稳定性能力：原地升级、拥塞控制、负载均衡等 2. Client改造对Tez侵入性：较高。需改动Tez Reduce的Shuffle逻辑 Map端同一个分区的数据未排序，导致reduce端获取的shuffle数据完全无序。所以Reduce端需要全局排序，改造成本大。 1. 属于Apache社区：开源地址 2022年调研时尚不属于Apache社区项目 2. 用户包括：阿里、网易、Shoppe、丁香园等 3. Star数：729；contributors数：81 字节 CloudShuffleService Spark CSS 与开源的 ESS 使用独占 Label 计算资源进行 1TB 的 TPC-DS Benchmark 测试对比，整体端到端的性能提升 15%左右，部分 Query 有 30%以上的性能提升。 1. 开发成本大：Map端同一个分区的数据未排序，导致reduce端获取的shuffle数据完全无序。所以Reduce端需要全局排序，改造成本大。 2. 没有实现MR Client。 1. 非Apache社区：开源地址 2. 社区不活跃 趣头条 Spark - - 未提到 京东 未提到 - - 未提到

Apache Uniffle在货拉拉的实践#

Apache Uniffle已有Spark、MR的Shuffle Client实现，但没有Tez的Shuffle Client。基于Apache Uniffle，货拉拉设计并与贝壳、Shein一起实现了Tez Client，并提交了若干patch优化Uniffle服务端。

Tez的Shuffle实现没有像Spark那样提供可插拔的ShuffleManager接口，Shuffle代码耦合度高；而且其使用的DAG编程模型因为包含多个Map和Reduce阶段，相比MR的单Map-Reduce模型更复杂，使得Shuffle管理变得更为复杂。

为此，我们对比了Tez、MR和Spark的本地Shuffle差异，并从ApplicationMaster、Shuffle Write和Shuffle Read三个模块上设计Tez Client。

Application Master Shuffle Write Shuffle Read Tez - 处理Session的多个SQL - DAG图 - Reduece Task并发度动态调整 - Shuffle类：根据SQL生成DAG时确认 - Shuffle数据按<PartitionId, Key>排序 - Shuffle类：未抽象接口；和Tez事件处理机制耦合 - Fetch：拉取数据，支持toMemory/toLocalFile - Merge：多路归并排序，spill磁盘 - Final Merge：全局排序 MR - 处理单个SQL - 只包含Map、Reduce - task并发度不会动态调整 - Shuffle类：可配置 - Shuffle数据按<PartitionId, Key>排序 - Shuffle类：可配置 - Fetch：拉取数据，支持toMemory/toLocalFile - Merge：多路归并排序，spill磁盘 - Final Merge：全局排序 Spark - 处理Session的多个SQL - DAG图 - task并发度不会动态调整 - Shuffle类：插件化，可配置 - Shuffle数据只按PartitionId排序 - Shuffle类：插件化，可配置 - Fetch：拉数据toMemory，Spill时局部排序 - Final Merge：多路归并排序，全局有序

Tez Client架构

whiteboard_exported_image (1).png

ApplicationMaster（简称AM）###

模块功能：

  • 和Uniffle服务交互：向Uniffle Coordinator注册Application、申请Uniffle Worker资源
  • 和Map/Reduce交互：给Map/Reduce Task分配Worker，用于Shuffle数据的写和读

技术挑战：

1. 如何做到对Tez源码0修改？ 2. 如何将分配的Worker信息传递给Map/Reduce Task?

解决方案：

1. RssDAGAppMaster继承Tez AM类，扩展与Uniffle Coordinator的交互逻辑；通过调参将AM的主类修改成RssDAGAppMaster，将Tez硬编码的主类DAGAppMaster降级为普通入参

AM启动命令示例，其中RssDAGAppMaster是新的主类，其余内容是它的入参： java org.apache.tez.dag.app.RssDAGAppMaster ... org.apache.tez.dag.app.DAGAppMaster --session

2. 在RssDAGAppMaster中，启动了一个新的RPC服务，用于处理Map/Reduce任务的worker分配请求。通过重载DAG初始化的event handles，将RPC服务地址信息传递给Map/Reduce。详细代码实现可查RssDAGAppMaster代码

Shuffle Write

模块功能：  将Shuffle数据写入到Uniffle Worker服务器

技术挑战：  Tez Shuffle和Tez其它功能的代码耦合度高，如何在不改动Tez代码的情况下，将Shuffle数据写入Uniffle Worker？例如Tez空分区处理机制和Tez task事件处理机制紧耦合

实现方案：  参考Apache Tez代码风格，我们创建了三种Write类（RssOrderedPartitionedKVOutput，RssUnorderedPartitionedKVOutput，RssUnorderedKVOutput），都继承自AbstractLogicalOutput并实现Output接口的start和getWriter方法：

1. 根据DAG生成的edge类型，调用相应的output类； 2. start方法初始化自定义排序器； 3. getWriter方法将K-V数据写入排序器，达到阈值后，将数据转换成Uniffle的Block并发送至Server；

Shuffle Read

模块功能：

  • 向Uniffle Coordinator请求Shuffle数据的Worker地址
  • 从Worker或者远端HDFS读取Shuffle数据，按需排序后返回给上层的Reducer进行逻辑处理

技术挑战：

1. 耦合度高：Tez Shuffle和Tez其它功能的代码耦合度高，可复用性低。需要考虑如何尽可能少的侵入代码和复用已有代码读取Shuffle数据。 Tez根据有序OrderInput和无序UnorderInput内部实现了2个RunShuffleCallalble类，2个类之间逻辑互相独立，在处理排序的逻辑上略有不同。

2. 拉取多Partition数据：与MR不同的是，Tez Reduce可能会同时拉取多个上游Partition的数据。

3. 兼容原有模块：在从Uniffle拉取Shuffle数据的代码修改过程中，我们需要确保兼容并复用原有的事件机制、与Reducer交互的Reader模块、Sort模块。

实现方案：

1. 获取Worker地址：通过在AM侧增加新的RPC接口和现实，Shuffle Read模块请求AM获取Worker地址。

2. 拉取Shuffle数据：通过编写RssOrderedGroupedKVInput和RssUnorderedKVInput，但是调用自定义实现的RssRunShuffleCallable，在该类中根据请求到的Worker地址拉取Shuffle数据。原有本地Shuffle根据每个上游Map Task粒度拉取Shuffle数据，需要修改成Partition粒度拉取多个Partition数据。并兼容和复用已有的event机制、Reader模块、Merge模块。

稳定性保障

Tez Client测试

代码开发完成之后，在上线之前要经过大量测试，通过抓取和回放线上真实业务SQL，开发数据质量比对工具，比较Tez on RSS和本地Shuffle的运行结果、运行效率等，修复发现的20多个Bug，最终通过了个功能测试和性能测试，在线上逐步灰度。包括：

  • 功能测试：验证Tez on RSS能正常运行完，任务不会hang住、task不会OOM等。
  • 数据质量测试：通过研发双跑工具（Remote Shuffle VS Local Shuffle），回放线上SQL对比查询结果
  • 性能测试：Tez on RSS与原有Local Shuffle相比，性能没有明显下降。

Uniffle服务稳定性

  • 监控告警：  接入线上流量之前，我们对RSS Server采集了各项业务和机器的监控指标，依托公司Monitor平台创建监控大盘和指标分级告警规则。 在线上长期运行时，发现有堆外内存缓慢上升的问题，因此提交了监控堆外内存的issue（ISSUE-1189） 当有大规模Shuffle任务或者业务高峰期时，针对可能出现性能瓶颈的方法和操作添加了监控，eg：清理过期任务的资源时，对removeResources()方法的耗时监控(ISSUE-1286)
  • 混沌测试：  在测试环境中，通过模拟单台、多台Worker/Coordinator挂掉，短时间和长时间的服务不可用。发现在服务不可用时，SQL任务因不断重试和不会快速失败导致hang住。 针对Master不可用、Worker不可用，进行了混沌测试，并增加了快速失败的机制，耗时Metric发送改为异步发送，提交Patch反馈社区（ISSUE-1045、ISSUE-1068、ISSUE-1100、ISSUE-1186等）

如下是混沌测试的结果：

大项 事项 期望结果 监控告警预期 是否符合预期 测试Worker挂掉 1. Kill多台Worker进程 1. Task快速失败（5min内） 1. Worker进程探活的监控项告警 2. total_server_num 监控项告警 符合预期 1. Kill Worker进程 2. 间隔5分钟后重启Worker进程 1. Tez任务：重启前提交到该Worker的Tez任务会快速失败（5分钟内） 2. Tez任务：重启后提交到该work的tez任务能正常运行 1. Worker 进程探活的监控项告警 2. total_require_buffer_failed 监控项告警 3. Worker重启后，Worker 进程探活的监控项恢复 符合预期 测试Coordinator挂掉 1. Kill 部分Coordinator进程 1. Tez任务：Application无感知，无报错 2. RSS Worker服务：无异常 1. Coordinator 进程探活的监控项告警 符合预期 1. Kill 部分Coordinator进程 2. 间隔5分钟后重启Coordinator进程 1. Tez任务：Application无感知，无报错 2. RSS Worker服务：无异常 1. Coordinator 进程探活的监控项告警 2. Coordinator重启后，Coordinator 进程探活的监控项恢复 符合预期 1. Coordinator全部挂掉 1. Tez任务：运行中的全失败；后续新提交的也全失败 2. RSS work服务：无异常，used_memory逐渐归0 1. Coordinator 进程探活的监控项告警 符合预期 1. Coordinator全部挂掉 2. 间隔1min重启其中一个协调者 1. Coordinator全部挂掉后： 1. Tez任务：运行中的全失败；后续新提交的也全失败 2. RSS work服务：used_memory逐渐归0 2. Coordinator重启后：提交Tez任务正常 1. Coordinator 进程探活的监控项告警 符合预期 测试Shuffle切换 set tez.shuffle.mode = local 1. Tez任务：使用本地Shuffle运行 2. RSS服务：无异常 - 符合预期 set tez.shuffle.mode = remote 1. Tez任务：使用remote Shuffle运行 2. RSS服务：无异常 - 符合预期 测试Worker优雅上下线 1. 将待下线/上线Worker IP和端口信息写入/移除配置文件：exclude_nodes.conf Tez任务
  • 逐步灰度： 经充分的测试，按照非核心队列、非核心任务，再到重要队列，最后核心队列的顺序，逐步接入线上的请求和流量，以保证线上核心服务的稳定性。

效果

现阶段Tez RSS在非核心队列的逐步灰度过程中，无稳定性、功能性和数据质量等问题，任务执行速度没有明显下降。 在逐步灰度过程中，RSS服务的磁盘使用率逐步上升，同时计算节点的磁盘使用率在下降。所以可通过使用磁盘容量低的计算节点，达到降低部分成本的目的。 从社区经验和理论上分析，使用RSS能达到HQL提效的目的，后续也将分析性能瓶颈，提升HQL在RSS服务上的执行效率和降低成本。

社区贡献

Tez on Uniffle的代码在研发完成并通过数据质量和功能测试后，已成功提交Tez Client模块的代码至Uniffle社区。 同时，我们也将在测试和上线过程中发现和开发的各种特性和补丁，如Fast Fail、Dag级别的数据删除、监控增强等，反馈给社区，促进社区的发展。

未来规划

在完成Tez on Uniffle的代码编写后，现在正处于线上灰度阶段。 计划通过如下措施，提升公司Shuffle服务的稳定性和性能，从而达到降本增效和促进云原生的目的。后续规划包括：

1. 逐步实施灰度接入，并将包括核心队列的所有本地Tez Shuffle替换为Remote Shuffle。

2. 将Shuffle数据写入对象存储。

3. 在Stage失败的情况下，无需重新运行。

4. 我们还将积极和Uniffle社区互动，负责维护Tez Client模块。

Reference

1. GitHub - apache/incubator-uniffle

2. Uniffle: 云原生时代Shuffle新篇章

3. Firestorm - 腾讯自研Remote Shuffle Service在Spark云原生场景的实践

4. [Umbrella] Support Tez Client #321

作者介绍:

  • 柳 晴：货拉拉大数据引擎组 ｜ Apache Uniffle Committer
  • 杨秋吉：货拉拉大数据引擎组 ｜ Apache Uniffle Contributor
  • 张 斌：货拉拉大数据引擎组 ｜ Apache Uniffle Contributor