Giraph通信模块分析

清疚 2022-06-17 22:48 172阅读 0赞

**1 Giraph通信模块**

#### 1.1 模块描述 ####

Hama是遵循BSP模型实现，有着独立的作业分配与任务调度系统， 与Hama不同，Giraph将MapReduce中的Map进行封装，一次作业，即相当于一次bsp作业，实际上相当于一次没有Reducer的MapReduc作业。尽管如此，Giraph并不是简单的主从结构，为了容错考虑，其任务执行过程比较复杂，本文不表，仅关注其消息数据的通信。

像正常的mapreduce一样，作业在提交之后，形成若干分区和mapper的对应关系，一个mapper对应一个或几个分区，每个mapper与一个BspServiceWorker相关，该类有一GraphMapper对象负责图计算，有NettyWorkerClient和NettyWorkerServer对象用来发送数据。Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序（来自百度）。显然，这里的BspServiceWorker即是客户端程序，又是网络服务器。

BSP的运算逻辑被封装在GraphMapper中。GraphMapper包含一个数据实体和一个运算实体。前者为comm.ServerData对象，它的成员包括incoming、 current messages, 聚集值等等；后者是ComputeCallable<I, V, E, M>对象（I-VertexIndexValue, V-VertexValue, E-edgeValue, M-Message）。

每次迭代中，GraphMapper取出上个超步得到的comm.ServerData和图数据传给 ComputeCallable对象进行计算。在该对象中，将完成“计算-通信-同步”的过程。

ComputeCalleble对每一个图顶点以及它的消息、聚集值调用用户定制的compute方法，将产生的消息数据存放到MessageStoreByPartition<I, M>中。本次迭代计算完毕后，所有的的outgoing消息便存储在MessageStoreByPartion对象中。

ComputeCalleble有一个对外通信的成员：workerClientRequestProcessor，该成员负责向远端的worker发送数据，包括消息数据，图顶点数据，边数据等。其中一个重要的方法是flush(), 在本地计算结束后，调用该方法将MessageStoreByPartition中的outgoing消息发送至对应分区。hama使用Hadoop-RPC传输数据，而Griaph使用Netty来传输。

#### 1.2 数据结构 ####

单个消息可以是任何Hadoop的Writable类型，其自身没有目的信息，仅仅表示消息内容。

ServerData包含了众多的数据，与消息相关的有incoming、 current messages,两个队列。incoming队列接收远端发来的消息；current 队列从incoming转换而来，供当次迭代计算使用。

incoming消息组织为两层map结构。图2-1所示。第一层是pairtitionID到发送到该partition消息的映射；第二层是VertexID 到发送给该Vertex的消息队列。

![200236_caxj_916490.png][]

图2-1 Giraph outgoing 消息组织结构

发送消息数据是以RPC方式将outgoing中的消息队列发送至 transientInMessages中，该消息队列的组织为VertexID 到发送给该vertex消息的映射，其结构如图2-2所示。另有一个相同结构的inMessages，在每次迭代计算开始，将transientInMessages转换为inMessages。

![200318_XO1g_916490.png][]

图2-2 Giraph transientInMessages消息队列组织结构

#### 1.3 消息数据的传输 ####

2.1中有过介绍，Griaph使用Netty通信框架在众workers之间交换数据，每台worker有一个workerClient，将本地消息数据包装为NettyRequest，发给对应Worker的NettyServer。从Griaph源码包的主要结构可以反映出来：

comm--  
\--1.aggregator  
\--2.messages  
\--3.netty  
\--4.request  
\-5.WorkerServer.java  
\-6.WorkerClient.java  
\-7.WorkerClientRequestProcessor.java  
\-8.ServerData.java  
\-9.SendMessageCache.java

通信包comm的根目录下有四个重要的接口/抽象类，WokerServer和WokerClient规定了worker作为服务器和客户端的基本行为，如初始化，建立/关闭连接，认证等；WorkerClientRequestProcessor与WokerClient相关，规定了应该向Server发起请求的类型，如sendMessageRequest。SendMessageCache.java是outgoing 消息缓存。

子包messages中是消息实体类，有基于内存和基于磁盘的实现；

子包netty中主要是对5-8的实现；

子包request中主要是各种客户端请求的封装实现。

其他不表。

comm.messages.SimpleMessageStore是对ServerData的一个实现，其中就存放着incoming和current队列。注意这里没有outgoing的实体，每当产生一个消息，就放到与Netty相关的一块发送缓存SendMessageCache里，足够大时，整块发送。

[200236_caxj_916490.png]: /images/20220618/6900313a1e734976b957593fbd39e7c4.png
[200318_XO1g_916490.png]: /images/20220618/420743d83dec4bf7a8eff3234c7083b1.png