Mahout贝叶斯算法源码分析（8）

接着上篇blog&＃xff0c;继续看log里面的信息如下&＃xff1a;

&＃43; echo &＃39;Training Naive Bayes model&＃39;
Training Naive Bayes model
&＃43; ./bin/mahout trainnb -i /home/mahout/mahout-work-mahout/20news-train-vectors -el -o /home/mahout/mahout-work-mahout/model -li /home/mahout/mahout-work-mahout/labelindex -ow
这里mahout的trainnb对应的源码文件是TrainNaiveBayesJob类&＃xff0c;该类主要的工作是&＃xff1a;&＃xff08;1&＃xff09;新建了一个LabelIndex的文件&＃xff1b;&＃xff08;2&＃xff09;执行了一个prepareJob&＃xff0c;Mapper和Reducer分别是&＃xff1a;IndexInstancesMapper、VectorSumReducer&＃xff1b;&＃xff08;2&＃xff09;执行了另外的一个prepareJob&＃xff0c;Mapper和Reducer分别是&＃xff1a;WeightsMapper、VectorSumReducer&＃xff1b;本篇主要分析前面两个工作。

新建LabelIndex的代码如下&＃xff1a;

private long createLabelIndex(Path labPath) throws IOException {long labelSize &＃61; 0;if (hasOption(LABELS)) {Iterable labels &＃61; Splitter.on(",").split(getOption(LABELS));labelSize &＃61; BayesUtils.writeLabelIndex(getConf(), labels, labPath);} else if (hasOption(EXTRACT_LABELS)) {SequenceFileDirIterable iterable &＃61;new SequenceFileDirIterable(getInputPath(), PathType.LIST, PathFilters.logsCRCFilter(), getConf());labelSize &＃61; BayesUtils.writeLabelIndex(getConf(), labPath, iterable);}return labelSize;}这里的主要工作是把相关的文件名转换为数字&＃xff0c;文件名如下图&＃xff1a;

下面看Mapper&＃xff0c;IndexInstancesMapper的主要代码如下&＃xff1a;

labelIndex &＃61; BayesUtils.readIndexFromCache(ctx.getConfiguration());
String label &＃61; labelText.toString().split("/")[1]; if (labelIndex.containsKey(label)) {ctx.write(new IntWritable(labelIndex.get(label)), instance);首先在setup函数中读取labelindex的map映射关系&＃xff0c;然后在map中针对输入/alt.atheism/51060解析/后面的字符串&＃xff0c;即文件名进行匹配&＃xff0c;输出对应的数字和相应的value不变&＃xff1b;

VectorSumReducer&＃xff1a;

Vector vector &＃61; null;for (VectorWritable v : values) {if (vector &＃61;&＃61; null) {vector &＃61; v.get();} else {vector.assign(v.get(), Functions.PLUS);}}ctx.write(key, new VectorWritable(vector));

上面的代码就是把相同的文件对应的word的单词的个数全部加起来&＃xff0c;由于一共有20个文件&＃xff0c;所以这里的reduce输出应该有20个&＃xff0c;对应log里面的信息&＃xff0c;可以看到确实匹配&＃xff0c;如下图&＃xff1a;

这里额可以通过下面的代码来测试相关的文件&＃xff1a;

package mahout.fansy.test.bayes.read;import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.mahout.math.VectorWritable;public class ReadLabelIndex {/*** &＃64;param args*/public static Configuration conf&＃61;new Configuration();static String fPath&＃61;"";static String trainPath&＃61;"";static{conf.set("mapred.job.tracker", "ubuntu:9001");fPath&＃61;"hdfs://ubuntu:9000/home/mahout/mahout-work-mahout/labelindex"; // lableindex 数据文件trainPath&＃61;"hdfs://ubuntu:9000/home/mahout/mahout-work-mahout/"&＃43;"20news-train-vectors/part-r-00000"; // 训练样本数据}public static void main(String[] args) throws IOException {// readFromFile(fPath);readFromFile(trainPath);}/*** 读取LabelIndex文件* &＃64;param fPath* &＃64;return* &＃64;throws IOException*/public static Map readFromFile(String fPath) throws IOException{FileSystem fs &＃61; FileSystem.get(URI.create(fPath), conf);Path path &＃61; new Path(fPath);Map map&＃61;new HashMap();SequenceFile.Reader reader &＃61; null;try {reader &＃61; new SequenceFile.Reader(fs, path, conf);Writable key &＃61; (Writable)ReflectionUtils.newInstance(reader.getKeyClass(), conf);Writable value &＃61; (Writable)ReflectionUtils.newInstance(reader.getValueClass(), conf);while (reader.next(key, value)) {// Writable k&＃61;; // 如何实现Writable的深度复制&＃xff1f;// map.put(key, value);System.out.println(key.toString()&＃43;", "&＃43;value.toString());System.exit(-1);// 只打印第一条记录}} finally {IOUtils.closeStream(reader);}return map;}}


这里在写的时候想做一个通用的&＃xff0c;所以需要对Writable深度复制&＃xff0c;但是一时间还没有想到办法&＃xff0c;所以这里留个问题&＃xff0c;有时间解决。

分享&＃xff0c;成长&＃xff0c;快乐

转载请注明blog地址&＃xff1a;http://blog.csdn.net/fansy1990