基于R的并行优化环境设计

姜海韬 羊帆 陈媛媛 付继发 张辉国

摘 要：文章对R语言已有的并行计算框架进行系统分类、比较，分析了适用于基于R+MPI与基于R+Hadoop的应用场景。在此基础上，结合两者优势提出了一种基于R+MPI的优化计算环境，该环境可以使用户仅修改少量代码就可以将原有的串行程序并行执行。最后，通过一个多元线性回归模型的代码实例展示编程的便捷性，通过与现有工具包的速度比较验证了其有效性。

关键词：代码并行化；分布式算法；MPI；Hadoop；多元线性回归

中图分类号：TP312 文献标志码：A 文章编号：2095-2945（2018）23-0037-04

Abstract： In this paper， the existing parallel computing frameworks in R language are classified and compared， and the application scenarios based on R+MPI and R+Hadoop are analyzed. On this basis， and in view of the advantages of both， an optimized computing environment based on R+MPI is proposed， which enables users to execute the original serial program in parallel only by modifying a small amount of code. Finally， the convenience of programming is demonstrated by a code example of a multivariate linear regression model， and its effectiveness is verified by comparing with the speed of the existing toolkits.

Keywords： code parallelization； distributed algorithm； MPI； Hadoop； multivariate linear regression

引言

由于人们数据收集能力的提高以及“数据即资产”意识的发展，数据集的规模与维度已今非昔比。以使用二维矩阵或者表格储存的数据为例，其属性数（列数）与样本数（行数）都达到了前所未有的数量级，这使得并行编程与分布式存储与计算必然地成为了处理大数据问题的主流解决方案。

R语言是一款基于GNU协议的开源软件，致力于统计计算、数据可视化以及数据分析。该语言拥有庞大的用户社区，在数据挖掘和机器学习盛行的今天发展速度极快。对于研究较多的数学模型，CRAN或GitHub上多有现成的R语言实现。但与此同时，R语言也有自身的不足之处：R语言以单线程运行，不能发挥目前多核处理器的全部计算能力；对于单台计算机内存无法存储的数据集，R语言无法直接处理。这些问题的解决方案之一是将R语言与其他辅助软件结合使用。

Message Passing Interface（MPI）是一个基于Socket的跨语言通讯协议，用于编写高性能的并行程序。调用MPI的R工具包就可以使多个R语言进程间具有数据交换的能力，从而能够分布式地对数据进行高效的计算处理。

Hadoop是一款开源的商业数据库，使用Hadoop Distributed File System（HDFS）储存构架，其良好的效率、可拓展性与高可用性吸引了很多工业界与学术界的科研工作者。基于HDFS的MapReduce计算框架为分布式计算提供了一种备选方案：R语言的接口包rhdfs，rmr允许R调用Hadoop进行海量的数据处理。

由于任务的差异性，目前并不存在统一的、将串行实现的代码并行化的解决方案，因此目前并不存在一个能高效解决一切并行计算问题的“完美系统”。但对于规模相对不太大的数学模型，在能发挥R语言现有优势的情况下提供一个相对统一的解决方案却能大大减少任务并行化的工作量。

本文在探讨目前R语言并行计算框架的基础上，提出一种基于R+MPI的优化并行计算环境，并以实例展示，验证了运用该环境编码的便捷性和有效性。

1 现有的并行框架分析

目前，能够为R语言提供并行计算能力的工具箱大体可分为两类：第一类是以parallel、Rmpi为代表的并行工具包，该类工具箱的实质是打开多个R终端并搭建终端之间的通讯网络；第二类是以rhdfs、rmr为代表的接口包，这类工具箱提供了R语言与其他数据库环境的接口。

一个较为理想的环境需要同时考虑单节点的计算能力、节点之间的数据传输能力、平台兼容性与代码的可移植性，其中单节点的计算能力受制于R语言自身，平台的兼容性受制于使用的具体软件。因此下文将从数据传输能力、代码可移植性两个方面讨论适宜两种工具包的应用场景。

1.1 有关数据传输能力的讨论

