说明: (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。

InnoDBBufferPool通过缓存数据和索引页来减少磁盘I/O,提升数据库性能。其工作原理包括:1.数据读取:从BufferPool中读取数据;2.数据写入:修改数据后写入BufferPool并定期刷新到磁盘;3.缓存管理:使用LRU算法管理缓存页;4.预读机制:提前加载相邻数据页。通过调整BufferPool大小和使用多个实例,可以优化数据库性能。

MySQL与其他编程语言相比,主要用于存储和管理数据,而其他语言如Python、Java、C 则用于逻辑处理和应用开发。 MySQL以其高性能、可扩展性和跨平台支持着称,适合数据管理需求,而其他语言在各自领域如数据分析、企业应用和系统编程中各有优势。

MySQL值得学习,因为它是强大的开源数据库管理系统,适用于数据存储、管理和分析。1)MySQL是关系型数据库,使用SQL操作数据,适合结构化数据管理。2)SQL语言是与MySQL交互的关键,支持CRUD操作。3)MySQL的工作原理包括客户端/服务器架构、存储引擎和查询优化器。4)基本用法包括创建数据库和表,高级用法涉及使用JOIN连接表。5)常见错误包括语法错误和权限问题,调试技巧包括检查语法和使用EXPLAIN命令。6)性能优化涉及使用索引、优化SQL语句和定期维护数据库。

MySQL适合初学者学习数据库技能。1.安装MySQL服务器和客户端工具。2.理解基本SQL查询,如SELECT。3.掌握数据操作:创建表、插入、更新、删除数据。4.学习高级技巧:子查询和窗口函数。5.调试和优化:检查语法、使用索引、避免SELECT*,并使用LIMIT。

MySQL通过表结构和SQL查询高效管理结构化数据,并通过外键实现表间关系。1.创建表时定义数据格式和类型。2.使用外键建立表间关系。3.通过索引和查询优化提高性能。4.定期备份和监控数据库确保数据安全和性能优化。

MySQL是一个开源的关系型数据库管理系统,广泛应用于Web开发。它的关键特性包括:1.支持多种存储引擎,如InnoDB和MyISAM,适用于不同场景;2.提供主从复制功能,利于负载均衡和数据备份;3.通过查询优化和索引使用提高查询效率。

SQL用于与MySQL数据库交互,实现数据的增、删、改、查及数据库设计。1)SQL通过SELECT、INSERT、UPDATE、DELETE语句进行数据操作;2)使用CREATE、ALTER、DROP语句进行数据库设计和管理;3)复杂查询和数据分析通过SQL实现,提升业务决策效率。

MySQL的基本操作包括创建数据库、表格,及使用SQL进行数据的CRUD操作。1.创建数据库:CREATEDATABASEmy_first_db;2.创建表格:CREATETABLEbooks(idINTAUTO_INCREMENTPRIMARYKEY,titleVARCHAR(100)NOTNULL,authorVARCHAR(100)NOTNULL,published_yearINT);3.插入数据:INSERTINTObooks(title,author,published_year)VA


热AI工具

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

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

Undress AI Tool
免费脱衣服图片

Clothoff.io
AI脱衣机

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

热门文章

热工具

Dreamweaver Mac版
视觉化网页开发工具

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

mPDF
mPDF是一个PHP库,可以从UTF-8编码的HTML生成PDF文件。原作者Ian Back编写mPDF以从他的网站上“即时”输出PDF文件,并处理不同的语言。与原始脚本如HTML2FPDF相比,它的速度较慢,并且在使用Unicode字体时生成的文件较大,但支持CSS样式等,并进行了大量增强。支持几乎所有语言,包括RTL(阿拉伯语和希伯来语)和CJK(中日韩)。支持嵌套的块级元素(如P、DIV),

安全考试浏览器
Safe Exam Browser是一个安全的浏览器环境,用于安全地进行在线考试。该软件将任何计算机变成一个安全的工作站。它控制对任何实用工具的访问,并防止学生使用未经授权的资源。

适用于 Eclipse 的 SAP NetWeaver 服务器适配器
将Eclipse与SAP NetWeaver应用服务器集成。