[关闭]
@gekeshi 2016-04-30T14:54:33.000000Z 字数 6511 阅读 375

Hadoop wordcount，PageRank实例

Hadoop

运行环境

在Hadoop伪分布式环境下运行，具体环境为：

wordcount example

  1. 在hdfs中新建input目录，将README.txt文件上传到input中
  1. hadoop fs -mkdir input
  2. hadoop fs -put /usr/hadoop-2.6.4/README.txt  input/

Hadoop 运行程序时，输出目录不能存在，否则会提示错误 “org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory hdfs://localhost:9000/user/hadoop/output
already exists” ，因此若要再次执行，需要执行命令删除 output 文件夹
2. 运行自带的wordcount example

  1. hadoop jar /usr/hadoop-2.6.4/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.6.4.jar wordcount input/README.txt output/wordcount

YARN监控过程：
yarnword
输出结果为：
wordresult

PageRank

编写基于mapreduce的PageRank程序如下：

  1. import java.io.BufferedReader;
  2. import java.io.BufferedWriter;
  3. import java.io.IOException;
  4. import java.io.InputStreamReader;
  5. import java.io.OutputStreamWriter;
  6. import java.net.URI;
  7. import java.util.ArrayList;
  8. import java.util.HashMap;
  9. import org.apache.hadoop.conf.Configuration;
  10. import org.apache.hadoop.fs.FSDataInputStream;
  11. import org.apache.hadoop.fs.FSDataOutputStream;
  12. import org.apache.hadoop.fs.FileSystem;
  13. import org.apache.hadoop.fs.Path;
  14. import org.apache.hadoop.io.LongWritable;
  15. import org.apache.hadoop.io.Text;
  16. import org.apache.hadoop.mapreduce.Job;
  17. import org.apache.hadoop.mapreduce.Mapper;
  18. import org.apache.hadoop.mapreduce.Reducer;
  19. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  20. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  22. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  24. public class PageRank {
  26.     public static class PageMapper extends Mapper<LongWritable, Text, Text, Text>{
  28.         private Text averageValue = new Text();
  29.         private Text node = new Text();
  31.         @Override
  32.         //把每行数据的对应节点的分pagerank找出，并输出，当前节点的值除以指向节点的总数
  33.         protected void map(LongWritable key, Text value,
  34.                 Context context)
  35.                 throws IOException, InterruptedException {
  36.             String string = value.toString();
  37.             String [] ss = string.split(",");
  38.             int length = ss.length;
  39.             double pageValue = Double.parseDouble(ss[1]);
  40.             double average = pageValue/(length-2);
  41.             averageValue.set(String.valueOf(average));
  42.             int i = 2;
  43.             while(i<=length-1){
  44.                 node.set(ss[i]);
  45.                 context.write(node,averageValue);
  46.                 i++;
  47.             }
  49.         }
  50.     } 
  52.     public static class PageReducer extends Reducer<Text, Text, Text, Text>{
  54.         private HashMap<String, String> content;
  55.         private Text res = new Text();
  57.         //reducer工作前，key相同的会分组分在一组，用迭代器操作，从总的图中找到所有该节点的分pagerank值
  58.         //利用公式计算该pagerank值，输出。因为下一次要用，因此输出可以凑近一些，把结果都放在value里输出
  59.         @Override
  60.         protected void reduce(Text text, Iterable<Text> intIterable,
  61.                 Context context)
  62.                 throws IOException, InterruptedException {
  63.             double sum = 0.0;
  64.             double v = 0.0;
  65.             for (Text t : intIterable) {
  66.                 v = Double.parseDouble(t.toString());
  67.                 sum = sum + v;
  68.             }
  69.             double a = 0.85;
  70.             double result = (1-a)+ a*sum;
  71.             String sRes = String.valueOf(result);
  72.             String back = content.get(text.toString());
  73.             String front = text.toString();
  74.             String comp = front + "," + sRes + back;
  75.             res.set(comp);
  76.             context.write(null,res);
  78.         }
  80.         @Override
  81.         //reducer的初始化时，先把节点对应文件的数据，存在hashmap中，也就是content中，供每次reduce方法使用，相当于数据库的作用
  82.         //方便查询
  83.         protected void setup(Context context)
  84.                 throws IOException, InterruptedException {
  85.             URI[] uri = context.getCacheArchives();
  86.             content = new HashMap<String, String>();
  87.             for(URI u : uri)
  88.             {
  89.                 FileSystem fileSystem = FileSystem.get(u.create("hdfs://master:9000"), context.getConfiguration());
  90.                 FSDataInputStream in = null;
  91.                 in = fileSystem.open(new Path(u.getPath()));
  92.                 BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
  93.                 String line;
  94.                 while((line = bufferedReader.readLine())!=null)
  95.                 {
  96.                     int index = line.indexOf(",");
  97.                     String first = line.substring(0,index);
  98.                     String last = line.substring(index,line.length());
  99.                     content.put(first, last);
  100.                 }
  102.             }
  103.         }
  104.     }
  106.     public static void main(String[] args) throws Exception{
  108.         //接受路径文件
  109.         Path inputPath = new Path(args[0]);
  110.         Path outputPath = new Path(args[1]);
  111.         Path cachePath = new Path(args[2]);
  112.         double result = 100;
  113.         int flag = 0;
  114.         //制定差值多大时进入循环
  115.         while(result>0.1)
  116.         {
  117.             if(flag == 1)
  118.             {
  119.                 //初次调用mapreduce不操作这个
  120.                 //这个是把mapreduce的输出文件复制到输入文件中，作为这次mapreduce的输入文件
  121.                 copyFile();
  122.                 flag = 0;
  123.             }
  124.             Configuration configuration = new Configuration();
  125.             Job job = Job.getInstance(configuration);
  127.             job.setJarByClass(PageRank.class);
  128.             job.setMapperClass(PageMapper.class);
  129.             job.setReducerClass(PageReducer.class);
  131.             job.setMapOutputKeyClass(Text.class);
  132.             job.setMapOutputValueClass(Text.class);
  134.             job.setOutputKeyClass(Text.class);
  135.             job.setOutputValueClass(Text.class);
  137.             job.setInputFormatClass(TextInputFormat.class);
  138.             job.setOutputFormatClass(TextOutputFormat.class);
  140.             FileInputFormat.addInputPath(job, inputPath);
  141.             FileOutputFormat.setOutputPath(job, outputPath);
  142.             job.addCacheArchive(cachePath.toUri());
  143.             outputPath.getFileSystem(configuration).delete(outputPath, true);
  144.             job.waitForCompletion(true);
  146.             String outpathString = outputPath.toString()+"/part-r-00000";
  147.             //计算两个文件的各节点的pagerank值差
  148.             result = fileDo(inputPath, new Path(outpathString));
  149.             flag = 1;
  150.         }
  151.             System.exit(0);   
  152.     }
  154.     //计算两个文件的每个节点的pagerank差值，返回
  155.     public static double fileDo(Path inputPath,Path outPath) throws Exception
  156.     {
  157.          Configuration conf = new Configuration();
  158.          conf.set("fs.defaultFS", "hdfs://master:9000");
  159.          FileSystem fs = FileSystem.get(conf);
  160.          FSDataInputStream in1 = null;
  161.          FSDataInputStream in2 = null;
  162.          in1 = fs.open(inputPath);
  163.          in2 = fs.open(outPath);
  164.          BufferedReader br1 = new BufferedReader(new InputStreamReader(in1));
  165.          BufferedReader br2 = new BufferedReader(new InputStreamReader(in2));
  166.          String s1 = null;
  167.          String s2 = null;
  168.          ArrayList<Double> arrayList1 = new ArrayList<Double>();
  169.          ArrayList<Double> arrayList2 = new ArrayList<Double>();
  170.          while ((s1 = br1.readLine()) != null)
  171.          {
  172.              String[] ss = s1.split(",");
  173.              arrayList1.add(Double.parseDouble(ss[1]));
  174.          }
  175.          br1.close();
  177.          while ((s2 = br2.readLine()) != null)
  178.          {
  179.              String[] ss = s2.split(",");
  180.              arrayList2.add(Double.parseDouble(ss[1]));
  181.          }
  182.          double res = 0;
  184.          for(int i = 0;i<arrayList1.size();i++)
  185.          {
  186.              res = res + Math.abs(arrayList1.get(i)-arrayList2.get(i));
  187.          }
  189.          return res;   
  190.     }
  192.     //将输出文件复制到输入文件中
  193.     public static void copyFile() throws Exception
  194.     {
  195.          Configuration conf = new Configuration();
  196.          conf.set("fs.defaultFS", "hdfs://master:9000");
  197.          FileSystem fs = FileSystem.get(conf);
  198.          FSDataInputStream in1 = null;
  199.          in1 = fs.open(new Path("output/part-r-00000"));
  200.          BufferedReader br1 = new BufferedReader(new InputStreamReader(in1));
  201.          //这里删除需要打开hdfs在/input目录下的权限操作，非常重要
  202.          //“hdfs dfs -chmod 777 /input”打开权限，这样才可以删除其下面的文件
  203.          fs.delete(new Path("input/test2.txt"),true);
  204.          //建立一个新文件，返回流
  205.          FSDataOutputStream fsDataOutputStream = fs.create(new Path("input/test2.txt"));
  206.          BufferedWriter bw1 = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream));
  207.          String s1 = null;
  208.          //写出并写入
  209.          while ((s1 = br1.readLine()) != null)
  210.          { 
  211.              bw1.write(s1);
  212.              bw1.write("\n");
  213.          }
  214.          bw1.close();
  215.          fsDataOutputStream.close();
  216.          br1.close();
  217.          in1.close();
  218.     }
  219. }

将 Hadoop 的 classhpath 信息添加到 CLASSPATH 变量中，在 ~/.bashrc 中增加如下几行：

export HADOOP_HOME=/usr/local/hadoop
export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath):$CLASSPATH

执行 source ~/.bashrc 使变量生效，接着就可以通过 javac 命令编译

javac PageRank.java

接着把 .class 文件打包成 jar

jar -cvf PageRank.jar ./PageRank*.class

上传测试数据input.txt到hdfs：

A,B,C,D,E,F
B,C,D,E,F
C,D,E,F
D,E,F
E,F
F,A,B,C,D,E

执行程序：

 hadoop jar PageRank.jar PageRank input/input.txt output cache

yarnpage

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注