文本挖掘分词mapreduce化

软件版本

paoding-analysis3.0

项目jar包和拷贝庖丁dic目录到项目的类路径下

修改paoding-analysis.jar下的paoding-dic-home.properties文件设置词典文件路径

paoding.dic.home=classpath:dic

分词程序demo

import?java.io.IOException;
import?java.io.StringReader;

import?org.apache.lucene.analysis.TokenStream;
import?org.apache.lucene.analysis.tokenattributes.CharTermAttribute;

import?net.paoding.analysis.analyzer.PaodingAnalyzer;


public?class?TokenizeWithPaoding?{
public?static?void?main(String[]?args)?{
????
????String?line="中华民族共和国";
????PaodingAnalyzer?analyzer?=new?PaodingAnalyzer();
????StringReader?sr=new?StringReader(line);
????TokenStream?ts=analyzer.tokenStream("",?sr);//分词流,第一个参数无意义
????//迭代分词流
????try?{
????????while(ts.incrementToken()){
????????????CharTermAttribute?ta=ts.getAttribute(CharTermAttribute.class);
????????????System.out.println(ta.toString());
????????}
????}?catch?(Exception?e)?{
????????
????????e.printStackTrace();
????}
}
}

新闻文文本分类源文件

http://people.csail.mit.edu/jrennie/20Newsgroups/20news-bydate.tar.gz

每个文件夹代表一个类别，每个类别下的文件代表一条新闻

中文新闻分类需要先分词

对于大量小文件可以使用FileInputFormat的另一个抽象子类CombineFileInputFormat实现createRecordReader方法

CombineFileInputFormat重写了getSpilt方法，返回的分片类型是CombineFileSpilt,是InputSpilt的子类，可包含多个文件

RecordReader怎么由文件生成key-value是由nextKeyValue函数决定

自定义的CombineFileInputFormat类

package?org.conan.myhadoop.fengci;



import?java.io.IOException;

import?org.apache.hadoop.fs.Path;
import?org.apache.hadoop.io.Text;
import?org.apache.hadoop.mapreduce.InputSplit;
import?org.apache.hadoop.mapreduce.JobContext;
import?org.apache.hadoop.mapreduce.RecordReader;
import?org.apache.hadoop.mapreduce.TaskAttemptContext;
import?org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;
import?org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;
import?org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;

/**
?*?自定义MyInputFormat类,?用于实现一个Split包含多个文件
?*?@author?BOB
?*
?*/
public?class?MyInputFormat?extends?CombineFileInputFormat<Text,?Text>{
????????
????????//禁止文件切分
????????@Override
????????protected?boolean?isSplitable(JobContext?context,?Path?file)?{
????????????????return?false;
????????}

????????@Override
????????public?RecordReader<Text,?Text>?createRecordReader(InputSplit?split,?TaskAttemptContext?context)?throws?IOException?{
????????????????return?new?CombineFileRecordReader<Text,?Text>((CombineFileSplit)split,?context,?MyRecordReader.class);
????????}

}

自定义的RecordReader类

package?org.conan.myhadoop.fengci;

import?java.io.IOException;

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.Text;
import?org.apache.hadoop.mapreduce.InputSplit;
import?org.apache.hadoop.mapreduce.RecordReader;
import?org.apache.hadoop.mapreduce.TaskAttemptContext;
import?org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;

/**
?*?自定义MyRecordReader类,?用于读取MyInputFormat对象切分的Split分片中的内容
?*?@author?BOB
?*
?*/
public?class?MyRecordReader?extends?RecordReader<Text,?Text>?{

????????private?CombineFileSplit?combineFileSplit;????????????????//当前处理的分片
????????private?Configuration?conf;????????????????????????//作业的配置信息
????????private?Text?currentKey?=?new?Text();????????????????//当前读入的key
????????private?Text?currentValue?=?new?Text();????????//当前读入的value
????????private?int?totalLength;????????????????????????//当前分片中文件的数量
????????private?int?currentIndex;????????????????????????????????//正在读取的文件在当前分片中的位置索引
????????private?float?currentProgress?=?0F;????????????????//当前进度
????????private?boolean?processed?=?false;????????//标记当前文件是否已经被处理过
????????
????????//构造方法
????????public?MyRecordReader(CombineFileSplit?combineFileSplit,????????????????????????TaskAttemptContext?context,?Integer?fileIndex)?{
????????????????super();
????????????????this.combineFileSplit?=?combineFileSplit;
????????????????this.currentIndex?=?fileIndex;
????????????????this.conf?=?context.getConfiguration();
????????????????this.totalLength?=?combineFileSplit.getPaths().length;
????????}

?????
????????@Override
????????public?void?initialize(InputSplit?split,?TaskAttemptContext?context)?throws?IOException,?InterruptedException?{

????????}
????????@Override
????????public?Text?getCurrentKey()?throws?IOException,?InterruptedException?{
????????????????return?currentKey;
????????}

????????@Override
????????public?Text?getCurrentValue()?throws?IOException,?InterruptedException?{
????????????????return?currentValue;
????????}

????????@Override
????????public?float?getProgress()?throws?IOException,?InterruptedException?{
????????????????if(currentIndex?>=?0?&&?currentIndex?<?totalLength)?{
????????????????????????return?currentProgress?=?(float)?currentIndex/totalLength;
????????????????}
????????????????return?currentProgress;
????????}

????????@Override
????????public?void?close()?throws?IOException?{

????????}


????????@Override
????????public?boolean?nextKeyValue()?throws?IOException,?InterruptedException?{
????????????????if(!processed)?{
????????????????????????//由文件的父目录,?文件名以及目录分割符组成key
????????????????????????Path?file?=?combineFileSplit.getPath(currentIndex);
????????????????????????StringBuilder?sb?=?new?StringBuilder();
????????????????????????sb.append("/");
????????????????????????sb.append(file.getParent().getName()).append("/");
????????????????????????sb.append(file.getName());
????????????????????????currentKey.set(sb.toString());
????????????????????????
????????????????????????//以整个文件的内容作为value
????????????????????????FSDataInputStream?in?=?null;
????????????????????????byte[]?content?=?new?byte[(int)combineFileSplit.getLength(currentIndex)];
????????????????????????FileSystem?fs?=?file.getFileSystem(conf);
????????????????????????in?=?fs.open(file);
????????????????????????in.readFully(content);
????????????????????????currentValue.set(content);
????????????????????????in.close();
????????????????????????processed?=?true;
????????????????????????return?true;
????????????????}
????????????????return?false;
????????}

}

