一、算法说明

       1、词频TF：是指给定词语在给定文件中出现的次数，一般会做归一化，即除以文件的总词数（注意是分词数，不是字数）。

                                          TF=词在文章出现次数 / 文章的总词数

      2、逆向文件频率IDF：普遍重要性度量，由文件总数除以包含该词的文件的数目，再对商取对数。

                                          IDF=log(文件总数 / 包含目标词的文件个数)

      3、各个分词占文件的权重：TF-DF = TF * IDF

二、MapReduce分析

      MapReduce程序的输入的数据集是多条文件id对应文件内容，MapReduce需要分以下几步工作：

1.  统计文件总数；

2. 对每个文件进行分词，可以用IKSegmenter进行分词，需要引入相关jar包；

3. 以及分词后每个词在各文件中出现的次数，即词频TF；

4. 对词频做归一化，并且统计每个词有出现在文件中的文件数目；

5. 计算IDF

6. 最后计算TF-IDF

   1、2、3可以放在一个MapReduce中完成；4需要放在一个MapReduce中完成；5、6可以放在一个MapReduce中完成。

三、MapReduce实现

测试数据

2259080 当年一曲《相思引》惊艳到不行，一段采薇，从去年听到今朝
2614152 莫名的心酸，多情或许只是女子才会犯的错[流感]
2733982 汉字里墨香温存的一笔一划，世代传承的表达
3029272 必须用清淡点的歌把小苹果的旋律从脑海里尽快抹去[撇嘴]
3193581 深夜听着这首歌看书，不能更美～
3247506 无声中 折伞 你背影零落，回忆青涩泼墨，缘分在纸上太薄，我以为 烟雨只为情留，这场雨 就能下到白头，可是远山云悠悠 各自去留，我们已回不到 从前时候，我以为 山水只为你秀，这一路 就能走到白头，隔世的你挥挥手，月光已旧 葬了谁的温柔，谁的愁。
3304550 梦醒后深爱已碎了心魂，天涯海角为你一骑绝尘，颠倒乾坤 血染白裳，风沙湮没参商永隔的泪痕，一念执迷为你烽火连城，换你心疼 斩不断，重来回首已三生，踏破千山挥剑如神，恩怨纠缠不分，惊鸿照影念你情真，一曲离歌倾城。
3419874 为什么不红呢[拜]
3482756 唉！这么多年了，虽然你不再唱了，但你的声音却在互联网上永久流传...估计楼上的没几个知道，心然完全是古风流派开山鼻祖...！
3559646 我们的口号是 岁月随心 终会淡然[大笑]
3565239 《犬夜叉》插曲《穿越时空的思念》[钻石]
3575668 愿初心常在

   爬的网抑云的评论，以上只是部分数据。

第一步——处理原始文件

一、mapper通过IKSegmenter分词器对文件进行分词，输出以下三种数据：

              1、word_id 1，文件分的每个词加文件id对应一条记录

              2、id 1，每个文件id对应一条记录

              3、count 1，每个文件分的一个词对应一条记录

public class HotCommentMapper extends Mapper<Text, Text, Text, IntWritable> {private final Text wordKey = new Text();public static final Text counter = new Text("count");private final IntWritable one = new IntWritable(1);@Overrideprotected void map(Text key, Text value, Context context) throws IOException, InterruptedException {//样本数据：5824431 我大剑三不负基三盛名，听的我都醉了//计算词频IFStringReader sr = new StringReader(value.toString());IKSegmenter ikSegmenter = new IKSegmenter(sr, true);Lexeme lexeme;while ((lexeme = ikSegmenter.next()) != null) {String word = lexeme.getLexemeText();wordKey.set(word + "_" + key);//输出每条中各词计数context.write(wordKey, one);//输出每条热评的分词的总个数context.write(key, one);}//输出第一类数据到reduce统计热评总数context.write(counter, one);}
}

这里用到了IKSegmenter来分词，需要引用对应依赖

<!-- ik分词器 -->
<dependency><groupId>com.janeluo</groupId><artifactId>ikanalyzer</artifactId><version>2012_u6</version>
</dependency>

二、分区器将数据分为两类：

              1、word_id 1和word 1分为一类数据

