学习日志---基于hadoop实现PageRank-创新互联
PageRank简单介绍:
目前累计服务客户近1000家,积累了丰富的产品开发及服务经验。以网站设计水平和技术实力,树立企业形象,为客户提供成都做网站、成都网站制作、网站策划、网页设计、网络营销、VI设计、网站改版、漏洞修补等服务。创新互联建站始终以务实、诚信为根本,不断创新和提高建站品质,通过对领先技术的掌握、对创意设计的研究、对客户形象的视觉传递、对应用系统的结合,为客户提供更好的一站式互联网解决方案,携手广大客户,共同发展进步。其值是通过其他值得指向值所决定,具体例子如下:
第一部分:
对应于每个mapReduce的计算:
由mapper算出每个点所指节点的分值,由reduce整个key相同的,由公式算出。
三角号表示的是迭代两次之间计算的差值,若小于某个值则计算完成,求的每个点的pagerank值。
自我实现的代码:如下
输入的数据分为:
input1.txt
A,B,D
B,C
C,A,B
D,B,C
表示每行第一个点所指向的节点,在reducer的setup会用到,构建hashmap供使用。
input2.txt
A,0.25,B,D
B,0.25,C
C,0.25,A,B
D,0.25,B,C
中间多的数字,表示当前每个节点的pagerank值,其文件可无,因为可以由上面的文件计算生成,有四个节点,即1/4。
自我实现的代码:
package bbdt.steiss.pageRank; import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.IOException; import java.io.InputStreamReader; import java.io.OutputStreamWriter; import java.net.URI; import java.util.ArrayList; import java.util.HashMap; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; public class PageRank { public static class PageMapper extends Mapper{ private Text averageValue = new Text(); private Text node = new Text(); @Override //把每行数据的对应节点的分pagerank找出,并输出,当前节点的值除以指向节点的总数 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String string = value.toString(); String [] ss = string.split(","); int length = ss.length; double pageValue = Double.parseDouble(ss[1]); double average = pageValue/(length-2); averageValue.set(String.valueOf(average)); int i = 2; while(i<=length-1){ node.set(ss[i]); context.write(node,averageValue); i++; } } } public static class PageReducer extends Reducer { private HashMap content; private Text res = new Text(); //reducer工作前,key相同的会分组分在一组,用迭代器操作,从总的图中找到所有该节点的分pagerank值 //利用公式计算该pagerank值,输出。因为下一次要用,因此输出可以凑近一些,把结果都放在value里输出 @Override protected void reduce(Text text, Iterable intIterable, Context context) throws IOException, InterruptedException { double sum = 0.0; double v = 0.0; for (Text t : intIterable) { v = Double.parseDouble(t.toString()); sum = sum + v; } double a = 0.85; double result = (1-a)/4 + a*sum; String sRes = String.valueOf(result); String back = content.get(text.toString()); String front = text.toString(); String comp = front + "," + sRes + back; res.set(comp); context.write(null,res); } @Override //reducer的初始化时,先把节点对应文件的数据,存在hashmap中,也就是content中,供每次reduce方法使用,相当于数据库的作用 //方便查询 protected void setup(Context context) throws IOException, InterruptedException { URI[] uri = context.getCacheArchives(); content = new HashMap (); for(URI u : uri) { FileSystem fileSystem = FileSystem.get(u.create("hdfs://hadoop1:9000"), context.getConfiguration()); FSDataInputStream in = null; in = fileSystem.open(new Path(u.getPath())); BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in)); String line; while((line = bufferedReader.readLine())!=null) { int index = line.indexOf(","); String first = line.substring(0,index); String last = line.substring(index,line.length()); content.put(first, last); } } } } public static void main(String[] args) throws Exception{ //接受路径文件 Path inputPath = new Path(args[0]); Path outputPath = new Path(args[1]); Path cachePath = new Path(args[2]); double result = 100; int flag = 0; //制定差值多大时进入循环 while(result>0.1) { if(flag == 1) { //初次调用mapreduce不操作这个 //这个是把mapreduce的输出文件复制到输入文件中,作为这次mapreduce的输入文件 copyFile(); flag = 0; } Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); job.setJarByClass(PageRank.class); job.setMapperClass(PageMapper.class); job.setReducerClass(PageReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.addInputPath(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); job.addCacheArchive(cachePath.toUri()); outputPath.getFileSystem(configuration).delete(outputPath, true); job.waitForCompletion(true); String outpathString = outputPath.toString()+"/part-r-00000"; //计算两个文件的各节点的pagerank值差 result = fileDo(inputPath, new Path(outpathString)); flag = 1; } System.exit(0); } //计算两个文件的每个节点的pagerank差值,返回 public static double fileDo(Path inputPath,Path outPath) throws Exception { Configuration conf = new Configuration(); conf.set("fs.defaultFS", "hdfs://hadoop1:9000"); FileSystem fs = FileSystem.get(conf); FSDataInputStream in1 = null; FSDataInputStream in2 = null; in1 = fs.open(inputPath); in2 = fs.open(outPath); BufferedReader br1 = new BufferedReader(new InputStreamReader(in1)); BufferedReader br2 = new BufferedReader(new InputStreamReader(in2)); String s1 = null; String s2 = null; ArrayList arrayList1 = new ArrayList (); ArrayList arrayList2 = new ArrayList (); while ((s1 = br1.readLine()) != null) { String[] ss = s1.split(","); arrayList1.add(Double.parseDouble(ss[1])); } br1.close(); while ((s2 = br2.readLine()) != null) { String[] ss = s2.split(","); arrayList2.add(Double.parseDouble(ss[1])); } double res = 0; for(int i = 0;i 注意:
在本地操作hdfs时,进行文件的删除和添加,需要打开hdfs的文件操作权限,
这里删除需要打开hdfs在/input目录下的权限操作,非常重要 “hdfs dfs -chmod 777 /input”打开权限,这样才可以删除其下面的文件打开/input路径的操作权限
第二部分
以上是自己实现的pagerank的算法;下面介绍一下别人的代码
robby的代码实现:
1.首先对节点定义节点类,用于存当前节点的pagerank值以及所指向的节点,存在一个数组中。
package org.robby.mr.pagerank; import org.apache.commons.lang.StringUtils; import java.io.IOException; import java.util.Arrays; //节点类,记录的是当前节点的pagerank值和其指向的节点 public class Node { private double pageRank = 0.25; private String[] adjacentNodeNames; //分割符号 public static final char fieldSeparator = '\t'; public double getPageRank() { return pageRank; } public Node setPageRank(double pageRank) { this.pageRank = pageRank; return this; } public String[] getAdjacentNodeNames() { return adjacentNodeNames; } //接受一个数组,复制在指向节点数组上 public Node setAdjacentNodeNames(String[] adjacentNodeNames) { this.adjacentNodeNames = adjacentNodeNames; return this; } public boolean containsAdjacentNodes() { return adjacentNodeNames != null; } //这个方法是从pagerank值开始+后面的指向的节点 @Override public String toString() { StringBuilder sb = new StringBuilder(); sb.append(pageRank); if (getAdjacentNodeNames() != null) { sb.append(fieldSeparator) .append(StringUtils .join(getAdjacentNodeNames(), fieldSeparator)); } return sb.toString(); } //通过字符串建立一个node public static Node fromMR(String value) throws IOException { String[] parts = StringUtils.splitPreserveAllTokens( value, fieldSeparator); if (parts.length < 1) { throw new IOException( "Expected 1 or more parts but received " + parts.length); } Node node = new Node() .setPageRank(Double.valueOf(parts[0])); if (parts.length > 1) { node.setAdjacentNodeNames(Arrays.copyOfRange(parts, 1, parts.length)); } return node; } }2.这个是mapper的实现
package org.robby.mr.pagerank; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; //这里map的输入时Text和Text类型,说明是两个文本,因此主函数中应设置job的输入类型格式为KeyValueTextInputFormat public class Map extends Mapper{ private Text outKey = new Text(); private Text outValue = new Text(); @Override protected void map(Text key, Text value, Context context) throws IOException, InterruptedException { //先把原始的数据输出,供reduce找指向节点使用 context.write(key, value); //传入时,key是第一个节点,以制表符分割,后面是value Node node = Node.fromMR(value.toString()); if(node.getAdjacentNodeNames() != null && node.getAdjacentNodeNames().length > 0) { double outboundPageRank = node.getPageRank() / (double)node.getAdjacentNodeNames().length; for (int i = 0; i < node.getAdjacentNodeNames().length; i++) { String neighbor = node.getAdjacentNodeNames()[i]; outKey.set(neighbor); Node adjacentNode = new Node() .setPageRank(outboundPageRank); outValue.set(adjacentNode.toString()); System.out.println( " output -> K[" + outKey + "],V[" + outValue + "]"); //这里输出计算出的节点分pagerank值 context.write(outKey, outValue); } } } } 输出的数据:例子 A 0.25 B D B 0.125 D 0.125 注意:
KeyValueTextInputFormat的输入格式(Text,Text),对每行的文本内容进行处理,以第一个制表符作为分割,分为key和value传入。
TextInputFormat的格式是(Longwritable,Text),以行标作为key,内容作为value处理;
3.reduce方法的实现
package org.robby.mr.pagerank; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class Reduce extends Reducer{ public static final double CONVERGENCE_SCALING_FACTOR = 1000.0; public static final double DAMPING_FACTOR = 0.85; public static String CONF_NUM_NODES_GRAPH = "pagerank.numnodes"; private int numberOfNodesInGraph; public static enum Counter { CONV_DELTAS } //reduce初始化时执行的方法,得到总节点个数,在conf对象里 @Override protected void setup(Context context) throws IOException, InterruptedException { numberOfNodesInGraph = context.getConfiguration().getInt( CONF_NUM_NODES_GRAPH, 0); } private Text outValue = new Text(); public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { System.out.println("input -> K[" + key + "]"); double summedPageRanks = 0; Node originalNode = new Node(); for (Text textValue : values) { System.out.println(" input -> V[" + textValue + "]"); Node node = Node.fromMR(textValue.toString()); //这里就是传入的是原始数据 if (node.containsAdjacentNodes()) { // the original node // originalNode = node; } else { //计算针对一个节点的pagerank总和 summedPageRanks += node.getPageRank(); } } double dampingFactor = ((1.0 - DAMPING_FACTOR) / (double) numberOfNodesInGraph); double newPageRank = dampingFactor + (DAMPING_FACTOR * summedPageRanks); //计算差值 double delta = originalNode.getPageRank() - newPageRank; //把原节点对象的pagerank改为新的 originalNode.setPageRank(newPageRank); outValue.set(originalNode.toString()); System.out.println( " output -> K[" + key + "],V[" + outValue + "]"); //把更改后的节点对象输出 context.write(key, outValue); int scaledDelta = Math.abs((int) (delta * CONVERGENCE_SCALING_FACTOR)); System.out.println("Delta = " + scaledDelta); //这个是计数器,mapreduce有很多计数器,自定义的要通过enum对象传入建立和取值 //increment是增值的意思 context.getCounter(Counter.CONV_DELTAS).increment(scaledDelta); } } 4.main函数的实现:
package org.robby.mr.pagerank; import org.apache.commons.io.*; import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.*; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.*; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.*; import java.util.*; public final class Main { public static void main(String... args) throws Exception { //传入输入文件的路径,与输出文件的路径 String inputFile = args[0]; String outputDir = args[1]; iterate(inputFile, outputDir); } public static void iterate(String input, String output) throws Exception { //因为这个是在hadoop上运行的(hadoop jar ...),因此conf会自动配上集群上hadoop的hdfs的入口 //后面的文件可以直接找filesystem,即hdfs的文件操作类 Configuration conf = new Configuration(); Path outputPath = new Path(output); outputPath.getFileSystem(conf).delete(outputPath, true); outputPath.getFileSystem(conf).mkdirs(outputPath); //建立输入文件 Path inputPath = new Path(outputPath, "input.txt"); //建立文件,返回节点个数 int numNodes = createInputFile(new Path(input), inputPath); int iter = 1; double desiredConvergence = 0.01; while (true) { //path建立时,outputpath+后面的是文件路径 Path jobOutputPath = new Path(outputPath, String.valueOf(iter)); System.out.println("======================================"); System.out.println("= Iteration: " + iter); System.out.println("= Input path: " + inputPath); System.out.println("= Output path: " + jobOutputPath); System.out.println("======================================"); //这里进行mapreduce if (calcPageRank(inputPath, jobOutputPath, numNodes) < desiredConvergence) { System.out.println( "Convergence is below " + desiredConvergence + ", we're done"); break; } inputPath = jobOutputPath; iter++; } } //这个类的作用是把file文件的内容加上pagerank值送到targetfile里 public static int createInputFile(Path file, Path targetFile) throws IOException { Configuration conf = new Configuration(); FileSystem fs = file.getFileSystem(conf); int numNodes = getNumNodes(file); double initialPageRank = 1.0 / (double) numNodes; //fs调用create方法根据path对象建立文件,返回该文件流 OutputStream os = fs.create(targetFile); //file文件的流迭代器 LineIterator iter = IOUtils .lineIterator(fs.open(file), "UTF8"); while (iter.hasNext()) { String line = iter.nextLine(); //获取每行的内容 String[] parts = StringUtils.split(line); //建立node对象 Node node = new Node() .setPageRank(initialPageRank) .setAdjacentNodeNames( Arrays.copyOfRange(parts, 1, parts.length)); IOUtils.write(parts[0] + '\t' + node.toString() + '\n', os); } os.close(); return numNodes; } //获取节点数量,也就是获取文件的行数 public static int getNumNodes(Path file) throws IOException { Configuration conf = new Configuration(); FileSystem fs = file.getFileSystem(conf); return IOUtils.readLines(fs.open(file), "UTF8").size(); } //进行mapreduce运算 public static double calcPageRank(Path inputPath, Path outputPath, int numNodes) throws Exception { Configuration conf = new Configuration(); conf.setInt(Reduce.CONF_NUM_NODES_GRAPH, numNodes); Job job = Job.getInstance(conf); job.setJarByClass(Main.class); job.setMapperClass(Map.class); job.setReducerClass(Reduce.class); //输入的key和value都是文本,因此使用这个class,以第一个分隔符作为分割符号,分为key和value job.setInputFormatClass(KeyValueTextInputFormat.class); //map输出定义下 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); FileInputFormat.setInputPaths(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); if (!job.waitForCompletion(true)) { throw new Exception("Job failed"); } long summedConvergence = job.getCounters().findCounter( Reduce.Counter.CONV_DELTAS).getValue(); double convergence = ((double) summedConvergence / Reduce.CONVERGENCE_SCALING_FACTOR) / (double) numNodes; System.out.println("======================================"); System.out.println("= Num nodes: " + numNodes); System.out.println("= Summed convergence: " + summedConvergence); System.out.println("= Convergence: " + convergence); System.out.println("======================================"); return convergence; } }注意:
这个是文件流操作的方法,使用 import org.apache.commons.io.IOUtils中的IOUtils类中的方法。
还有一个Arrays方法copyOfRange,可以返回数组的指定位置,返回一个数组
OutputStream os = fs.create(targetFile); //file文件的流迭代器 LineIterator iter = IOUtils .lineIterator(fs.open(file), "UTF8"); while (iter.hasNext()) { String line = iter.nextLine(); String[] parts = StringUtils.split(line); Node node = new Node() .setPageRank(initialPageRank) .setAdjacentNodeNames( Arrays.copyOfRange(parts, 1, parts.length)); IOUtils.write(parts[0] + '\t' + node.toString() + '\n', os); }使用readLines方法,返回的是一个String数组,每个单元里放的是每行的内容
IOUtils.readLines(fs.open(file), "UTF8").size();TextOutPutFormat的输出的键值对可以是任何类型,输出是自动调用toString方法,把对象转为字符串输出。
使用stringUtils,截字符串为数组
String[] parts = StringUtils.splitPreserveAllTokens( value, fieldSeparator);另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。
本文名称:学习日志---基于hadoop实现PageRank-创新互联
网站链接:http://pwwzsj.com/article/djishi.html