关联规则二项集Hadoop实现

近期看mahout的关联规则源码，颇为头痛，本来打算写一个系列分析关联规则的源码的，但是后面看到有点乱了，可能是稍微有点复杂吧，所以就打算先实现最简单的二项集关联规则。

算法的思想还是参考上次的图片：

这里实现分为五个步骤：

1. 针对原始输入计算每个项目出现的次数；
2. 按出现次数从大到小（排除出现次数小于阈值的项目）生成frequence list file;
3. 针对原始输入的事务进行按frequence list file进行排序并剪枝；
4. 生成二项集规则；
5. 计算二项集规则出现的次数，并删除小于阈值的二项集规则；

第一步的实现：包括步骤1和步骤2，代码如下：

GetFlist.java:

1. package org.fansy.date1108.fpgrowth.twodimension;
3. import java.io.BufferedReader;
4. import java.io.IOException;
5. import java.io.InputStreamReader;
6. import java.util.ArrayList;
7. import java.util.Comparator;
8. import java.util.Iterator;
9. import java.util.List;
10. import java.util.PriorityQueue;
11. import java.util.regex.Pattern;
13. import org.apache.Hadoop.conf.Configuration;
14. import org.apache.hadoop.fs.FSDataInputStream;
15. import org.apache.hadoop.fs.FSDataOutputStream;
16. import org.apache.hadoop.fs.FileSystem;
17. import org.apache.hadoop.fs.Path;
18. import org.apache.hadoop.io.IntWritable;
19. import org.apache.hadoop.io.LongWritable;
20. import org.apache.hadoop.io.Text;
21. import org.apache.hadoop.mapreduce.Job;
22. import org.apache.hadoop.mapreduce.Mapper;
23. import org.apache.hadoop.mapreduce.Reducer;
24. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
25. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
27. // the specific comparator
28. class MyComparator implements Comparator<String>{
29. private String splitter=",";
30. public MyComparator(String splitter){
31. this.splitter=splitter;
32. }
33. @Override
34. publicint compare(String o1, String o2) {
35. // TODO Auto-generated method stub
36. String[] str1=o1.toString().split(splitter);
37. String[] str2=o2.toString().split(splitter);
38. int num1=Integer.parseInt(str1[1]);
39. int num2=Integer.parseInt(str2[1]);
40. if(num1>num2){
41. return -1;
42. }elseif(num1<num2){
43. return1;
44. }else{
45. return str1[0].compareTo(str2[0]);
46. }
47. }
48. }
50. publicclass GetFList {
51. /**
52. * the program is based on the picture
53. */
54. // Mapper
55. publicstaticclass MapperGF extends Mapper<LongWritable ,Text ,Text,IntWritable>{
56. private Pattern splitter=Pattern.compile("[ ]*[ ,|\t]");
57. privatefinal IntWritable newvalue=new IntWritable(1);
58. publicvoid map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{
59. String [] items=splitter.split(value.toString());
60. for(String item:items){
61. context.write(new Text(item), newvalue);
62. }
63. }
64. }
65. // Reducer
66. publicstaticclass ReducerGF extends Reducer<Text,IntWritable,Text ,IntWritable>{
67. publicvoid reduce(Text key,Iterable<IntWritable> value,Context context) throws IOException, InterruptedException{
68. int temp=0;
69. for(IntWritable v:value){
70. temp+=v.get();
71. }
72. context.write(key, new IntWritable(temp));
73. }
74. }
75. publicstaticvoid main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
76. // TODO Auto-generated method stub
78. if(args.length!=3){
79. System.out.println("Usage: <input><output><min_support>");
80. System.exit(1);
81. }
82. String input=args[0];
83. String output=args[1];
84. int minSupport=0;
85. try {
86. minSupport=Integer.parseInt(args[2]);
87. } catch (NumberFormatException e) {
88. // TODO Auto-generated catch block
89. minSupport=3;
90. }
91. Configuration conf=new Configuration();
92. String temp=args[1]+"_temp";
93. Job job=new Job(conf,"the get flist job");
94. job.setJarByClass(GetFList.class);
95. job.setMapperClass(MapperGF.class);
96. job.setCombinerClass(ReducerGF.class);
97. job.setReducerClass(ReducerGF.class);
98. job.setOutputKeyClass(Text.class);
99. job.setOutputValueClass(IntWritable.class);
100. FileInputFormat.setInputPaths(job, new Path(input));
101. FileOutputFormat.setOutputPath(job, new Path(temp));
102. boolean succeed=job.waitForCompletion(true);
103. if(succeed){
104. // read the temp output and write the data to the final output
105. List<String> list=readFList(temp+"/part-r-00000",minSupport);
106. System.out.println("the frequence list has generated ... ");
107. // generate the frequence file
108. generateFList(list,output);
109. System.out.println("the frequence file has generated ... ");
110. }else{
111. System.out.println("the job is failed");
112. System.exit(1);
113. }
114. }
115. // read the temp_output and return the frequence list
116. publicstatic List<String> readFList(String input,int minSupport) throws IOException{
117. // read the hdfs file
118. Configuration conf=new Configuration();
119. Path path=new Path(input);
120. FileSystem fs=FileSystem.get(path.toUri(),conf);
121. FSDataInputStream in1=fs.open(path);
122. PriorityQueue<String> queue=new PriorityQueue<String>(15,new MyComparator("\t"));
123. InputStreamReader isr1=new InputStreamReader(in1);
124. BufferedReader br=new BufferedReader(isr1);
125. String line;
126. while((line=br.readLine())!=null){
127. int num=0;
128. try {
129. num=Integer.parseInt(line.split("\t")[1]);
130. } catch (NumberFormatException e) {
131. // TODO Auto-generated catch block
132. num=0;
133. }
134. if(num>minSupport){
135. queue.add(line);
136. }
137. }
138. br.close();
139. isr1.close();
140. in1.close();
141. List<String> list=new ArrayList<String>();
142. while(!queue.isEmpty()){
143. list.add(queue.poll());
144. }
145. return list;
146. }
147. // generate the frequence file
148. publicstaticvoid generateFList(List<String> list,String output) throws IOException{
149. Configuration conf=new Configuration();
150. Path path=new Path(output);
151. FileSystem fs=FileSystem.get(path.toUri(),conf);
152. FSDataOutputStream writer=fs.create(path);
153. Iterator<String> i=list.iterator();
154. while(i.hasNext()){
155. writer.writeBytes(i.next()+"\n");// in the last line add a \n which is not supposed to exist
156. }
157. writer.close();
158. }
159. }