              2、count 1分为一类

/*** 注意，这里集成HashPartitioner，可以复用它的hash分区*/
public class HotCommentPartitioner extends HashPartitioner<Text, IntWritable> {@Overridepublic int getPartition(Text key, IntWritable value, int partitonNum) {if (HotCommentMapper.counter.equals(key)) {//第四个分区统计热评总条数和每条评论的总分词数return 3;} else {//其他三个分区计算各个热评总词数以及词频，利用hash取模计算return super.getPartition(key, value, partitonNum - 1);}}
}

       注意，这里在提交job时要设置分区数为4。

//设置4个reduce task，即4个分区
job.setNumReduceTasks(4);

三、reducer统计数据：

              1、每个词在每个文件出现的次数（一个词对应一个文件一条记录，计数为1）和每个文件分了多少词

              2、总的文件数

/*** 输出统计结果。* 因为前面已经通过分区映射，所以热评总条数的结果在part-r-00003中，每条热评统计词数和词频保存在其他三个文件中*/
public class HotCommentReducer extends Reducer<Text, IntWritable, Text, IntWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {//reduce原语：同一个分区的数据在一个reducer task中执行，执行结果写在同一个文件汇总；相同key的数据作为一组调用一次reduce方法int count = 0;for (IntWritable value : values) {count += 1;}context.write(key, new IntWritable(count));}
}

四、数据结果数据集，四个分区就会有四个结果文件

part-r-00000、part-r-00001和part-r-00002保存两类结果数据

              1、分词统计词数数据

             2、每个词有出现在文件中的文件的数据，一个文件一条记录

结果集数据格式如下

路_10001814	0.024390243902439025
花开_10001814	0.024390243902439025
其实_10001814	0.024390243902439025
花_10001814	0.024390243902439025
1147863561	8
1152817034	43
1154420935	6
11569089	16
1157136429	19
彼岸花_10001814	0.04878048780487805
却_10001814	0.024390243902439025
1167186188	14
我们_10001814	0.024390243902439025
不得_10001814	0.024390243902439025
错_10001814	0.024390243902439025
1154420935	6
就像_10001814	0.024390243902439025
11867253	73
11年_57467843	1
11张_1267959822	1
11日_69687368	1
1208473319	15
12087800	9
1210328462	1
独自_10001814	0.024390243902439025
来了_10001814	0.04878048780487805
了_10001814	0.024390243902439025
念_10001814	0.024390243902439025
我_10001814	0.04878048780487805

part-r-00003保存总文件数

count	4340.0

第二步——对词频TF做归一化

一、输入结果集为第一步的所有结果集，即四个结果集中的所有数据，输入数据格式如下：

就像_10001814	0.024390243902439025
11867253	73
11年_57467843	1
11张_1267959822	1
11日_69687368	1
1208473319	15
12087800	9
1210328462	1
独自_10001814	0.024390243902439025
来了_10001814	0.04878048780487805
count	4340.0

二、mapper将数据映射成三种数据

       1、词在文件中出现的次数

       2、文件总个数count

       3、各个词出现的文件的个数

前面两种数据都是输入的原数据，直接输出就可以，第三类数据需要设置标志

/*** map输出以下三种数据：* 原数据* 1、词在文件中出现的次数* 2、文件总个数count* 新数据* 3、各个词出现的文件的个数*/
public class HotComment2Mapper extends Mapper<Text, Text, Text, IntWritable> {private final Text word = new Text();private final IntWritable one = new IntWritable(1);@Overrideprotected void map(Text key, Text value, Context context) throws IOException, InterruptedException {//数据样本//台词_40270056	1//10247087	48FileSplit fs = (FileSplit) context.getInputSplit();if (!fs.getPath().getName().contains("part-r-00003")) {//part-r-00003的数据不处理，一个分区对应一个mapper taskStringTokenizer st = new StringTokenizer(key.toString(), "_");//文件总分词数和分词计数直接输出one.set(Integer.parseInt(value.toString()));context.write(key, one);if (st.countTokens() > 1) {word.set(st.nextToken());//设置value为0，用于后面分区作为区分条件one.set(0);//输出词出现的文件的个数context.write(word, one);}} else {//直接输出原始文件总数统计数据one.set(Integer.parseInt(value.toString()));context.write(key, one);}}
}

三、分区器将数据分成三类，输出到不同结果文件：

     1、count，文件总数，就一条记录，放在3分区

