设为首页 收藏本站
查看: 1225|回复: 0

[经验分享] 学习日志---基于hadoop实现PageRank

[复制链接]

尚未签到

发表于 2018-10-30 08:29:23 | 显示全部楼层 |阅读模式
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(i0.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

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.yunweiku.com/thread-628231-1-1.html 上篇帖子: Hadoop集群环境安装部署 下篇帖子: hadoop2.7环境的编译安装
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表