共同朋友

1. import java.io.IOException; 
2. import java.util.Set; 
3. import java.util.StringTokenizer; 
4. import java.util.TreeSet; 
5.  
6. import org.apache.hadoop.conf.Configuration; 
7. import org.apache.hadoop.fs.Path; 
8. import org.apache.hadoop.io.Text; 
9. import org.apache.hadoop.mapreduce.Job; 
10. import org.apache.hadoop.mapreduce.Mapper; 
11. import org.apache.hadoop.mapreduce.Reducer; 
12. import org.apache.hadoop.mapreduce.Mapper.Context; 
13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
15. import org.apache.hadoop.util.GenericOptionsParser; 
16.  
17. public class FindFriend { 
18. 19. public static class ChangeMapper extends Mapper<Object, Text, Text, Text>{ 20. @Override 
21. public void map(Object key, Text value, Context context) throws 
IOException, InterruptedException { 
22. StringTokenizer itr = new StringTokenizer(value.toString()); 
23. Text owner = new Text(); 
24. Set<String> set = new TreeSet<String>(); 
25. owner.set(itr.nextToken()); 
26. while (itr.hasMoreTokens()) { 
27. set.add(itr.nextToken()); 
28.} 29. String[] friends = new String[set.size()]; 
30. friends = set.toArray(friends); 
31. 32. for(int i=0;i<friends.length;i++){ 
33. for(int j=i+1;j<friends.length;j++){ 
34. String outputkey = friends[i]+friends[j]; 
35. context.write(new Text(outputkey),owner); 
36. } 37. } 
38. } 
39. } 
40. 41. public static class FindReducer extends Reducer<Text,Text,Text,Text> { 42. public void reduce(Text key, Iterable<Text> values,  
43. Context context) throws IOException, InterruptedException { 
44. String  commonfriends ="";  
45. for (Text val : values) { 
46. if(commonfriends == ""){ 
47. commonfriends = val.toString(); 
48. }else{ 
49. commonfriends = commonfriends+":"+val.toString(); 
50. } 
51. } 
52. ontext.write(key, new Text(commonfriends)); 53. } 54. } 
55. 56.  
57. public static void main(String[] args) throws IOException, 
58. InterruptedException, ClassNotFoundException { 
59. 60. Configuration conf = new Configuration(); 
61. String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs(); 
62. if (otherArgs.length < 2) { 
63. System.err.println("args error"); 
64. System.exit(2); 
65. } 
66. Job job = new Job(conf, "word count"); 
67. job.setJarByClass(FindFriend.class); 
68. job.setMapperClass(ChangeMapper.class); 
69. job.setCombinerClass(FindReducer.class); 
70. job.setReducerClass(FindReducer.class); 
71. job.setOutputKeyClass(Text.class); 
72. job.setOutputValueClass(Text.class); 
73. for (int i = 0; i < otherArgs.length - 1; ++i) { 
74. FileInputFormat.addInputPath(job, new Path(otherArgs[i])); 
75. } 
76. FileOutputFormat.setOutputPath(job, 
77. new Path(otherArgs[otherArgs.length - 1])); 
78. System.exit(job.waitForCompletion(true) ? 0 : 1); 
79. 80. } 
81.  
82. }