Hama框架学习(一) 从源码角度分析job的提交和运行过程

作者：王连平

如有转载，请注明文章出处：/article/1975728.html

首先，简单介绍一下hama，hama是一个并行计算框架，它是对经典的BSP模型的开源的实现，该计算框架与HDFS结合，同时在其之上开发了Graph包，实现了Pregel分布式图数据计算模型。该计算框架主要包含三个部分，BSPMaster/GroomServers计算引擎、Zookeeper分布式协同管理以及HDFS/HBase文件存储系统。

虽然之前知道Hama的运行机制，但是只是理论上的理解，最近几天从源码出发对Hama的job提交和运行过程进行了学习，对Hama的整体框架有了更深刻的认识。首先Hama源代码的整体结构可以分成hama-core、hama-exanples、 harm-graph、hama-parent、 hama-yarn五部分。Hama-core是hama的核心，其中包含了hama的计算框架，Hama-graph则是hama用于处理分布式图数据的代码，这里主要实现的是Pregel计算模型。Hama的Job提交机制为JobClient将作业提交给BSPMaster，之后BSPMaster将任务进行分配，在集群的GroomServers上运行BSPPeer，他们之间协同地完成分布式任务。理论上的理解很多关于Hama框架的文章中会提到，因此，这里不再赘述，直接分析源码。

本文以SSSP程序为例，这里要涉及到Hama的graph包。

1. 首先，看下SSSP的提交作业部分，如下：





上图中主要是SSSP中对Job的一些参数的设置，之后是提交作业，画红线的部分为作业提交。

2. 用户应用程序提交过作业后实际上是把任务交给了BSPJob，下面介绍在BSPJob中的Job提交的几个相关方法，如下：



这个函数中，可以看到它调用了自身的submit( )函数，这里有几个状态的判断,具体的判断内容在图中已经标记出来。那么接下来我们要分析BSPJob的submit( )函数了，先看下源码：




从上述源码中可以看到，这时BSPJob实际调用了BSPJobClient实例对象，这个实例对象是专门用来提交作业的客户端，其中传入的参数为BSPJob本身。接下来我们要进入BSPJobClient这个对象一探究竟，到底是如何提交作业的。

2. BSPJobClient对象中有关Job提交的函数

1>首先，先看系BSPJobClient.submitJob函数，源码如下所示：




这里可以看出，又多了一个对象---jobSubmitClient，这个对象又是干什么的呢，它的来头可不小，是在作业提交中起到关键性作用的对象，在此函数中我们只看到它创建新JobID的功能，冲这点是不是能给你带来点启发呢？对此对象的分析放到下文，这里暂不做分析。那么下来我们将要看到的是BSPJobClient.submitJobInternal这个函数，如下：









从这个代码量上来看，这个函数肯定在作业提交时起到了不小的作用。为了搞清楚该函数到底做了哪些任务，我们这里分步骤来讲。

①通过jobSubmitClient对象来获得集群状态对象，并且从配置的环境文件中获得每个Job可以分配的最大任务数目。该任务数可能没有配置，这时用集群中最大的任务数减去当前已经开启的任务数目。

②接下来设置job的文件目录，这里包括了jar、xml等详细的请看源码。

③然后，判断是否有输入目录，若有进行分区。此部分将在另外一篇博客中专门讲hama输入分区的；如果没有则将其按照普通的job提交。所以，值得一提的是，对于有输入数据的程序，一般都会创建两个job，一个为分区job，另一个为真正的job。同时，从源码中可以看到，对输入splits的数目以及集群还能开启的最大任务数有个对比，如果前者比后者大会抛出异常，这里的原因是因为hama在进行分区时一般的对每个split都会启动一个task。

④在文件系统中创建job的文件目录，这里的文件系统一般是hdfs，因为hama是架构在hadoop之上的。其中主目录为submitJobDir，在这个目录中的文件包括job.xml、job.jar、job.split。实际上这些目录是临时的，当整个job完成后这些文件将要被删除。创建完成后，将文件上传到文件系统中，以供所有的groomserver下载到本地。

⑤下面一切准备工作都就绪了，到真正启动任务的时候了launchJob(jobId, job, submitJobFile, fs)。



2>接下来我们看下BSPJobClient.launchJob函数，这个函数具体内容如下：



这里，不得不说一下神秘的jobSubmitClient这个对象了，前面已经提到了这个对象，那么我们来找一下这个对象的定义和赋值就一目了然了。

定义




从上面代码中可以看到，jobSubmitClient这个对象的赋值有两种情况。第一种，当用户配置的“bsp.master.address”的值为“local”时，即为hama为本机调试版本时，jobSubmitClient的值为LocalBSPRunner；第二种情况，当不是本地调试版本时，jobSubmitClient的值为远程对象，这个对象是从RPC.getProxy( )获得的，可以看到这个函数的一个参数是JobSubmissionProtocol.class，也就意味着获得的是远端的JobSubmissionProtocol对象。通过查看源码，我们知道BSPMaster这个类继承了JobSubmissionProtocol这个类，那么通过RPC获得的对象是BSPMaster对象，这个对象的地址从哪里获取的呢？是RPC.getProxy( )从conf中获取的。

至此为止，客户端的在job提交过程中的任务已经完成，这里已经将job的所有信息和文件上传到了dfs中，并且调用了远端的BSPMaster代理对象，进行Job提交。后面的内容将在下一篇博文中看到。