面向高性能应用的MPI大数据处理

王 鹏，周 岩

(西南民族大学 计算机科学与技术学院，成都 610225)(*通信作者电子邮箱22954936@qq.com)

0 引言

消息传递接口(Message Passing Interface, MPI)是一种基于信息传递的并行编程接口，在规模化集群环境中具备高效的并行运算与数据处理能力。作为计算密集型的高性能计算标准，MPI在气候模拟、高能物理、生命科学、工业仿真、人工智能、大数据处理等高性能计算和科学计算领域应用广泛[1]。但MPI缺乏与其自身计算系统深度融合的大数据管理机制，目前的数据集中存储模式不便于系统实现“计算向存储迁移”的大数据思想理念。计算过程中数据分发受网络传输时长的制约，限制了MPI在大数据领域的应用扩展。

目前业界在处理海量数据时多采用分布式系统模式，无论是Lustre并行文件系统、并行虚拟文件系统(Parallel Virtual File System, PVFS)，还是通用并行文件系统(General Parallel File System, GPFS)。这些并行化系统各有特点，在不同领域均有相应应用。Lustre的可用性和扩展性不错，但需要特殊存储设备的支持，而且没有实现分布式的元数据服务器管理。GPFS支持串行应用和并行应用，允许任何节点的并行应用同时访问同一文件或不同文件。PVFS将数据文件存储于集群的不同节点，多个客户端可以同时访问这些数据[2]。但这类系统本身都进行过封装，应用程序在调用时只需知道文件逻辑地址，并不关心文件具体存储位置，无法避免运算过程中大量数据的跨节点传输。

作为目前大数据应用领域最为广泛的Hadoop，有一套主从式文件系统HDFS在底层支撑其Map/Reduce的数据处理功能。主从式基础存储和主从式数据处理构成Hadoop的基本架构模型。它的计算系统能够通过HDFS快速定位文件存储位置，这让Hadoop方便地实现“计算向数据存储位置的迁移”，从而大大提高系统计算效率，具备处理超大数据集(large data set)应用程序的能力[3]。

Hadoop采用机制完备的分布式文件系统是为了响应以互联网商业应用为主体的复杂数据处理场景，应对大量数据的频繁读写操作。MPI的应用场景是高性能计算领域，面对的数据应用机制相对简单，通常一次大规模运算需要的源数据相对固定，得出的计算数据相对完整，不需要进行频繁的更新迭代。

在高性能计算领域的应用场景中，扩展MPI针对海量数据的处理能力，是本文研究的主要目标。因此，以最简化的方式，开发设计基于MPI的数据存储组件(MPI Data Storage Plug-in, MPI-DSP)，将其与MPI计算系统有机融合，提高其在文件管理和大数据方面的处理能力，并以实验对组件效果进行初步验证。MPI-DSP适用于一次写入多次读取的应用环境，与大多科学计算的应用场景相吻合，但对于需要频繁写入更新数据的计算任务，并无特别优势，因此希望将其作为现行MPI计算体系的一个扩展和补充。

1 MPI-DSP的设计分析

MPI-DSP的设计目标是希望在高性能计算应用场景中增强MPI的大数据处理能力。与互联网领域应用不同，高性能计算应用往往集中于某一科研领域[4]，有如下特点：1)高频次的文件并发访问相对较少；2)实验源数据相对稳定，不会频繁写入，但数据量较大。根据高性能应用场景的特殊性，明确MPI-DSP的设计原则：1)简化功能设计，避免文件管理的复杂功能，保持系统简洁与轻量级；2)完成“计算向存储迁移”，实现数据在本地读取的核心功能。

采用简洁的方式不影响构建在MPI上的原有应用，能够减少系统研发和维护代价。原有MPI的集中存储机制决定了数据的计算与存储相分离，在MPI程序设计时，程序人员会考虑尽量增加计算操作而减少节点间的通信操作，也就是常说的“计算换通信”设计原则[5-6]。本文在不影响MPI原有系统的情况下，采用轻量级的数据存储组件函数，调整优化MPI现有的数据集中存储模式，使其具备“计算向存储迁移”的能力，减少MPI实际运算时数据在网络中的传输量。

