江苏龙网

  1. 首页 > 生活 > >

怎样并行计算 pageRank

spark版本：怎样并行计算 pageRank
【怎样并行计算 pageRank】

■网友
可以尝试一下用hadoop解决。先答到这里，明天抽空贴上源代码。2015年4月23号更新-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------现在我具体说明一下map-reduce的page-rank计算方法。1. 程序说明： （i）输入：txt文件,记录图的信息 其中每一行为： 节点\\t节点初始化PR值\\t节点的链出链表（注：链表以逗号分隔） （ii）输出：txt文件，记录每个节点，以及其收敛的pagerank值2. 实现方法： （i）first-map： 对每个节点，输出一下的键值对： \u0026lt;节点，(链出链表，"LL")\u0026gt; 注：“LL”表明的是值里面记录的是链表 for 链出链表的每一个节点，循环输出下面的兼职对 \u0026lt;该节点链出节点，(PageRank值/链出长度,"PR")\u0026gt; 注：“PR”表明的是平均的PageRank值 (ii) first-reduce: 我们注意到，reduce将map输出的相同key下value组合成一个链表 所以，对相同key的链表中的每一个元素，我们做如下判断： 如果我们找到“LL”,我们提出linklist 如果我们找到“PR”, 我们计算所有平均的PageRank的和sumPR 最后， PageRank值就可以通过(1-q)+q*sumPR计算出来 这里输出 \u0026lt;节点, (节点PR值\\t节点的链出链表)\u0026gt;的键值对 (iii) second-mapper: 迭代计算first-mapper, first-reducer, 收敛后，把mapper结果输出来： \u0026lt;节点，节点PR值\u0026gt;3.项目源代码：package pagerank;import java.io.*;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class MyPageRank {\tpublic static class PageRankMapper\textends Mapper\u0026lt;Object, Text, Text, Text\u0026gt;{\t\tString URL = new String();\t\tdouble PR;\t\tdouble donatePR;\t\t@Override\t\tprotected void map(Object key, Text value, Context context)\t\tthrows IOException, InterruptedException{\t\t\tString line = value.toString().split("\\t");\t\t\tURL = line;\t\t\tPR = Double.parseDouble(line);\t\t\tString linklist = line.split(","); \t\t\tcontext.write(new Text(URL),new Text(line\t\t\t\t\t+"\\tLL"));\t\t\tdonatePR = PR*1.0/linklist.length;\t\t\tfor (String link: linklist){\t\t\t\tcontext.write(new Text(link), \t\t\t\t\t\tnew Text(String.valueOf(donatePR)+"\\tPR"));\t\t\t}\t\t\t}\t}\tpublic static class PageRankReducerextends Reducer\u0026lt;Text, Text, Text, Text\u0026gt;{\tdouble damping=0.85;\tString URL = new String();\tString linklist = new String();\tdouble finalPR;\tint linklistLen;\t@Override\tprotected void reduce(Text key, Iterable\u0026lt;Text\u0026gt; values, Context context)\tthrows IOException, InterruptedException{\t\tdouble sumPR=0.0;\t\tfor (Text val: values){\t\t\tString line = val.toString().split("\\t");\t\t\tif (line.equals("LL")){\t\t\t\tlinklist = line;\t\t\t}\t\t\telse{\t\t\t\tsumPR+=Double.parseDouble(line);\t\t\t}\t\t}\t\tlinklistLen = linklist.split(",").length;\t\tfinalPR = sumPR*damping+1-damping;\t\tcontext.write(key, new Text(String.valueOf(finalPR)+\t\t\t\t"\\t"+linklist));\t\t}\t\t\t} public static class ViewMapper extends Mapper\u0026lt;Object, Text, Text, Text\u0026gt;{ \t@Override \tprotected void map(Object key, Text value, Context context) \tthrows IOException, InterruptedException{ \t\tString line = value.toString().split("\\t"); \t\tcontext.write(new Text(line), new Text(line)); \t} }\tpublic static void main(String args) throws Exception {\t\tint loop;\t\tint iternum = 15;\t\tfor(loop= 0; loop\u0026lt;iternum; loop++){\t\t\tConfiguration conf = new Configuration();\t\t\targs = new String;\t\t\targs=new String("hdfs://192.168.164.128:9000/user/root/in"+\t\t\t\t\tString.valueOf(loop));\t\t\targs=new String("hdfs://192.168.164.128:9000/user/root/in"+\t\t\tString.valueOf(loop+1));\t\t Job job = new Job(conf, "PageRank"); \t\t job.setJarByClass(MyPageRank.class); \t\t job.setMapperClass(PageRankMapper.class);\t\t job.setReducerClass(PageRankReducer.class);\t\t job.setOutputKeyClass(Text.class); \t\t job.setOutputValueClass(Text.class); \t\t FileInputFormat.addInputPath(job, new Path(args)); \t\t FileOutputFormat.setOutputPath(job,new Path(args)); \t\t job.waitForCompletion(true);\t\t System.out.println(String.valueOf(loop+1)+" Time");\t\t}\t\tConfiguration newconf = new Configuration();\t\tString newargs = new String;\t\tnewargs=new String("hdfs://192.168.164.128:9000/user/root/in"+\t\t\t\tString.valueOf(loop));\t\tnewargs=new String("hdfs://192.168.164.128:9000/user/root/Result");\t Job sortjob = new Job(newconf, "PageRank"); \t sortjob.setJarByClass(MyPageRank.class); \t sortjob.setMapperClass(ViewMapper.class);\t sortjob.setOutputKeyClass(Text.class); \t sortjob.setOutputValueClass(Text.class);\t FileInputFormat.addInputPath(sortjob, new Path(newargs)); \t FileOutputFormat.setOutputPath(sortjob,new Path(newargs)); \t sortjob.waitForCompletion(true);\t System.out.println("Ok"); } }4. 最后给作者推荐一点东西，方便理解代码的设计思路 ：(i)《hadoop权威指南》(ii) 还有一本南大的介绍hadoop的书，我觉得写的很好，想起来接着补充哈！！！！！


推荐阅读


上一篇：Socket 和 WebSocket 哪个大并发性能高

下一篇：淘宝店铺天天特价疲劳期多长时间