1、启动hadoop工程
2、MapReduce统计文本单词数量
1public class WordCount { 2 3 private static class WordMapper extends 4 Mapper<LongWritable, Text, Text, IntWritable> { 5 6 @Override 7 protected void map(LongWritable key, Text value, 8 Mapper<LongWritable, Text, Text, IntWritable>.Context context) 9 throws IOException, InterruptedException { 10 11 String string = value.toString(); 12 String[] strs = string.split(" "); 13 for (String str : strs) { 14 context.write(new Text(str), new IntWritable(1)); 15 } 16 17 } 18 19 } 20 21 private static class WordReduce extends 22 Reducer<Text, IntWritable, Text, IntWritable> { 23 // key 单词 //value：{1,1} 24 25 @Override 26 protected void reduce(Text key, Iterable<IntWritable> values, 27 Reducer<Text, IntWritable, Text, IntWritable>.Context context) 28 throws IOException, InterruptedException { 29 30 int count = 0; 31 for (IntWritable value : values) { 32 count += value.get(); 33 } 34 35 context.write(key, new IntWritable(count)); 36 37 } 38 39 } 40 41 public static void main(String[] args) throws IOException, 42 ClassNotFoundException, InterruptedException { 43 44 Configuration configuration = HadoopConfig.getConfiguration(); 45 Job job = Job.getInstance(configuration, "统计单词数目"); 46 job.setJarByClass(WordCount.class); 47 job.setMapperClass(WordMapper.class); 48 job.setMapOutputKeyClass(Text.class); 49 job.setMapOutputValueClass(IntWritable.class); 50 job.setReducerClass(WordReduce.class); 51 job.setOutputKeyClass(Text.class); 52 job.setOutputValueClass(IntWritable.class); 53 54 FileInputFormat.addInputPath(job, new Path("/data")); 55 FileOutputFormat.setOutputPath(job, new Path("/ouput")); 56 job.waitForCompletion(true); 57 System.exit(job.waitForCompletion(true) ? 0 : 1); 58 59 }
2、MapReduce排除文本重复数据
1public class Dup { 2 3 private static class DupMapper extends 4 Mapper<LongWritable, Text, Text, NullWritable> { 5 6 @Override 7 protected void map(LongWritable key, Text value, 8 Mapper<LongWritable, Text, Text, NullWritable>.Context context) 9 throws IOException, InterruptedException { 10 11 context.write(new Text(value), NullWritable.get()); 12 13 } 14 15 } 16 17 private static class DupReduce extends 18 Reducer<Text, NullWritable, Text, NullWritable> { 19 20 @Override 21 protected void reduce(Text key, Iterable<NullWritable> values, 22 Reducer<Text, NullWritable, Text, NullWritable>.Context context) 23 throws IOException, InterruptedException { 24 25 context.write(new Text(key), NullWritable.get()); 26 27 } 28 29 } 30 31 public static void main(String[] args) throws IOException, 32 ClassNotFoundException, InterruptedException { 33 34 Configuration configuration = HadoopConfig.getConfiguration(); 35 Job job = Job.getInstance(configuration, "去重"); 36 37 job.setJarByClass(Dup.class); 38 job.setMapperClass(DupMapper.class); 39 40 job.setMapOutputKeyClass(Text.class); 41 job.setMapOutputValueClass(NullWritable.class); 42 job.setOutputKeyClass(Text.class); 43 job.setOutputValueClass(NullWritable.class); 44 45 job.setReducerClass(DupReduce.class); 46 FileInputFormat.addInputPath(job, new Path("/data")); 47 FileOutputFormat.setOutputPath(job, new Path("/dup")); 48 job.waitForCompletion(true); 49 System.exit(job.waitForCompletion(true) ? 0 : 1); 50 51 }
3、MapReduce实线文本数据的简单排序
1public class Sort { 2 3 private static class SortMapper extends 4 Mapper<LongWritable, Text, IntWritable, IntWritable> { 5 //输出，输入 6 @Override 7 protected void map( 8 LongWritable key, 9 Text value_text, 10 Mapper<LongWritable, Text, IntWritable, IntWritable>.Context context) 11 throws IOException, InterruptedException { 12 13 int value = Integer.parseInt(value_text.toString()); 14 context.write(new IntWritable(value), new IntWritable(1)); 15 16 } 17 18 } 19 20 private static class SortReduce extends 21 Reducer<IntWritable, IntWritable, IntWritable, NullWritable> { 22 23 @Override 24 protected void reduce( 25 IntWritable key, 26 Iterable<IntWritable> values, 27 Reducer<IntWritable, IntWritable, IntWritable, NullWritable>.Context context) 28 throws IOException, InterruptedException { 29 30 for (IntWritable value : values) { 31 context.write(key, NullWritable.get()); 32 } 33 34 } 35 36 } 37 38 public static void main(String[] args) throws IOException, 39 ClassNotFoundException, InterruptedException { 40 41 Configuration configuration = HadoopConfig.getConfiguration(); 42 Job job = Job.getInstance(configuration, "排序"); 43 44 job.setJarByClass(Sort.class); 45 job.setMapperClass(SortMapper.class); 46 47 job.setMapOutputKeyClass(IntWritable.class); 48 job.setMapOutputValueClass(IntWritable.class); 49 job.setOutputKeyClass(IntWritable.class); 50 job.setOutputValueClass(NullWritable.class); 51 52 job.setReducerClass(SortReduce.class); 53 FileInputFormat.addInputPath(job, new Path("/data")); 54 FileOutputFormat.setOutputPath(job, new Path("/sort")); 55 job.waitForCompletion(true); 56 57 }
4、MapReduce实线单表连接
文本数据如下：
child parent
tom lucy
tom jack
lucy mary
lucy ben
1public class Single { 2 3 private static class SingleMapper extends 4 Mapper<LongWritable, Text, Text, Text> { 5 6 @Override 7 protected void map(LongWritable key, Text value, 8 Mapper<LongWritable, Text, Text, Text>.Context context) 9 throws IOException, InterruptedException { 10 11 String string = value.toString(); 12 if (!string.contains("child")) { 13 14 String[] strings = string.split(" "); 15 context.write(new Text(strings[0]), new Text(strings[1] + ":1")); 16 context.write(new Text(strings[1]), new Text(strings[0] + ":2")); 17 18 } 19 } 20 } 21 22 // reduce是执行key的次数 23 private static class SingleReduce extends Reducer<Text, Text, Text, Text> { 24 25 @Override 26 protected void reduce(Text key, Iterable<Text> values, 27 Reducer<Text, Text, Text, Text>.Context context) 28 throws IOException, InterruptedException { 29 30 List<String> left = Lists.newArrayList(); 31 List<String> right = Lists.newArrayList(); 32 33 for (Text value : values) { 34 35 String[] strings = value.toString().split(":"); 36 37 if (strings[1].equals("1")) { 38 right.add(strings[0]); 39 } else { 40 left.add(strings[0]); 41 } 42 } 43 44 for (String lef : left) { 45 for (String rig : right) { 46 context.write(new Text(lef), new Text(rig)); 47 } 48 } 49 50 } 51 52 } 53 54 public static void main(String[] args) throws IOException, 55 ClassNotFoundException, InterruptedException { 56 57 Configuration configuration = HadoopConfig.getConfiguration(); 58 Job job = Job.getInstance(configuration, "单表连接"); 59 60 job.setJarByClass(Sort.class); 61 job.setMapperClass(SingleMapper.class); 62 63 job.setMapOutputKeyClass(Text.class); 64 job.setMapOutputValueClass(Text.class); 65 job.setOutputKeyClass(Text.class); 66 job.setOutputValueClass(Text.class); 67 68 job.setReducerClass(SingleReduce.class); 69 FileInputFormat.addInputPath(job, new Path("/data")); 70 FileOutputFormat.setOutputPath(job, new Path("/single")); 71 job.waitForCompletion(true); 72 73 }
输出结果如下：
grandchild grandparent //额外加入的，表达思路
tom mary
tom ben