从系统架构分析，MPI-DSP是现有MPI的一个扩展，它与MPI的计算系统紧密结合，这种结合通过开发新的应用程序接口(Application Program Interface, API)函数来实现。新开发的组件使MPI-DSP具备文件切割、分配和调用功能。MPI-DSP在整个系统中的定位如图1所示。

从运行机制分析，MPI丰富的函数调用和灵活的消息传递机制可以让集群各计算节点有条不紊地处理复杂运算任务[7-8]。MPI-DSP充分继承MPI的已有优势，不改变MPI原有函数调用和消息传递机制，通过开发新的功能接口函数，用最简洁有效的方式对数据文件进行存储管理。MPI-DSP在计算任务开始前，将任务所需目标数据以分布式形式分配到各计算节点，当执行计算任务时，在各计算节点的本地读取存储数据，不用跨节点读取，可以有效规避网络传输瓶颈，提高计算任务的处理效率。它借鉴了Hadoop中HDFS的大数据处理模式，同时保持MPI原有的高效运算能力和灵活的消息传递机制[9]。

图1 MPI-DSP系统架构Fig. 1 MPI-DSP system architecture

2 MPI-DSP的实现

具体实现中，分两个阶段进行：第一阶段，文件上传、切分与发送；第二阶段，任务处理与结果汇总。

第一阶段将计算任务所需数据文件上传到专用存储空间。以集群子节点进程数量为依据切分数据文件，目的是平衡各节点的计算与存储负载，减小计算误差。将切片文件分别发送到对应的子节点上。第二阶段是MPI计算任务的核心执行阶段。各子节点根据任务要求读取数据进行计算操作，完成计算任务后汇总结果到主节点。

其中文件的切分、传输和操作是MPI-DSP的核心部分。组件接口函数的设计有三条原则：1)不与现有MPI开发模式产生冲突，实现无缝对接；2)尽可能简化接口函数复杂度和调用方法；3)使MPI对文件的读取清晰明确。

2.1 文件切分

Hadoop的HDFS以64 MB作为文件切块的标准大小，这有其优势，但不够灵活，作业分割力度不够弹性[10]。Hadoop的切分方式并不合适MPI环境：如果源文件不大，切分数量不多而MPI集群子节点数目庞大，将使大量子节点闲置，造成资源浪费；如果源文件过大，将产生大量切片文件，在集群子节点数量有限的情况下会开启大量进程处理切片文件，造成系统拥塞。因此有必要对Hadoop的切片算法进行改进，发挥MPI自有的并行特性，提高整体运行效率。

MPI-DSP以计算节点进程数作为切分依据，文件切分采用等分切割、平均分配的原则，避免了文件切分大小不均匀的问题。根据集群中子节点的进程数量，将大数据源文件连续、平均地分成若干切片文件，充分利用MPI已有的并行性，实现文件操作并行化，确保集群中每个子节点进程分配到相应文件。同时对切片文件进行顺序编号，文件顺序与各计算节点顺序保持一致，可以保障对源文件的有效复原。切分时将切片文件的路径信息存入主控节点数据库，便于各子节点读取调用。文件切分的具体操作通过MPI-DSP中的MPI_Split实现。

实现过程如下：1)主节点通过hostfile查询MPI集群中子节点进程数目；2)找到将要切片操作的源文件，获取源文件大小；3)将源文件大小按子节点进程数均分后设定切片文件大小；4)通过函数get_split_name传入源文件名前缀及切片文件编号；5)将切片文件信息存储到数据库中；6)从源文件中按设定的切片文件大小读取数据写入切片文件中；7)后一次读取位置在源文件前一次读取切片文件大小之后，依次循环；8)直至源文件最后一部分内容写入最后一个切片文件中，完成文件切分操作。

2.2 切片文件传输

