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
- 1
- 2
- 3
- 4
前往页