分词驱动类

package?org.conan.myhadoop.fengci;

import?java.io.IOException;
import?java.io.StringReader;



import?net.paoding.analysis.analyzer.PaodingAnalyzer;

import?org.apache.hadoop.conf.Configuration;
import?org.apache.hadoop.conf.Configured;
import?org.apache.hadoop.fs.FileStatus;
import?org.apache.hadoop.fs.FileSystem;
import?org.apache.hadoop.fs.FileUtil;
import?org.apache.hadoop.fs.Path;
import?org.apache.hadoop.io.Text;
import?org.apache.hadoop.mapreduce.Job;
import?org.apache.hadoop.mapreduce.Mapper;
import?org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import?org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import?org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import?org.apache.hadoop.util.Tool;
import?org.apache.hadoop.util.ToolRunner;
import?org.apache.lucene.analysis.Analyzer;
import?org.apache.lucene.analysis.TokenStream;
import?org.apache.lucene.analysis.tokenattributes.CharTermAttribute;



/**
?*?分词驱动器类,?用于给输入文件进行分词
?*?@author?BOB
?*
?*/
public?class?TokenizerDriver?extends?Configured?implements?Tool{

????????
????????public?static?void?main(String[]?args)?throws?Exception{
????????????????int?res?=?ToolRunner.run(new?Configuration(),?new?TokenizerDriver(),?args);
????????????????System.exit(res);
????????}

????????@Override
????????public?int?run(String[]?args)?throws?Exception?{
????????????????Configuration?conf?=?new?Configuration();
????????????????//参数设置
????????????????conf.setLong("mapreduce.input.fileinputformat.split.maxsize",?4000000);
??????????????//作业名称
????????????????Job?job?=?new?Job(conf,"Tokenizer");
????????????????job.setJarByClass(TokenizerDriver.class);
????????????????
????????????????job.setMapperClass(Map.class);
????????????????
????????????????job.setInputFormatClass(MyInputFormat.class);
????????????????
????????????????job.setOutputFormatClass(SequenceFileOutputFormat.class);
????????????????
????????????????job.setOutputKeyClass(Text.class);
????????????????job.setOutputValueClass(Text.class);
????????????????
????????????????Path?inpath=new?Path(args[0]);
????????????????Path?outpath=new?Path(args[1]);
????????????????FileSystem?fs?=?inpath.getFileSystem(conf);
????????????????FileStatus[]?status?=?fs.listStatus(inpath);
????????????????Path[]?paths?=?FileUtil.stat2Paths(status);
????????????????for(Path?path?:?paths)?{
????????????????????????FileInputFormat.addInputPath(job,?path);
????????????????}
????????????????FileOutputFormat.setOutputPath(job,?outpath);
????????????????
????????????????//输出文件夹已经存在则删除
????????????????FileSystem?hdfs?=?outpath.getFileSystem(conf);
????????????????if(hdfs.exists(outpath)){
????????????????????hdfs.delete(outpath,true);
????????????????????hdfs.close();
????????????????}
????????????????//没有Reduce任务
????????????????job.setNumReduceTasks(0);?
????????????????return?job.waitForCompletion(true)???0?:?1;
????????}
????????
????????/**
?????????*?Hadoop计算框架下的Map类,?用于并行处理文本分词任务
?????????*?@author?BOB
?????????*
?????????*/
????????static?class?Map?extends?Mapper<Text,?Text,?Text>?{
????????????????
????????????????@Override
????????????????protected?void?map(Text?key,?Text?value,?Context?context)?throws?IOException,?InterruptedException?{
????????????????????????//创建分词器
????????????????????????Analyzer?analyzer?=?new?PaodingAnalyzer();
????????????????????????String?line?=?value.toString();
????????????????????????StringReader?reader?=?new?StringReader(line);
????????????????????????//获取分词流对象
????????????????????????TokenStream?ts?=?analyzer.tokenStream("",?reader);
????????????????????????StringBuilder?sb?=?new?StringBuilder();
????????????????????????
????????????????????????//遍历分词流中的词语
????????????????????????while(ts.incrementToken())?{
????????????????????????????????CharTermAttribute?ta?=?ts.getAttribute(CharTermAttribute.class);
????????????????????????????????if(sb.length()?!=?0)?{
????????????????????????????????????????sb.append("?").append(ta.toString());
????????????????????????????????}?else?{
????????????????????????????????????????sb.append(ta.toString());
????????????????????????????????}
????????????????????????}
????????????????????????value.set(sb.toString());
????????????????????????context.write(key,?value);
????????????????}
????????????????
????????}
}

分词预先处理结果，将所有新闻集中到一个文本中，key为类别，一行代表一篇新闻，单词之间用空格分开

处理后的数据可用于mahout做贝叶斯分类器

参考文章：

http://f.dataguru.cn/thread-244375-1-1.html

http://www.cnblogs.com/panweishadow/p/4320720.html