MPI-DSP的运行机制是将切片文件在计算任务开始前就分别发送给各计算子节点，有效避免计算节点在执行计算任务时从主节点调用数据文件，以提高整体任务的运行效率。MPI可以通过自有的消息传递机制Send和Recv来发送接收文件，但对于大数据量的文件传输，会产生大量通信，造成通信阻塞，影响系统稳定性和运行效率。为实现切片文件的有效传输，MPI-DSP采用TCP协议的Socket通信方式[11]，以C/S模式，开发两个驻留程序Master和Slave。主节点运行Master，负责数据发送；子节点驻留Slave，负责数据接收[12]。当启动文件传输服务时，这两个驻留程序将被启动执行，分别执行发送和接收操作。

采用Socket方式，一次只能传输1 MB数据，无法应对大文件的传输，通过对分布式存储系统文件传输的优化[13]和断点续传技术[14]可以成功突破Socket对传输文件大小的限制，实现海量数据的传输。Master程序通过自定义的get_trans_count函数获取待传文件(即已切分好的文件)需要传输的次数。文件传输前通过自定义函数get_filename_master将待传文件名称、大小、每次传输数据量及传输次数发送给接收端(子节点)Slave程序。接收端确认收到信息后，主节点的Master程序开始向接收端分批发送数据，直至待传文件全部传输完毕。

2.3 切片文件读取

“计算向存储迁移”的关键点是让计算程序能够快速准确地定位数据文件位置，从而读取文件进行相应操作。因此开发出MPI_Open函数。各计算子节点通过MPI_Open函数提取数据库中的切片文件路径，定位切片文件存储位置，读取切片文件。关键在于，读取的是子节点本地存储的数据，而无需跨节点网络传输。

相应步骤如下：1)MPI_Open函数导入源文件名和文件指针；2)连接数据库，通过源文件名和主机名查找对应切片文件存储信息；3)通过数据库中切片文件路径信息，打开该节点对应的切片文件；4)对该节点文件进行相应计算操作；5)待各节点数据读取、操作完毕，将各节点运算结果发送给主节点，最终由主节点统一汇总输出结果。

MPI_Open函数设计采用二级指针作为参数，利用参数的返回值进行值传递。实现流程如图2所示。

MPI_Open接口函数部分核心代码如下:

void MPI_Open(char *file_name,FILE **fp)

{

…

node_file=(char*)malloc(100*sizeof(char));

MPI_Get_processor_name(node_name,&namelen);

//获取节点名称

MPI_Comm_rank(MPI_COMM_WORLD,&myid);

//获取节点当前进程号

node_name[strlen(node_name)]=′′;

if(myid != 0)

{

get_node_file(file_name,node_name,node_file);

//从数据库中获取所在节点切片文件路径

node_file[strlen(node_file)]=′′;

FILE *fp1=NULL;

fp1=fopen(node_file,"r");

if(fp1==NULL)

{

printf("open %s error ",node_file);

exit(1);

}

*fp=fp1;

}

}

图2 MPI_Open函数流程Fig. 2 Flow chart of MPI_Open function

MPI_Open函数是为MPI提供跨节点操作文件设计的，为实现MPI对海量数据的批处理操作提供依据。设计函数时，充分利用MPI已有的消息传递机制，使用MPI_Get_processor_named函数获取节点名称，MPI_Comm_rank函数获取节点的进程ID，再通过MPI_Open函数调用源文件名称，在数据库中进行信息匹配，获取当前节点切片文件的具体路径。各子节点任务处理完成后，利用MPI原有的MPI_Reduce函数将结果进行归约，主节点输出结果。

3 实验结果及分析

以下通过具体实验验证MPI-DSP的可行性，将MPI-DSP与原有MPI标准模式进行比较，验证在加入数据存储组件后实现“计算向存储迁移”模式的时间优势，以此表明MPI-DSP能够显著提高MPI在大数据下的运行处理效率。

3.1 实验设置

3.1.1 实验环境

实验的硬件环境为：1)8核CPU Intel i7-6700 3.40 GHz，8 MB缓存，内存8 GB，硬盘1 TB。2)5台物理PC，分别为node0～node4，node0为主节点，作为元数据存储节点，同时安装配置MySQL数据库。实验中主节点不参与具体计算任务。node1～node4为计算子节点。

实验的软件环境为：1)操作系统为Linux OS x64 (CentOS 6.8)；2)MPI版本为mpich-3.2.tar.gz；3)MySQL版本为5.7.12；4)API接口函数开发环境为C。

