java使用hadoop实现关联商品统计

package com;  import java.io.IOException; import java.util.HashMap; import java.util.Map.Entry;  import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; 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.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner;  public class Test extends Configured implements Tool{    /**    * map类，实现数据的预处理    * 输出结果key为商品A value为关联商品B    * @author lulei    */   public static class MapT extends Mapper<LongWritable, Text, Text, Text> {     public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{       String line = value.toString();       if (!(line == null || "".equals(line))) {         //分割商品         String []vs = line.split(",");         //两两组合，构成一条记录         for (int i = 0; i < (vs.length - 1); i++) {           if ("".equals(vs[i])) {//排除空记录             continue;           }           for (int j = i+1; j < vs.length; j++) {             if ("".equals(vs[j])) {               continue;             }             //输出结果             context.write(new Text(vs[i]), new Text(vs[j]));             context.write(new Text(vs[j]), new Text(vs[i]));           }         }       }      }   }      /**    * reduce类，实现数据的计数    * 输出结果key 为商品A|B value为该关联次数    * @author lulei    */   public static class ReduceT extends Reducer<Text, Text, Text, IntWritable> {     private int count;          /**      * 初始化      */     public void setup(Context context) {       //从参数中获取最小记录个数       String countStr = context.getConfiguration().get("count");       try {         this.count = Integer.parseInt(countStr);       } catch (Exception e) {         this.count = 0;       }     }     public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{       String keyStr = key.toString();       HashMap<String, Integer> hashMap = new HashMap<String, Integer>();       //利用hash统计B商品的次数       for (Text value : values) {         String valueStr = value.toString();         if (hashMap.containsKey(valueStr)) {           hashMap.put(valueStr, hashMap.get(valueStr) + 1);         } else {           hashMap.put(valueStr, 1);         }       }       //将结果输出       for (Entry<String, Integer> entry : hashMap.entrySet()) {         if (entry.getValue() >= this.count) {//只输出次数不小于最小值的           context.write(new Text(keyStr + "|" + entry.getKey()), new IntWritable(entry.getValue()));         }       }     }   }      @Override   public int run(String[] arg0) throws Exception {     // TODO Auto-generated method stub     Configuration conf = getConf();     conf.set("count", arg0[2]);          Job job = new Job(conf);     job.setJobName("jobtest");          job.setOutputFormatClass(TextOutputFormat.class);     job.setOutputKeyClass(Text.class);     job.setOutputValueClass(Text.class);          job.setMapperClass(MapT.class);     job.setReducerClass(ReduceT.class);          FileInputFormat.addInputPath(job, new Path(arg0[0]));     FileOutputFormat.setOutputPath(job, new Path(arg0[1]));          job.waitForCompletion(true);          return job.isSuccessful() ? 0 : 1;        }      /**    * @param args    */   public static void main(String[] args) {     // TODO Auto-generated method stub     if (args.length != 3) {       System.exit(-1);     }     try {       int res = ToolRunner.run(new Configuration(), new Test(), args);       System.exit(res);     } catch (Exception e) {       // TODO Auto-generated catch block       e.printStackTrace();     }   }  }