说明: (1) 实验环境. 三台服务器:test165、test62、test63。test165同时是JobTracker和TaskTracker. 测试例子:官网自带的SSSP程序,数据是自己模拟生成。 运行命令:hadoop jar giraph-examples-1.0.0-for-hadoop-0.20.203.0-jar-with-dependencies.jar o
说明:
(1) 实验环境.
三台服务器:test165、test62、test63。test165同时是JobTracker和TaskTracker.
测试例子:官网自带的SSSP程序,数据是自己模拟生成。
运行命令:hadoop jar giraph-examples-1.0.0-for-hadoop-0.20.203.0-jar-with-dependencies.jar org.apache.giraph.GiraphRunner org.apache.giraph.examples.SimpleShortestPathsVertex -vif org.apache.giraph.io.formats.JsonLongDoubleFloatDoubleVertexInputFormat -vip /user/giraph/SSSP -of org.apache.giraph.io.formats.IdWithValueTextOutputFormat -op /user/giraph/output-sssp-debug-7 -w 5
(2). 为节约空间,下文中所有代码均为核心代码片段。
(3). core-site.xml中hadoop.tmp.dir的路径设为:/home/hadoop/hadooptmp
(4).写本文是多次调试完成的,故文中的JobID不一样,读者可理解为同一JobID.
(5). 后续文章也遵循上述规则。
1. org.apache.giraph.graph.GraphMapper类
Giraph中自定义org.apache.giraph.graph.GraphMapper类来继承Hadoop中的 org.apache.hadoop.mapreduce.Mapper
This mapper that will execute the BSP graph tasks alloted to this worker. All tasks will be performed by calling the GraphTaskManager object managed by this GraphMapper wrapper classs. Since this mapper will not be passing data by key-value pairs through the MR framework, the Mapper parameter types are irrelevant, and set to Object type.
BSP的运算逻辑被封装在GraphMapper类中,其拥有一GraphTaskManager对象,用来管理Job的tasks。每个GraphMapper对象都相当于BSP中的一个计算节点(compute node)。
在GraphMapper类中的setup()方法中,创建GraphTaskManager对象并调用其setup()方法进行一些初始化工作。如下:
@Override public void setup(Context context) throws IOException, InterruptedException { // Execute all Giraph-related role(s) assigned to this compute node. // Roles can include "master," "worker," "zookeeper," or . . . ? graphTaskManager = new GraphTaskManager<I, V, E, M>(context); graphTaskManager.setup( DistributedCache.getLocalCacheArchives(context.getConfiguration())); }map()方法为空,因为所有操作都被封装在了GraphTaskManager类中。在run()方法中调用GraphTaskManager对象的execute()方法进行BSP迭代计算。
@Override public void run(Context context) throws IOException, InterruptedException { // Notify the master quicker if there is worker failure rather than // waiting for ZooKeeper to timeout and delete the ephemeral znodes try { setup(context); while (context.nextKeyValue()) { graphTaskManager.execute(); } cleanup(context); // Checkstyle exception due to needing to dump ZooKeeper failure } catch (RuntimeException e) { graphTaskManager.zooKeeperCleanup(); graphTaskManager.workerFailureCleanup(); } }
2. org.apache.giraph.graph.GraphTaskManager 类
功能:The Giraph-specific business logic for a single BSP compute node in whatever underlying type of cluster our Giraph job will run on. Owning object will provide the glue into the underlying cluster framework and will call this object to perform Giraph work.
下面讲述setup()方法,代码如下。
/** * Called by owner of this GraphTaskManager on each compute node * @param zkPathList the path to the ZK jars we need to run the job */ public void setup(Path[] zkPathList) throws IOException, InterruptedException { context.setStatus("setup: Initializing Zookeeper services."); locateZookeeperClasspath(zkPathList); serverPortList = conf.getZookeeperList(); if (serverPortList == null && startZooKeeperManager()) { return; // ZK connect/startup failed } if (zkManager != null && zkManager.runsZooKeeper()) { LOG.info("setup: Chosen to run ZooKeeper..."); } context.setStatus("setup: Connected to Zookeeper service " +serverPortList); this.graphFunctions = determineGraphFunctions(conf, zkManager); instantiateBspService(serverPortList, sessionMsecTimeout); }依次介绍每个方法的功能:
1) locateZookeeperClasspath(zkPathList):找到ZK jar的本地副本,其路径为:/home/hadoop/hadooptmp/mapred/local/taskTracker/root/jobcache/job_201403270456_0001/jars/job.jar ,用于启动ZooKeeper服务。
2) startZooKeeperManager(),初始化和配置ZooKeeperManager。定义如下,
/** * Instantiate and configure ZooKeeperManager for this job. This will * result in a Giraph-owned Zookeeper instance, a connection to an * existing quorum as specified in the job configuration, or task failure * @return true if this task should terminate */ private boolean startZooKeeperManager() throws IOException, InterruptedException { zkManager = new ZooKeeperManager(context, conf); context.setStatus("setup: Setting up Zookeeper manager."); zkManager.setup(); if (zkManager.computationDone()) { done = true; return true; } zkManager.onlineZooKeeperServers(); serverPortList = zkManager.getZooKeeperServerPortString(); return false; }
org.apache.giraph.zk.ZooKeeperManager 类,功能:Manages the election of ZooKeeper servers, starting/stopping the services, etc.
ZooKeeperManager类的setup()定义如下:
/** * Create the candidate stamps and decide on the servers to start if * you are partition 0. */ public void setup() throws IOException, InterruptedException { createCandidateStamp(); getZooKeeperServerList(); }createCandidateStamp()方法在 HDFS上 的_bsp/_defaultZkManagerDir/job_201403301409_0006/_task 目录下为每个task创建一个文件,文件内容为空。文件名为本机的Hostname+taskPartition,如下截图:
运行时指定了5个workers(-w 5),再加上一个master,所有上面有6个task。
getZooKeeperServerList()方法中,taskPartition为0的task会调用createZooKeeperServerList()方法创建ZooKeeper server List,也是创建一个空文件,通过文件名来描述Zookeeper servers。
createZooKeeperServerList核心代码如下:
/** * Task 0 will call this to create the ZooKeeper server list. The result is * a file that describes the ZooKeeper servers through the filename. */ private void createZooKeeperServerList() throws IOException, InterruptedException { Map<String, Integer> hostnameTaskMap = Maps.newTreeMap(); while (true) { FileStatus [] fileStatusArray = fs.listStatus(taskDirectory); hostnameTaskMap.clear(); if (fileStatusArray.length > 0) { for (FileStatus fileStatus : fileStatusArray) { String[] hostnameTaskArray = fileStatus.getPath().getName().split(HOSTNAME_TASK_SEPARATOR); if (!hostnameTaskMap.containsKey(hostnameTaskArray[0])) { hostnameTaskMap.put(hostnameTaskArray[0], new Integer(hostnameTaskArray[1])); } } if (hostnameTaskMap.size() >= serverCount) { break; } Thread.sleep(pollMsecs); } } }首先获取taskDirectory(_bsp/_defaultZkManagerDir/job_201403301409_0006/_task)目录下文件,如果当前目录下有文件,则把文件名(Hostname+taskPartition)中的Hostname和taskPartition存入到hostNameTaskMap中。扫描taskDirectory目录后,若hostNameTaskMap的size大于serverCount(等于GiraphConstants.java中的ZOOKEEPER_SERVER_COUNT变量,定义为1),就停止外层的循环。外层循环的目的是:因为taskDirectory下的文件每个task文件时多个task在分布式条件下创建的,有可能task 0在此创建server List时,别的task还没有生成后task文件。Giraph默认为每个Job启动一个ZooKeeper服务,也就是说只有一个task会启动ZooKeeper服务。
经过多次测试,task 0总是被选为ZooKeeper Server ,因为在同一进程中,扫描taskDirectory时,只有它对应的task 文件(其他task的文件还没有生成好),然后退出for循环,发现hostNameTaskMap的size等于1,直接退出while循环。那么此处就选了test162 0。
最后,创建了文件:_bsp/_defaultZkManagerDir/job_201403301409_0006/zkServerList_test162 0
onlineZooKeeperServers(),根据zkServerList_test162 0文件,Task 0 先生成zoo.cfg配置文件,使用ProcessBuilder来创建ZooKeeper服务进程,然后Task 0 再通过socket连接到ZooKeeper服务进程上,最后创建文件 _bsp/_defaultZkManagerDir/job_201403301409_0006/_zkServer/test162 0 来标记master任务已完成。worker一直在进行循环检测master是否生成好 _bsp/_defaultZkManagerDir/job_201403301409_0006/_zkServer/test162 0,即worker等待直到master上的ZooKeeper服务已经启动完成。
启动ZooKeeper服务的命令如下:
3) determineGraphFunctions()。
GraphTaskManager类中有CentralizedServiceMaster对象和CentralizedServiceWorker 对象,分别对应于master和worker。每个BSP compute node扮演的角色判定逻辑如下:
a) If not split master, everyone does the everything and/or running ZooKeeper.
b) If split master/worker, masters also run ZooKeeper
c) If split master/worker == true and giraph.zkList is set, the master will not instantiate a ZK instance, but will assume a quorum is already active on the cluster for Giraph to use.
该判定在GraphTaskManager 类中的静态方法determineGraphFunctions()中定义,片段代码如下:
private static GraphFunctions determineGraphFunctions( ImmutableClassesGiraphConfiguration conf, ZooKeeperManager zkManager) { // What functions should this mapper do? if (!splitMasterWorker) { if ((zkManager != null) && zkManager.runsZooKeeper()) { functions = GraphFunctions.ALL; } else { functions = GraphFunctions.ALL_EXCEPT_ZOOKEEPER; } } else { if (zkAlreadyProvided) { int masterCount = conf.getZooKeeperServerCount(); if (taskPartition < masterCount) { functions = GraphFunctions.MASTER_ONLY; } else { functions = GraphFunctions.WORKER_ONLY; } } else { if ((zkManager != null) && zkManager.runsZooKeeper()) { functions = GraphFunctions.MASTER_ZOOKEEPER_ONLY; } else { functions = GraphFunctions.WORKER_ONLY; } } } return functions; }
默认的,Giraph会区分master和worker。会在master上面启动zookeeper服务,不会在worker上启动ZooKeeper服务。那么Task 0 就是master+ZooKeeper,其他Tasks就是workers。

如何使用MySQL数据库进行预测和预测分析?概述:预测和预测分析在数据分析中扮演着重要角色。MySQL作为一种广泛使用的关系型数据库管理系统,也可以用于预测和预测分析任务。本文将介绍如何使用MySQL进行预测和预测分析,并提供相关的代码示例。数据准备:首先,我们需要准备相关的数据。假设我们要进行销售预测,我们需要具有销售数据的表。在MySQL中,我们可以使用

随着大数据时代的到来,数据可视化分析在各行各业中扮演着至关重要的角色。而Go语言作为一种快速、高效、安全的编程语言,也逐渐在数据可视化分析领域占据一席之地。本文将探讨如何使用Go语言进行数据可视化分析。一、Go语言常用的数据可视化库Plotly:可用于在浏览器中创建交互式的图形,支持多种图形类型,如线图、条形图、散点图、热力图等。Gonum/plo

Python语言作为一种高级编程语言,具有简单易学、易读易写等特点,在软件开发领域中得到了广泛的应用。然而,由于Python的开源特性,源代码很容易被他人轻易获取,这就给软件源码保护带来了一些挑战。因此,在实际应用中,我们常常需要采取一些方法来保护Python源代码,确保其安全性。在软件源码保护中,有多种针对Python的应用实践可供选择。下面将介绍几种常见

idea查看tomcat源码的步骤:1、下载Tomcat源代码;2、在IDEA中导入Tomcat源代码;3、查看Tomcat源代码;4、理解Tomcat的工作原理;5、注意事项;6、持续学习和更新;7、使用工具和插件;8、参与社区和贡献。详细介绍:1、下载Tomcat源代码,可以从Apache Tomcat的官方网站上下载源代码包,通常这些源代码包是以ZIP或TAR格式等等。

作为一种流行的服务端语言,PHP在网站开发和运行中扮演着重要的角色。然而,随着PHP代码量的不断增加和应用程序的复杂性提高,性能瓶颈也越来越容易出现。为了避免这种问题,我们需要进行性能分析和调优。本文将简单介绍如何使用PHP进行性能分析和调优,为您的应用程序提供更高效的运行环境。一、PHP性能分析工具1.XdebugXdebug是一款广泛使用的代码分析工具,

PHP代码在浏览器中如何显示源码而不被解释执行?PHP是一种服务器端脚本语言,通常用于开发动态网页。当PHP文件在服务器上被请求时,服务器会解释执行其中的PHP代码,并将最终的HTML内容发送到浏览器以供显示。然而,有时我们希望在浏览器中直接展示PHP文件的源代码,而不是被执行。本文将介绍如何在浏览器中显示PHP代码的源码,而不被解释执行。在PHP中,可以使

vue能显示源码,vue查看看源码的方法是:1、通过“git clone https://github.com/vuejs/vue.git”获取vue;2、通过“npm i”安装依赖;3、通过“npm i -g rollup”安装rollup;4、修改dev脚本;5、调试源码即可。


热AI工具

Undresser.AI Undress
人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover
用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool
免费脱衣服图片

Clothoff.io
AI脱衣机

AI Hentai Generator
免费生成ai无尽的。

热门文章

热工具

螳螂BT
Mantis是一个易于部署的基于Web的缺陷跟踪工具,用于帮助产品缺陷跟踪。它需要PHP、MySQL和一个Web服务器。请查看我们的演示和托管服务。

PhpStorm Mac 版本
最新(2018.2.1 )专业的PHP集成开发工具

VSCode Windows 64位 下载
微软推出的免费、功能强大的一款IDE编辑器

记事本++7.3.1
好用且免费的代码编辑器

Atom编辑器mac版下载
最流行的的开源编辑器