讨论串行任务的并行化时，最核心的一个问题就是并行任务的粒度大小。所谓粒度，即将任务并行化时“切分”的细度。对于同一个数学模型，一般粗粒度的并行化方法会对单节点的计算能力有较高的要求，而细粒度的方法却要求结点之间有较强的数据传输能力。

并行工具包的运行模式可以归结为：主节点向各子节点分发数据与需要运行的程序，子节点计算后返回结果，经过主节点整合后得到最终结果[1]。为了达到负载均衡的目的，一些并行函数采用了所有数据傳输均通过主节点的解决方案（如“parallel：：parApply”），易造成主节点通讯管道的堵塞。这类函数更适于粗粒度的、计算量大而传输量较小的任务（比如迭代任务）。对于中间数据传输要求较高的任务，这类函数在数据传输过程中会消耗较多时间，从而影响了运行效率。

而大多数接口包的本质是借用商业数据库的强劲数据处理能力，使R语言能处理远大于单机内存容量规模的数据。目前对于Hadoop平台上的任务并行化已经有较多的研究，如[2-3]，通过R语言接口包rhdfs操作hdfs就可以使用R平台进行数学模型的计算。但商业数据库多基于高访问、低更改的业务特性进行优化，因此这些接口包更擅长于对海量数据进行求和、筛选等具有数据库特点的细粒度操作，却不适合每次都需要将特定数据进行更改的迭代任务。以分布式矩阵乘法为例，Hadoop将原数据切分为大小一定大小（默认为128Mb）的块，以多备份的方式存入HDFS。由于在计算（Reduce）过程中需要整合矩阵中特定的（aij，bjk）进行计算，没有针对矩阵计算的特点进行储存优化的自动分配地址会使shuffle数据处理过程需要较长的时间[5]，这使其计算速度远远不及理论上的并行粒度。

1.2 有关代码可移植性的讨论

R语言拥有庞大的用户社区与开源的可用程序包。同时，R语言又是一种向量化的语言，基本操作为向量/矩阵之间的计算。因此若能做到向量/矩阵计算的并行化，就能尽可能地利用R语言的现有代码，缩短软件的开发周期。

以Parallel为代表的并行工具包已经实现了向量计算的并行化，在不考虑运行效率的情况下，这些包仅通过少量的代码更改即可实现并行版本的向量计算[4]。但目前并没有广泛使用的、用于矩阵并行化计算的软件包。若使用Parallel包中的现有函数，不但要在每次计算前将矩阵切分为列表（list）储存，还必须面对数据传输速度的瓶颈。

文献[5-7]探讨了Hadoop平台上的矩阵操作问题，使用R语言的接口包调用Hadoop进行矩阵计算，并读取计算结果至R中就可以使R语言拥有分布式矩阵计算的能力。但基于此方案的所有代码需要修改为数据库语言，这会使代码的可移植性变弱，难以发挥R语言所拥有的庞大用户社区、海量开源脚本的优势。

1.3 适宜场景小结

使用R语言进行分布式并行处理的两种方法的优缺点浅析见表1，可以发现两种方式各有其优缺点。并行包有较强的代码可移植性，从而能发挥R大量现有工具包的优势，但数据传输能力有待提高；接口包的数据传输能力相对较强，能够突破内存容量的限制，但数据库语言的使用使其代码可移植性降低。

2 并行框架设计

本文搭建的并行环境基于R语言与MPI，由于使用的工具都具有跨平台的优良特性，该环境在Windows与linux操作系统上均可使用，具有很好的操作系统兼容性。考虑到Windows系统是目前被广大高校所使用的，也是开发者保有量最多的环境，且R语言在该系统下可用的分布式环境一直较少，因此后文均在Windows系统上讨论并行环境的搭建与应用。

2.1 基本框架

本环境中的并行计算运行流程可以用图1表示。首先将计算环境初始化，然后将计算所需的矩阵存入环境中。至此就可以对已存入系统的矩阵进行并行迭代计算，而且每一步的计算结果仍然以相同的格式储存在系统中，这样可以做到灵活地调用、输出计算过程中任意一步的结果。体现了本环境在使用过程中的友好性。

