| 1 介绍 SortBenchmark作为大规模数据排序的评价基准，已经被众多知名公司和科研机构所认可，并在每年4月展示世界上在排序算法研究和应用上取得的最好成果。SortBenchmark主要由六种Benchmark组成，其中被Yahoo，微软等知名公司和高校所公认的TeraSort Benchmark已经由09年合并到MinuteSort Benchmark，用于评价单位分钟内所能处理的数据规模。因此，根据MinuteSort Benchmark得到成绩便可被用来评价当今在大规模数据排序应用中运算最快的系统，以及对比系统之间的差距。 本文介绍Typhoon（台风）平台上应用MapReduce计算框架来实现并运行MinuteSort Benchmark所取得的成绩以及所使用的优化方法。 2 系统介绍 Typhoon（台风）是由基础架构部开发的集分布式存储和分布式计算于一体的云计算平台。它由提供高可靠性的分布式文件系统XFS[3]、分布式半结构化存储系统XCube，分布式计算框架MapReduce，以及机群调度系统Torca等组件构成。作为实现MinuteSort最相关的两个组件，下面主要介绍MapReduce计算框架以及XFS文件系统及其特点，其他组件可以参考。 2.1 MapReduce计算框架 MapReduce框架设计思想来源于传统单机系统中的多线程处理模型，它是这种模型在分布式操作系统(Torca)上的实践。在Typhoon系统中，基于MapReduce框架实现的应用程序会被Torca调度到分布式集群上，并以一个Master和一组Worker相互协作的方式运行。Master承担了任务管理和调度的职责；而这组Worker则像单机中的多个线程一样承担了计算处理的职责；而类似于单机环境下的存储设备（如磁盘文件系统和数据库）则对应于Typhoon平台中的XFS和XCube。 图1 MapReduce计算机框架图 计算框架有如下特点： Master对于Task的下发，支持随机、本地优先、负载限制等多种调度方式。对于不同的数据源和数据分布都能有效地支持。优先考虑本地调度，利用本地读取的特性，加快IO速度。同时，还考虑节点间处理能力严重不均衡的问题，采用负载限制调度。 框架对Worker和Master均提供容错支持。框架能够将临时失败的Task完成的数据进行恢复，以及对不可恢复的Task进行重新下发，从而保证任务的运行完成。而单点存在的Master，则会定期在XFS记录CheckPoint，当进程被迁移时，利用CheckPoint进行状态的恢复，从而继续完成作业。 支持多数据源。除了原生支持XFS和XCube数据外，用户还可以自定义自己的输入输出流处理接口，来处理特定格式的数据源。 2.2 XFS文件系统 XFS分布式文件系统提供高可靠和高性能的数据存储服务。XFS按全局目录树方式组织文件，管理超大容量的硬盘。XFS中文件一般保存三份副本，应用不必关心数据存放的具体位置，不必关心单个磁盘故障或机器宕机等，由XFS保障数据稳定可靠、随时可用。 XFS分为三层系统架构（如图2所示），最上层是Master，管理文件的目录树结构；中间一层是MetaServer，记录每个文件块的实际位置；最下面一层是NodeServer，负责实际文件数据的读写。此外，提供Scheduler完成辅助工作，包括收集统计各NodeServer上空间状况，以及备份和扫描数据等后台功能。XFS的SDK提供了统一的接口和全局唯一的路径解析标准，支持以标准接口接入不同的文件系统。XFS的Master和MetaServer都依赖Zookeeper提供的分布式锁服务来选举主节点，SDK也依赖Zookeeper获取相应机群的Master IP。 图2 XFS文件系统的结构图 XFS由单主控节点（Master）+多组元数据服务器（MetaServer）+多数据节点（NodeServer）组成。Master提供了统一的目录结构，和数据均衡分布等重要决策；MetaServer分担Master的元数据的读写请求和内存压力；NodeServer则提供高性能的数据读写服务。XFS具有下述特点： 可扩展的分层元数据管理：XFS目录树结构仍然在单一的Master节点，而文件块存储位置等放到多组MetaServer中，因此MetaServer可水平扩展，以分担Master的内存压力和性能压力。XFS每次写数据的时候，只更新MetaServer，可大幅分担Master的性能压力。 内建的高可靠的元数据管理：XFS将元数据管理的高可靠和高可用目标作为基本的内建要求，贯穿整个设计。Master利用两阶段提交协议实现元数据实时热备，利用Zookeeper提供分布式锁服务来选举主Master，并自动完成滞后Master的数据对齐，实现了高效和自动的元数据高可靠和高可用。同时，每组MetaServer通过主备实时热备保证高可靠。基于上述主备机制，XFS支持rolling update，系统升级不中断服务。 XFS提供append语义，支持已关闭文件重新打开追加数据，但不支持随机写。XFS避免lease导致的较长不可写时间，而通过可抢占的序列号保证多个冲突的写客户端中仅最后一个成功。 XFS完全采用C++开发，以利于性能调优和精确的内存管理。 3 实验 3.1硬件和操作系统 我们在Typhoon tjd2-websearch做MinuteSort benchmark的测试。tjd2-websearch包含大约950个节点： 每个节点8 core (超线程) Intel(R) Xeon(R) CPU E5640 @ 2.67GHz 每个节点有64G RAM 每个节点有12块1T 硬盘（SATA） 任何两个节点之间保证1G带宽 每个机架10-20台机器 操作系统: 2.6.32.43-tlinux-1.0.0-container gcc (GCC) 4.5.1 XFS分布式文件系统， Torca集群调度系统共同部署在这个集群上。 MapReduce任务包含一个master 作业和一个worker作业。Master作业包含一个master任务，对应一个MapReduce master进程，它申请资源为： 2 core 4G RAM 1 块磁盘 Worker作业包含1850个任务，每个任务对应一个MapReduce的worker进程， 申请资源为： 1 core 4G RAM 8块磁盘 网络带宽由所有master和worker共享。 3.2实验结果 在1850个worker（每个机器大约有两个worker），1800个reduce，1T sort 耗时53s，其中map阶段大约20s, map与reduce阶段的间隔：10s，Reduce时间大约为23s。运行中的map和reduce任务个数随时间变化如下图： 图3 Map/Reduce任务变化图 与google，microsoft，hadoop等已发布的测试结果对比如下： 表1 与最近几届MinuteSort的世界记录的结果对比[1]
表2 与08年至今TeraSort的世界记录的结果对比
从上表可看出：Typhoon MapReduce系统已经成功在1分钟内完成TB级别的数据处理。在性能上仅次于今年Microsoft发布的1.4TB / 60s的成绩。但是，Microsoft取得该成绩所运行的Flat Datacenter Storage是经过定制的非通用平台[8]。因此，相比于应用在通用平台的Typhoon MapReduce，我们所取得的成绩更接近实际场景更有说服力。 4优化 4.1 XFS读写性能优化 基于locality的数据读写 XFS中数据写操作按NodeServer形成流水线方式写，并尽量选取SDK所在节点作为其中一个NodeServer，以减小数据写过程中的网络带宽消耗。 XFS数据读取过程中，如果SDK所在的节点是当前chunk所在的某个NodeServer，也大幅减少带宽消耗。 p2sp模式的数据并发读 XFS中SDK读数据的时候，把逻辑上需要读的一大块数据分成小块的请求发给不同的Nodeserver，Nodeserver从磁盘读指定区间的数据，返回给SDK。SDK收集到所有分片的数据，合并成逻辑上的完整一块数据返回给用户。如果某个Nodeserver暂时无法访问，SDK则把对它的读请求重新发给可用的Nodeserver。 节点分配策略调优 XFS面向的网络环境中，机群中任意两点的带宽和延迟相同，因此节点分配策略调优的目标中，除了要求跨机架保障数据可靠性，以及基于locality减少网络带宽消耗，最重要的优化是做NodeServer间的负载均衡，以保障整机群有良好的总数据吞吐量。 高效的元数据操作 XFS对元数据操作的性能进行大量优化，以提升整机群读写吞吐率，并大幅提高元数据操作密集类应用的性能。Master对两阶段提交协议进行批量操作、piggyback提交请求、流水线和预测执行等优化。Master对目录树基于LSM-tree提供高效的元数据读写，支持不停机完成checkpoint，并大幅优化checkpoint性能以维持较小的增量目录树和减少内存占用。 控制流的QoS保障 云计算的存储服务器(XFS)和计算服务器是共用的，云计算之上的众多用户任务执行时会出现一些随机的多个不同的任务读或写某些相同的node server，引起fan-in,fan-out。造成网络带宽竞争加剧，带宽流量效率低下，降低整体性能。 XFS通过续读和续写以复用部分成功的读写请求，减少超时导致的流量浪费。部署在网络架构4.0上的XFS还进一步在交换机侧引入QoS保障，对控制流设置高优先级，保障控制流顺畅以协调读写减少超时。IDC网络架构4.0为服务器打造了一个规模化、高效的互联矩阵，可以保证IDC内任意两台服务器之间互联带宽可以达到1Gbps，也就是说任意两台服务器之间的互联带宽相当于两台服务器直接相连的带宽；同时，通过配合高精准度的控制流机制，提高了整个XFS集群的稳定性。 4.2 Map任务长尾优化 在map 任务阶段，发现大量的map task（每个map读取数据量大小64M，与XFS的一个chunk的大小保持一致）的运行时间在1-2s内。这些map 任务都是由于本地化优先的调度策略，被调度到数据源所在的机器上。与XFS本地化优先读的策略配合，这些map 任在执行的时候直接从本地XFS NodeServer读取数据，从而减少网络开销。 图4 任务的运行时间图 但是存在两种类型的慢的map任务。第一类是在这个map阶段后期出现的运行时间8-10s中的map任务。对比正常map任务，发现这类map任务是由于在map任务调度的后期，一些worker执行速度快，或者数据在其上map 任务相对比较少，这些worker在执行完数据在其上map任务后，随机执行数据源在附机器上的map 任务。这些map在执行时，由于数据源不在本地，必须从远程的XFS的NodeServer读取数据，从而导致这个map任务运行时间变长。由于XFS系统在分配chunk给那个机器的时候考虑到数据源的均匀分布，所以相对来说每个机器上分配的数据量大小相差不大，因此在MinuteSort的时候修改调度策略，取消调度后期从执行数据源在其他机器上的map任务， 保证每个map task都是从本地XFS NodeServer上读取数据， 减少这类长尾的map task。 第二类慢的的map 任务执行时间在60s左右，出现的时间点没有明显规律。在观察运行日志发现这些map任务在执行过程都遇到的从NodeServer读取数据超时的现象，而XFS SDK默认情况下从NodeServer读取超时时间为60s，因此map任务运行大部分时间都是在等待XFS SDK超时。修改这个默认超时时间为5s，就大大减少超时等待时间，减少map阶段的长尾。 4.3 Shuffle策略 调节下发Shuffle任务窗口 将Master端下发Shuffle任务的窗口大小由1000调整为5000。在worker数目较多的情况下，一个心跳周期内能够拿到的Shuffle数目小于一个心跳周期内能够完成Shuffle的数目，对性能产生了影响。调整参数后Shuffle的时间显著减少。 加入基于IP的Shuffle Server惩罚机制 当Shuffle Client从Shuffle Server拉取Shuffle失败之后，会对Shuffle Server进行惩罚，在一段时间内不再从此Shuffle Server拉取Shuffle数据，避免Shuffle Server请求过多而带来的性能下降。并且基于IP的惩罚主要目的是针对Worker发生迁移的情况，当Worker从IP1迁移到IP2时，IP1上的Shuffle数据不可再用，但是IP2上的Shuffle数据可用，如果不按IP进行区分，则会造成惩罚不当。 加入甄别Shuffle Client的机制 当Shuffle Client拉取Shuffle数据失败过多时，包括Shuffle Server失败数目过多，Shuffle Task失败数目过多等，Shuffle Client会主动上报Reduce Task失败，使得Reduce Task重新被Master调度。这样可以有效解决因为Shuffle Client因为网络故障，或者磁盘问题等引起的Reduce Task 长尾。 加入检查Shuffle失败原因机制 在Shuffle Client加入Shuffle失败原因的检查，区分网络问题或者Shuffle Server的原因。当Shuffle Server发生错误时（磁盘读写，Shuffle Data不存在等），直接上报Master Shuffle Task失败，使得Master重新调度对用的Map Task；如果是网络原因，则会通过基于IP的Shuffle Server惩罚机制，延期调度此Shuffle Task。另外如果发现Shuffle Client发生类似于磁盘不足的情况，则直接上报Reduce Task失败。 4.4性能优化结果 经过上述的优化，我们分别获得如下的优化结果： XFS性能 以81台NodeServer的XFS测试机群为环境，千兆以太网互联。XFS多客户端的读写性能如下（不启用localize模式，数据写3副本）： 表3 XFS读写吞吐性能
从上表可知，整机群每秒写入2.0GB数据，产生的网络流量为6.0GB，平均每台NodeServer入口网络流量75MB/s。读数据中整机器每秒读8.9GB数据。 表4 XFS的元数据操作性能
长尾作业优化 分析了上述两种慢的map任务的原因后，经过优化map阶段的长尾基本消除，整个map阶段可以在20内完成。如图6所示。 图5 优化后的任务运行时间图 5结论 基于Typhoon平台以及MapReduce框架实现的MinuteSort是用来评价整个Typhoon系统数据处理能力的一个benchmark。它充分利用了XFS数据存储的特点以及高效的执行操作来实现数据的读写，并通过调整MapReduce框架在运行过程所使用的参数来消除长尾和Shuffle策略来进行优化，从而实现在53秒内完成对1TB数据的排序，仅仅次于微软2012年所取得的1分钟完成1.4TB的数据排的成绩。 |
