使用MapReduce分析数据.doc

云计算及应用实验报告 项目名称 使用MapReduce分析数据 专业班级 学 号 姓 名 实验成绩：批阅教师：2016年6 月3 日实验3使用MapReduce分析数据实验学时： 2 实验地点： 实验日期： 一、实验目的1.掌握MapReduce的使用方法2.学会使用MapReduce分析数据二、实验内容1.运行MapReduce的wordcount实例2.使用MapReduce完成二度朋友的实验三、实验方法包括实验方法、原理、技术、方案等。四、实验步骤1. 打开hadoop的各个节点2. 运行hadoop中的wordcount实例3. 编写并编译二度朋友实验的代码4. 将要处理的数据传送到hadoop中5. 运行代码处理数据。五、实验结果记录实验输出数据和结果。六、实验结论1. 创建wordcount_in文件夹2. 创建text.txt文件用来保存要处理的数据3. 将要处理的文件text.txt上传到hadoop4. 运行hadoop-example实例5 查看运行的结果6. 编写二度朋友实验的代码import java.io.IOException;import java.util.Random;import java.util.Vector;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;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.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;public class deg2friend public static class job1Mapper extends Mapperprivate Text job1map_key = new Text();private Text job1map_value = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException String eachterm = value.toString().split(,|t); for(int i=1; ieachterm.length; i+) if(pareTo(eachtermi)0) job1map_value.set(eachtermi+t+eachterm0); job1map_key.set(eachterm0); context.write(job1map_key, job1map_value); job1map_key.set(eachtermi); context.write(job1map_key, job1map_value); public static class job1Reducer extends Reducer private Text job1reduce_key = new Text();private Text job1reduce_value = new Text();public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException String someperson = key.toString();Vector hisfriends = new Vector();for (Text val : values) String eachterm = val.toString().split(t);if(eachterm0.equals(someperson)hisfriends.add(eachterm1);job1reduce_value.set(deg1friend);context.write(val, job1reduce_value);else if(eachterm1.equals(someperson)hisfriends.add(eachterm0);job1reduce_value.set(deg1friend);context.write(val, job1reduce_value);for(int i = 0; ihisfriends.size(); i+)for(int j = 0; jhisfriends.size(); j+)if (hisfriends.elementAt(i).compareTo(hisfriends.elementAt(j)0)/job1reduce_key.set(hisfriends.elementAt(j)+t+hisfriends.elementAt(i);/ public static class job2Mapper extends Mapperprivate Text job2map_key = new Text();private Text job2map_value = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException String lineterms = value.toString().split(t); if(lineterms.length = 3) job2map_key.set(lineterms0+t+lineterms1); job2map_value.set(lineterms2); context.write(job2map_key,job2map_value); public static class job2Reducer extends Reducer private Text job2reducer_key = new Text();private Text job2reducer_value = new Text();public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException Vector relationtags = new Vector();String deg2friendpair = key.toString();for (Text val : values) relationtags.add(val.toString();boolean isadeg1friendpair = false;boolean isadeg2friendpair = false;int surport = 0;for(int i = 0; irelationtags.size(); i+)if(relationtags.elementAt(i).equals(deg1friend)isadeg1friendpair = true;else if(relationtags.elementAt(i).equals(deg2friend)isadeg2friendpair = true;surport += 1; if (!isadeg1friendpair) & isadeg2friendpair)job2reducer_key.set(String.valueOf(surport);job2reducer_value.set(deg2friendpair); context.write(job2reducer_key,job2reducer_value);public static void main(String args) throws Exception Configuration conf = new Configuration(); String otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) System.err.println(Usage: deg2friend ); System.exit(2); Job job1 = new Job(conf, deg2friend); job1.setJarByClass(deg2friend.class); job1.setMapperClass(job1Mapper.class); job1.setReducerClass(job1Reducer.class); job1.setOutputKeyClass(Text.class); job1.setOutputValueClass(Text.class); /定义一个临时目录，先将任务的输出结果写到临时目录中, 下一个排序任务以临时目录为输入目录。 FileInputFormat.addInputPath(job1, new Path(otherArgs0);Path tempDir = new Path(deg2friend-temp- + Integer.toString(new Random().nextInt(Integer.MAX_VALUE); FileOutputFormat.setOutputPath(job1, tempDir);if(job1.waitForCompletion(true)Job job2 = new Job(conf, deg2friend);job2.setJarByClass(deg2friend.class);FileInputFormat.addInputPath(job2, tempDir);job2.setMapperClass(job2Mapper.class); job2.setReducerClass(job2Reducer.class);FileOutputFormat.setOutputPath(job2, new Path(otherArgs1);job2.setOutputKeyClass(Text.class);job2.setOutputValueClass(Text.class);FileSystem.get(conf).deleteOnExit(tempDir);System.exit(job2.wait