3.1.2 实验代码示例

利用组件中的MPI_Split，将实验文档word.txt切分成4份，平均分配给4个计算节点node1～node4。在计算节点下通过MPI_Open打开split文件夹下存放的切片文件。

char *file_name="word.txt";

//调用文件名

…

MPI_open(file_name,&fp);

//这里的fp是指向计算节点本地的文件指针

MPI_Open实现了fopen的文件打开功能，但打开的是计算子节点本地存储的那部分数据文件。从上面程序中可以看出，接口函数的设计符合设计原则，与现有的MPI完全兼容，函数命名上也符合MPI函数命名规则，函数的运行不会对原有MPI造成任何影响。

3.1.3 实验数据类别

实验数据分两类：第一类采用MPI-DSP组件对实验文件进行切分，分配到各子节点，子节点分别进行Wordcount统计，汇总到主节点输出结果，实验结果以MPI-DSP表示；第二类采用标准MPI方式，实验文件存放在主节点共享目录下，各子节点执行时从主节点共享目录调用实验文件，进行Wordcount统计，结果汇总到主节点进行输出，实验结果以MPI表示。

3.2 实验对比分析

3.2.1 任务总耗时对比实验

第一组实验分别统计MPI-DSP和传统MPI处理计算任务的总时长，包括文件打开时间和运算处理时间。实验结果如表1所示。

表1 任务总耗时对比Tab. 1 Total time-consuming comparison of tasks

从表1可以看出，采用MPI-DSP在Wordcount统计任务的总耗时远远低于传统MPI的处理时长，随着处理数据文件的不断增大，两者皆呈线性增长，但时间差距愈发明显。

3.2.2 任务计算耗时对比实验

第二组实验分别统计MPI-DSP和传统MPI在计算任务上的耗时，结果如表2所示。

表2 计算任务耗时对比Tab. 2 Time comparison of computing tasks

从表2可以看出，在Wordcount统计任务的用时上，MPI-DSP与MPI相比并没有显著差异，表明MPI-DSP并没有影响到MPI原有计算任务的用时。

3.2.3 文件读取耗时对比实验

第三组实验分别统计MPI-DSP和传统MPI读取数据文件的时间,结果如表3所示。

从表3可以看出，随着数据文件大小的不断增加，MPI模式下文件打开读取耗时呈线性增长，而MPI-DSP并没有明显增加。两者明显差异的原因在于MPI从网络读取文件，MPI-DSP从本地调用文件。

通过以上三组对比实验可以看出，MPI-DSP与标准MPI在执行相同文件大小的Wordcount统计运算时，计算统计的时间大致相同，差别在于数据文件的读取时间上。MPI是各子节点在任务开始执行时通过网络从主节点读取运算数据，数据越大，读取耗时越长，主要时间耗费在网络传输上。而MPI-DSP在运算之前已经将数据文件切分并分配给各子节点，开始执行任务时各子节点从本地存储上读取相应数据，时间优势由此体现。这实现了计算向存储迁移的运算模式，与最初设计MPI-DSP的目标相吻合。同时，实验也初步验证了该组件基本功能的可行性。

表3 文件读取耗时对比Tab. 3 Time comparison of reading files

4 结语

MPI-DSP通过简化的函数调用控制方法，实现了MPI计算向数据存储迁移的运算模式，针对高性能计算的应用场景，使MPI既保持原有在高性能领域的计算优势，又扩展了在特定环境下对大数据任务的存储和处理能力。但这套组件有一定局限性，它适用于一次写入多次读取的任务环境，而不适合多次写入与频繁交互的计算任务，因此只能在相应的适用范围内应用。目前该组件只具备基本功能，研究还在继续进行，后续将不断完善，增强其稳定性、易用性和扩展性。

在大数据与人工智能飞速发展的时代，海量数据的挖掘和分析应用需求越来越多。为MPI增加这套MPI-DSP组件，希望能够打破MPI对海量数据和非结构化大数据的操作瓶颈，充分利用其并行化高效计算能力和跨平台灵活性的特点，在特定应用场景下进一步发挥MPI集群的优势。