Twenty Newsgroups Classification任务之二seq2sparse(2)

接上篇,SequenceFileTokenizerMapper的输出文件在/home/mahout/mahout-work-mahout0/20news-vectors/tokenized-documents/part-m-00000文件即可查看,同时可以编写下面的代码来读取该文件(该代码是根据前面读出聚类中心点文件改编的),如下:

package mahout.fansy.test.bayes.read;

import java.util.ArrayList;
import java.util.List; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.apache.mahout.common.StringTuple;
import org.apache.mahout.common.iterator.sequencefile.PathFilters;
import org.apache.mahout.common.iterator.sequencefile.PathType;
import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable; public class ReadFromTokenizedDocuments { /**
* @param args
*/
private static Configuration conf; public static void main(String[] args) {
conf=new Configuration();
conf.set("mapred.job.tracker", "ubuntu:9001");
String path="hdfs://ubuntu:9000/home/mahout/mahout-work-mahout0/20news-vectors/tokenized-documents/part-m-00000"; getValue(path,conf);
} /**
* 把序列文件读入到一个变量中;
* @param path 序列文件
* @param conf Configuration
* @return 序列文件读取的变量
*/
public static List<StringTuple> getValue(String path,Configuration conf){
Path hdfsPath=new Path(path);
List<StringTuple> list = new ArrayList<StringTuple>();
for (Writable value : new SequenceFileDirValueIterable<Writable>(hdfsPath, PathType.LIST,
PathFilters.partFilter(), conf)) {
Class<? extends Writable> valueClass = value.getClass();
if (valueClass.equals(StringTuple.class)) {
StringTuple st = (StringTuple) value;
list.add(st);
} else {
throw new IllegalStateException("Bad value class: " + valueClass);
}
}
return list;
} }

通过上面的文件可以读取到第一个StringTuple的单词个数有1320个(去掉stop words的单词数);

然后就又是一堆参数的设置,一直到267行,判断processIdf是否为非true,因为前面设置的是tfdif,所以这里进入else代码块,如下:

if (!processIdf) {
DictionaryVectorizer.createTermFrequencyVectors(tokenizedPath, outputDir, tfDirName, conf, minSupport, maxNGramSize,
minLLRValue, norm, logNormalize, reduceTasks, chunkSize, sequentialAccessOutput, namedVectors);
} else {
DictionaryVectorizer.createTermFrequencyVectors(tokenizedPath, outputDir, tfDirName, conf, minSupport, maxNGramSize,
minLLRValue, -1.0f, false, reduceTasks, chunkSize, sequentialAccessOutput, namedVectors);
}

这里直接调用DictionaryVectorizer的createTermFrequencyVectors方法,进入该方法(DictionaryVectorizer的145行),可以看到首先也是一些参数的设置,然后就到了startWordCounting方法了,进入这个方法可以看到这个是一个Job的基本设置,其Mapper、Combiner、Reducer分别为:TermCountMapper、TermCountCombiner、TermCountReducer,下面分别来看各个部分的作用(其实和最基本的wordcount很相似):

TermCountMapper,首先贴代码:

protected void map(Text key, StringTuple value, final Context context) throws IOException, InterruptedException {
OpenObjectLongHashMap<String> wordCount = new OpenObjectLongHashMap<String>();
for (String word : value.getEntries()) {
if (wordCount.containsKey(word)) {
wordCount.put(word, wordCount.get(word) + 1);
} else {
wordCount.put(word, 1);
}
}
wordCount.forEachPair(new ObjectLongProcedure<String>() {
@Override
public boolean apply(String first, long second) {
try {
context.write(new Text(first), new LongWritable(second));
} catch (IOException e) {
context.getCounter("Exception", "Output IO Exception").increment(1);
} catch (InterruptedException e) {
context.getCounter("Exception", "Interrupted Exception").increment(1);
}
return true;
}
});

该部分代码首先定义了一个Mahout开发人员定义的Map类,然后遍历value中的各个单词(比如第一个value中有1320个单词);当遇到map中没有的单词就把其加入map中,否则把map中该单词的数量加1更新原来的单词的数量,即for循环里面做的事情;然后就是forEachPair方法了,这里应该是复写了该方法?好像是直接新建了一个类然后把这个新建的类作为forEachPair的参数;直接看context.write吧,应该是把wordCount(这个变量含有每个单词和它的计数)中的各个单词和单词计数分别作为key和value输出;

然后是TermCountCombiner和TermCountReducer,这两个代码一样的和当初学习Hadoop入门的第一个例子是一样的,这里就不多说了。查看log信息,可以看到reduce一共输出93563个单词。

然后就到了createDictionaryChunks函数了,进入到DictionaryVectorizer的215行中的该方法:

 List<Path> chunkPaths = Lists.newArrayList();

    Configuration conf = new Configuration(baseConf);

    FileSystem fs = FileSystem.get(wordCountPath.toUri(), conf);

    long chunkSizeLimit = chunkSizeInMegabytes * 1024L * 1024L;
int chunkIndex = 0;
Path chunkPath = new Path(dictionaryPathBase, DICTIONARY_FILE + chunkIndex);
chunkPaths.add(chunkPath); SequenceFile.Writer dictWriter = new SequenceFile.Writer(fs, conf, chunkPath, Text.class, IntWritable.class); try {
long currentChunkSize = 0;
Path filesPattern = new Path(wordCountPath, OUTPUT_FILES_PATTERN);
int i = 0;
for (Pair<Writable,Writable> record
: new SequenceFileDirIterable<Writable,Writable>(filesPattern, PathType.GLOB, null, null, true, conf)) {
if (currentChunkSize > chunkSizeLimit) {
Closeables.closeQuietly(dictWriter);
chunkIndex++; chunkPath = new Path(dictionaryPathBase, DICTIONARY_FILE + chunkIndex);
chunkPaths.add(chunkPath); dictWriter = new SequenceFile.Writer(fs, conf, chunkPath, Text.class, IntWritable.class);
currentChunkSize = 0;
} Writable key = record.getFirst();
int fieldSize = DICTIONARY_BYTE_OVERHEAD + key.toString().length() * 2 + Integer.SIZE / 8;
currentChunkSize += fieldSize;
dictWriter.append(key, new IntWritable(i++));
}
maxTermDimension[0] = i;
} finally {
Closeables.closeQuietly(dictWriter);
}

这里看到新建了一个Writer,然后遍历该文件的key和value,但是只读取key值,即单词,然后把这些单词进行编码,即第一个单词用0和它对应,第二个单词用1和它对应。

上面代码使用的dictWriter查看变量并没有看到哪个属性是存储单词和对应id的,所以这里的写入文件的机制是append就写入?还是我没有找到正确的属性?待查。。。

分享,快乐,成长

转载请注明出处:http://blog.csdn.net/fansy1990

上一篇:Twenty Newsgroups Classification任务之二seq2sparse


下一篇:[HBase Manual]CH5 HBase运行模式:单实例和分布式