naive bayes在hadoop上的实现

import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.util.StringTokenizer; import java.util.Random; import java.lang.String; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.lang.Math; import java.net.URI; import java.util.Arrays; //import java.io.FileReader; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.DoubleWritable; 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.Mapper.Context; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class NativeBayes { // public static int SumCases=0;// 总共的样例数量； // public static HashMap ModelData=new HashMap<String, Double>(); // public static HashMap CategoryKinds=new HashMap<Integer,String>(); // public static HashSet CategoryKinds=new HashSet<String>(); public static class CategoryMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text category = new Text(),tempAttrib=new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); String tempCategory=new String(); int CountAttrib=itr.countTokens()-2; while (itr.hasMoreTokens()) { //category.set(itr.nextToken()); tempCategory=new String(itr.nextToken()); } category.set("CF_"+tempCategory); // CategoryKinds.add(tempCategory); context.write(category, one); context.write(new Text("SumCases"), one); // SumCases++; itr = new StringTokenizer(value.toString()); int tempi=0; tempAttrib.set(itr.nextToken()); while(tempi<CountAttrib) { tempi++; tempAttrib.set(""+tempi+"_"+itr.nextToken()+"_"+tempCategory); context.write(tempAttrib, one); } } } public static class CategoryCountReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); // ModelData.put(key.toString(), sum); context.write(key, result); } } public static class ComputeMapper extends Mapper<Object, Text, Text, Text>{ // private final static DoubleWritable result=new DoubleWritable(1); private Text id=new Text(),result=new Text(); public static HashMap ModelData=new HashMap<String, Double>(); public static HashSet CategoryKinds=new HashSet<String>(); @Override protected void setup(Context context) throws IOException, InterruptedException { // TODO Auto-generated method stub super.setup(context); try{ FileSystem fs = null; fs = FileSystem.get(URI.create("hdfs://node2:9000"), new Configuration()); FSDataInputStream in; BufferedReader bufread; String strLine; String[] strList; String[] tempField=new String[2]; Path ModelPath = new Path("/NativeBayes/ModelData/ModelOut"); if (!fs.exists(ModelPath)) throw new IOException("Input file not found"); if (!fs.isFile(ModelPath)) throw new IOException("Input should be a file"); in = fs.open(ModelPath); bufread = new BufferedReader(new InputStreamReader(in)); while ((strLine = bufread.readLine()) != null) { strList =strLine.split(","); // if(strList.length < 2) // continue; // IpField nodeIp = new IpField(strList[0], strList[1], strList[2]); ModelData.put(strList[0], Double.parseDouble(strList[1])); if(strList[0].startsWith("CF_")){ CategoryKinds.add(strList[0].substring(3)); } } in.close(); } catch(IOException e) { // ModelData.put(e.toString(), 0); e.printStackTrace(); } } public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { double result_temp=1.0; // StringTokenizer itr = new StringTokenizer(value.toString()); int tempi=0; int CountAttrib=new StringTokenizer(value.toString()).countTokens()-1; // int CountAttrib=4; // double[] result_all_kinds=new double[CategoryKinds.size()];//CategoryKinds.size()是种类的数量 // for(;tempi<CategoryKinds.size();tempi++)result_all_kinds[tempi]=1.0; String tempAttrib=new String(),tempCategory=new String(); // id.set(itr.nextToken());//key为ID值??+类别 Iterator it=CategoryKinds.iterator(); tempi=0; while(it.hasNext()){ tempCategory=new String(it.next().toString()); // id.set(id.toString()+tempCategory); StringTokenizer itr = new StringTokenizer(value.toString()); id.set(itr.nextToken()); while(itr.hasMoreTokens()){ tempi++; tempAttrib=new String(""+tempi+"_"+itr.nextToken()+"_"+tempCategory);//为了获取类条件概率的值 // System.out.println(tempAttrib); try{ result_temp*=Double.parseDouble(ModelData.get(tempAttrib).toString()); } catch(NullPointerException e) { result_temp=0; } } result_temp/=Math.pow(Double.parseDouble(ModelData.get("CF_"+tempCategory).toString()), CountAttrib-1); result_temp/=Double.parseDouble(ModelData.get("SumCases").toString()); result.set(tempCategory+"_"+result_temp); context.write(id, result); // temp_resulti++; result_temp=1.0; tempi=0; } } } public static class ComputeReducer extends Reducer<Text,Text,Text,Text>{ private Text category = new Text(); public void reduce(Text key, Iterable<Text> values, Context context ) throws IOException, InterruptedException{ String tempC_V=new String(); String[] C_V=new String[2]; HashMap Category_Value=new HashMap<Double,String>(); for(Text val:values){ tempC_V=val.toString(); System.out.println(tempC_V); C_V=tempC_V.split("_"); Category_Value.put(Double.parseDouble(C_V[1]),C_V[0] ); } Object[] Key=Category_Value.keySet().toArray(); Arrays.sort(Key); // for (int i = 0; i < Key.length; i++) { // System.out.println(map.get(key[i])); // category.set(Category_Value.get(Key[Key.length-1]).toString()); // } // category.set(Category_Value.get(Key[Key.length-1]).toString()+"_"+Key[Key.length-1]); category.set(Category_Value.get(Key[Key.length-1]).toString()); context.write(key, category); } } public static class NameTextOutputFormat extends MultipleOutputFormat<Text, IntWritable> { @Override protected String generateFileNameForKeyValue(Text key, IntWritable value, Configuration conf) { return "ModelOut"; } } /** * @param args */ public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 3) { System.err.println("Usage: categorycount <in> <in> <out>"); System.exit(3); } Job job1 = new Job(conf, "category count"); job1.setJarByClass(NativeBayes.class); job1.setMapperClass(CategoryMapper.class); job1.setCombinerClass(CategoryCoun