web 工程调用hadoop集群1.3

package org.hadoop; import java.io.IOException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; 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.SequenceFileOutputFormat; import org.apache.mahout.math.RandomAccessSparseVector; import org.apache.mahout.math.Vector; import org.apache.mahout.math.VectorWritable; import org.fz.algorithm.hadoop.model.Text2vectorVo; import util.hadoop.HadoopUtil; /** * transform the text to vector * @author fansy * */ public class Text2VectorDriver { private final static String SPLIT ="split"; private static Log log=LogFactory.getLog(Text2VectorDriver.class); public boolean run(Text2vectorVo text2vectorVo) throws IOException, ClassNotFoundException, InterruptedException{ Configuration conf=new Configuration(); conf.set("mapred.job.tracker", HadoopUtil.getJobTracker()); conf.set(SPLIT, text2vectorVo.getSplit()); Job job=new Job(conf,"text2vector with input"+text2vectorVo.getInput()); job.setJarByClass(Text2VectorDriver.class); job.setOutputFormatClass(SequenceFileOutputFormat.class); job.setReducerClass(T2VReducer.class); job.setMapperClass(T2VMapper.class); job.setMapOutputKeyClass(LongWritable.class); job.setMapOutputValueClass(VectorWritable.class); job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(VectorWritable.class); FileInputFormat.addInputPath(job, new Path(text2vectorVo.getInput())); SequenceFileOutputFormat.setOutputPath(job, new Path(text2vectorVo.getOutput())); boolean flag=job.waitForCompletion(true); return flag; } public static class T2VMapper extends Mapper<LongWritable,Text,LongWritable,VectorWritable>{ private String split=","; @Override public void setup(Context cxt){ if(cxt.getConfiguration().get(SPLIT)!=null){ split=cxt.getConfiguration().get(SPLIT); } } @Override public void map(LongWritable key,Text line,Context cxt)throws IOException,InterruptedException{ String[] str=line.toString().split(split); // split data Vector vector=new RandomAccessSparseVector(str.length); for(int i=0;i<str.length;i++){ double tempVal=-1; try { tempVal = Double.parseDouble(str[i]); } catch (Exception e) { log.info("Exception when parsing string to double,with error:"+e.getMessage()); return; } vector.set(i, tempVal); } VectorWritable va=new VectorWritable(vector); cxt.write(key, va); } } public static class T2VReducer extends Reducer<LongWritable,VectorWritable,LongWritable,VectorWritable>{ public void reduce(LongWritable key,Iterable<VectorWritable> values,Context context)throws IOException,InterruptedException{ for(VectorWritable v:values){ context.write(key, v); } } } }