对于主从节点拓扑网络的设置，选择二维的矩形排布，其中节点标号以行优先。特别地，为了增加本系统的通用性，对于存入环境的矩阵均按块状划分，所有矩阵行与列的切割策略相同。这是由于块状划分的分布式矩阵计算方法具有较高的执行效率[8]，同时可以保证存入的矩阵无论作为矩阵乘法的被乘矩阵或者乘子矩阵都可以直接进行矩阵计算，而且计算结果还可以继续迭代直接参与计算。避免了每一步计算之前都对矩阵进行切分并重复发送的繁琐操作，为需要迭代或操作较为复杂的任务提供了较多的便利。

2.2 函数设计

本工具箱中的函数分为两类：第一类是基本函数，用于环境的启动、将数据存入/取出并行环境与资源的回收。第二类是附加函数，用于对存入本环境中的数据进行操作，可以根据具体需要灵活增删。以第三节要实现的多元线性回归为例：需要加入矩阵预分配函数、数据传输函数、数据计算函数并包装成统一的并行矩阵乘模板。

基本函数：

（1）初始化函数Mpinit（nrow，ncol）用于mpi环境的初始化、二维网格的设置。用户可以自由设置二维网格的行列结点数量，一共开启nrow*ncol个mpi进程。初始化完成后会返回“done”。

（2）发送函数SendtoSlave（Data，Name），用于将变量分发给各个子节点、存入并行环境。若Data为矩阵，则Name为一个字符串代表储存进入环境中的变量名；若Data为一个列表，则Name也是一个长度相同的列表代表每个变量的名字。运行完毕后会返回成功存入环境中的矩阵数量。

（3）回收函数GetfromSlave（Name），用于从各个子节点回收特定变量名的数据，并以矩阵类型返回。

（4）退出函数mpiexit（），用于释放空间，退出mpi环境。

附加函数（例）：

（1）函数getGrid（Rank，nrow，ncol），在输入子节点编号与网格配置后返回当前结点的二维坐标。（2）函数getPos（Pos，nrow，ncol），在输入子节点二维坐标与网格配置后返回当前结点的编号。

（3）函数DDD（Name1，Name2，Name3），用于分布式计算已经存在于环境中的名为Name1与Name2的矩阵的乘积，结果以名字Name3存入环境。返回值为当次计算所消耗的时间。

3 数值实验

3.1 多元线性回归

回归分析作为统计学的一个重要研究方向，在生物、医学、农业、林业、金融等领域都有广泛应用。而多元回归分析是最为基本的一种回归模型，利用所得到的回归方程，可以评价单个预测变量的重要性，也可以预测一组给定的已知变量所对应的目标值[9]。伴随大数据时代所带来的样本在激增、数据量巨大等问题，使得多元回归分析的应用受到了制约。因此，在并行优化环境中实现回归分析无论是在理论上还是在应用上都有重要的意义。

假设y是可观测的随机变量，共有p个影响因素，分别为x1，x2，…，xp。对于n组独立观测样本：（yi；xi1，xi2，…，xip），i=1，2，…n可以建立如下p元线性回归模型：

在样本量n与属性数p较大的时候，如何计算XT X与XT Y是需要考虑的主要问题。这里在实现了分布式矩阵乘（DDD算法[10]）的前提下，给出本环境中实现多元线性回归的R语言关键代码（见代码1）。

代码1显示了在本环境中实现数学模型的便捷性——只需要将大矩阵存入环境，并用环境自带的矩阵计算函数进行计算，最后从环境中取出计算结果即可。取定n=10000，p=30，運行该代码后观察图2可以发现回归系数的估计值基本以微小的误差均匀分布在真值的两侧，可以认为是噪声ε的影响，回归模型的输出结果是正确的。

3.2 矩阵乘法速度比较

本节通过比较R语言现有的Parallel工具包与本环境中实现的分布式矩阵乘法（DDD算法）来验证本环境的高