     2、词在多少个文件出现的计数，放在4分区

     3、每个文件分的词数以及原始词在文件中出现的词数两种数据，放在其他3个分区

public class HotComment2Partitioner extends Partitioner<Text, IntWritable> {@Overridepublic int getPartition(Text key, IntWritable value, int numReduceTasks) {if (HotCommentMapper.counter.equals(key)) {//第四个分区统计热评总条数和每条评论的总分词数return 3;} else if (0 == value.get()) {//词对应出现的文件的个数的数据放在5个分区return 4;} else {StringTokenizer sza = new StringTokenizer(key.toString(), "_");String id = "";//map输出的第一类数据有两种形态while (sza.hasMoreTokens()) {//分割后取最后一个，肯定是idid = sza.nextToken();}//文件id对剩余分区数取模，保证同一个文件的数据再同一分区return Integer.parseInt(id) % (numReduceTasks - 2);}}
}

注意！！！在客户端提交job事要设置分区数为5。

四、排序比较器，主要针对第3类数据

    1、相同文件id的记录排在一起

    2、id相同的文件的分词数的记录放在最前面

/*** 自定义排序比较器，将文件的分词数排在第一个*/
public class HotComment2Comparator extends WritableComparator {public HotComment2Comparator() {super(Text.class, true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {int i = compareId(a, b);if (i == 0) {//如果是同一个文件，id-count的记录放在前面if (a.toString().contains("_")) {return 1;} else if (b.toString().contains("_")) {return -1;}}return i;}public static int compareId(WritableComparable a, WritableComparable b) {StringTokenizer sza = new StringTokenizer(a.toString(), "_");StringTokenizer szb = new StringTokenizer(b.toString(), "_");String aId = "", bId = "";while (sza.hasMoreTokens()) {aId = sza.nextToken();}while (szb.hasMoreTokens()) {bId = szb.nextToken();}return aId.compareTo(bId);}
}

五、分组比较器

     1、第一类数据，相同的词放在一组

     2、第三类数据，相同的文件id放在一组

/*** 自定义分组比较器，让同一文件的数据分到一组*/
public class HotCommentGroup2Comparator extends WritableComparator {public HotCommentGroup2Comparator() {super(Text.class, true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {return HotComment2Comparator.compareId(a, b);}
}

六、reduce对数据进行统计

     1、第一类数据count直接输出

     2、第二类数据统计出现的文件数输出

     3、第三类数据，先取第一个文件分词数，然后每个词计算归一化TF输出

public class HotComment2Reducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {private final DoubleWritable rval = new DoubleWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {// 输入数据样本，有三类数据：// 第一组// 10247087	48// 能_10247087	1// 相逢_10247087	1// 第二组//能	0//能	0//第三组//count 	4340if (HotCommentMapper.counter.equals(key)) {rval.set(values.iterator().next().get());context.write(key, rval);} else {int fileWordCount = 0;boolean flag = true;double countFile = 0;for (IntWritable value : values) {if (0 == value.get()) {//统计出现某词的文件个数countFile += 1;} else if (flag) {//获取文件的总分词数fileWordCount = value.get();flag = false;} else {//对if做归一化double wordCount = value.get();//这里必须用double除才能获得double，从而保留小数rval.set(wordCount / fileWordCount);context.write(key, rval);}}if (countFile > 0) {rval.set(countFile);context.write(key, rval);}}}
}

七、输出结果数据集，五个分区会有五个结果文件

part-r-00000、part-r-00001和part-r-00002保存做了归一化之后的词频TF值数据，数据格式如下

唱_10010971	0.06666666666666667
you_10010971	0.06666666666666667
键_10010971	0.06666666666666667
好_10010971	0.06666666666666667
开口_10010971	0.06666666666666667
跪_10010971	0.06666666666666667
声音_10010971	0.06666666666666667
就按_10010971	0.06666666666666667
want_10010971	0.06666666666666667
下了_10010971	0.06666666666666667
好比_10010971	0.06666666666666667
一开_10010971	0.06666666666666667
的_10010971	0.06666666666666667
i_10010971	0.06666666666666667
就_10010971	0.06666666666666667

part-r-00003保存文件总数

count	4340.0

part-r-00004保存累加后的各分词出现在的文件的个数的数据，数据格式如下

看不见	3.0
看不起	1.0
看中	1.0
看么	1.0
看书	3.0
看了	69.0
看什么	1.0
看他	4.0
看似	1.0
看你	11.0

第三步——计算TF-IDF

第三步可以分两小步：先计算逆向文件频率IDF，让后计算TF * IDF得到TF-IDF，因为TF前面已经计算出来的了。但是计算IDF用到part-r-00003和part-r-00004数据。所以在mapper的setUp中要加载这两个文件的数据。注意！！！这一步的计算是在mapper中完成的！所以需要再客户端设置缓存文件。

一、客户端主要代码

	public static void step3() {job.setJobName("hot comment-3");//当客户端在windows启动，程序在liunx运行时，该参数需要设置为true，做格式兼容，默认fasejob.getConfiguration().set("mapreduce.app-submission.cross-platform", "true");//运行平台，这个可以不用设置，默认为yarnconf.set("mapreduce.framework.name", "local");//集群分布式启动，因为part-r-00003和part-r-00004两个文件数据需要移动到mapper的计算节点job.setJar("G:\\bigdata\\hadoop-test\\target\\hadoop-test-1.0-SNAPSHOT.jar");//把文件总数加载到job，任务运行时会把该文件推送到计算节点的服务器上job.addCacheFile(new Path("/test/hot/output1/" + TOTAL_FILE).toUri());//把词对应出现的文件数数据加载到jobjob.addCacheFile(new Path("/test/hot/output1/" + WORD_COUNT_FILE).toUri());job.setInputFormatClass(KeyValueTextInputFormat.class);job.setMapperClass(HotComment3Mapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);job.setSortComparatorClass(HotComment3Comparator.class);job.setGroupingComparatorClass(HotCommentGroup3Comparator.class);job.setReducerClass(HotComment3Reducer.class);}

二、mapper加载文件数据，计算TF-IDF

    1、在setUp中加载推送节点本地的part-r-00003和part-r-00004文本里的数据

    2、计算IDF

    3、计算TF-IDF=TF*IDF

public class HotComment3Mapper extends Mapper<Text, Text, Text, Text> {private double fileTotal = 0;private Map<String, Double> wordFileCount = new HashMap<>();private final Text mkey = new Text();private final Text mval = new Text();private final NumberFormat nf = NumberFormat.getInstance();@Overrideprotected void setup(Context context) throws IOException, InterruptedException {//设置double取5位小数nf.setMaximumFractionDigits(5);//从各节点服务器本地读取part-r-00003和part-r-00004文件URI[] uris = context.getCacheFiles();if (uris != null && uris.length > 0) {for (URI uri :uris) {String file = uri.getPath();boolean isFileTotal;if (file.endsWith(HotCommentDriver.TOTAL_FILE)) {isFileTotal = true;} else if (file.endsWith(HotCommentDriver.WORD_COUNT_FILE)) {isFileTotal = false;} else {continue;}//本地跑需要配置本地文件路径file = "G:\\学习\\大数据\\hadoop\\项目\\tf-idf" + file.substring(5);BufferedReader reader = new BufferedReader(new FileReader(file));String line;try {if (isFileTotal) {line = reader.readLine();StringTokenizer st = new StringTokenizer(line, "\t");st.nextToken();fileTotal = Double.parseDouble(st.nextToken());} else {while (reader.ready()) {line = reader.readLine();StringTokenizer st = new StringTokenizer(line, "\t");wordFileCount.put(st.nextToken(), Double.parseDouble(st.nextToken()));}}} catch (IOException e) {e.printStackTrace();} catch (NumberFormatException e) {e.printStackTrace();} finally {reader.close();}}}}@Overrideprotected void map(Text key, Text value, Context context) throws IOException, InterruptedException {//输入样本数据//好听_10033640	0.14285714285714285FileSplit fs = (FileSplit) context.getInputSplit();String filename = fs.getPath().getName();if (filename.contains(HotCommentDriver.TOTAL_FILE) || filename.contains(HotCommentDriver.WORD_COUNT_FILE)) {return;}double tf = Double.parseDouble(value.toString());StringTokenizer st = new StringTokenizer(key.toString(), "_");String word = st.nextToken();String id = "";while (st.hasMoreTokens()) {id = st.nextToken();}Double wfc = wordFileCount.get(word);if (wfc == null) {wfc = 1.0;}//计算idfdouble idf = Math.log(fileTotal/wfc);double tf_idf= tf * idf;mkey.set(id + "_" + nf.format(tf_idf));mval.set(word);context.write(mkey, mval);}
}

三、排序比较器，先根据文件id排序，同id内按TF-IDF值倒序

public class HotComment3Comparator extends WritableComparator {public HotComment3Comparator() {super(Text.class, true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {StringTokenizer sta = new StringTokenizer(a.toString(), "_");StringTokenizer stb = new StringTokenizer(b.toString(), "_");int i = sta.nextToken().compareTo(stb.nextToken().toString());if (i == 0) {double ad = Double.parseDouble(sta.nextToken());double ab = Double.parseDouble(stb.nextToken());return Double.compare(ab, ad);}return i;}
}

四、组比较器，自定义根据文件id分组

public class HotCommentGroup3Comparator extends WritableComparator {public HotCommentGroup3Comparator() {super(Text.class, true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {StringTokenizer sta = new StringTokenizer(a.toString(), "_");StringTokenizer stb = new StringTokenizer(b.toString(), "_");return sta.nextToken().compareTo(stb.nextToken());}
}

五、reducer统计每个文件的所有分词

public class HotComment3Reducer extends Reducer<Text, Text, Text, Text> {@Overrideprotected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {//输入数据样例：//7103300_0.03846	风月String id = "";StringBuffer sb = new StringBuffer();for (Text value : values) {StringTokenizer st = new StringTokenizer(key.toString(), "_");id = st.nextToken();sb.append(value.toString()).append(":").append(st.nextToken()).append("\t");}context.write(new Text(id), new Text(sb.toString()));}
}

六、输出最终结果，数据格式如下

315556648	版权:0.3883	购买:0.1551	jw15:0.1551	发行:0.1551	收录于:0.1551	歌曲:0.1448	熟知:0.14227	当中:0.14227	商业:0.13476	五音:0.13476	16年:0.12943	喜爱:0.1253	许多人:0.1253	聆:0.1253	官方:0.1253	宣传:0.11907	问题:0.11907	尊重:0.11441	做过:0.11441	剪辑:0.11246	任何:0.1107	背景:0.10623	天涯:0.10376	收:0.10264	创作:0.10158	原创:0.10158	不在:0.10058	专辑:0.09704	和:0.09476	明月:0.09407	支持:0.09151	太多:0.09092	刀:0.09035	年:0.0898	游戏:0.08381	音:0.07838	下:0.0778	这是:0.07039	曲:0.0702	天:0.06554	因为:0.06195	被:0.06089	而:0.06	为:0.05755	首歌:0.04929	在:0.04334	这:0.03559	是:0.02985	的:0.01892	
315600154	蛋卷:0.38119	两首歌:0.20499	支持:0.1392	发:0.13347	抽奖:0.11797	安靖:0.11797	会受:0.11797	咯:0.11797	奖品:0.11797	网易:0.10856	波及:0.1082	这也:0.10249	复杂:0.10249	云:0.09882	5:0.09844	经过:0.09273	羽:0.09273	生了:0.09273	抽:0.09273	小伙伴:0.09273	ps:0.09056	同意:0.09056	上传:0.09056	算是:0.08868	来吧:0.08868	没有:0.08813	赶紧:0.08702	并没有:0.08554	本人:0.08419	选择:0.08419	原因:0.08419	婶:0.08419	怎么样:0.08297	珍惜:0.08184	下载:0.0808	事情:0.07892	并不是:0.0765	然而:0.07509	不管:0.07263	好好:0.07208	这些:0.0696	应该:0.0696	可是:0.06054	下:0.05917	所以:0.05643	大家:0.05539	到了:0.05443	这:0.05414	已经:0.05383	吧:0.05089	但是:0.05066	被:0.04631	还是:0.04348	很:0.04259	不:0.03456	他:0.03246	了:0.02335	的:0.02158	我:0.01443	
315923049	识:0.69893	阙:0.65839	幸:0.61784	昭:0.6073	一首:0.44438	诗:0.43866	红:0.42981	音:0.42325	愿:0.39097	听:0.26073	

七、完整代码及测试数据详见码云：hadoop-test传送门