hadoop2.5.2学习13-MR之新浪微博TF-IDF算法简介

完整代码下载

一、TF_IDF算法简介

TF-IDF（term frequency–inverse document frequency）是一种用于信息检索与数据挖掘的常用加权技术。

TF-IDF是一种统计方法，用以评估一词条对于一个文件集或一个语料库中的其中一份文件的重要程度。词条的重要性随着它在文件中出现的次数成正比增加，但同时会随着它在语料库中出现的频率成反比下降。

二、TF_IDF的原理

  如果一个词条在一个文件中的频率越高(TF)，同时在其他文件中很少出现（IDF），则认为此词条具有很好的类别区分能力，适合用来分类。

TF: 一个词条在一个文件中出现的次数

IDF： 一个词条在多少个文件中出现的次数，在一个文件中出现多次，只计算一次。

N:为总文章数

文档中的词条的权重的公式：

三、算法的MR实现

3.1、原始样例数据

微博ID 微博内容

3823890453754000    哇哈，支持！
3823890482816512    开心过周末，火锅一定要约起来
3823890499663321    今天你约了吗！成千上万的父母相约幼儿园门口…今天是孩子幼儿园的报名日！可怜天下父母心！我们今天相约这里
3823890520592496    朋友相聚，又岂能少了美食的陪伴，放开肚子吃吧，吃饱了再一起减肥吧！
3823890520647794    今天我和一班好朋友约好了一起吃大餐
3823890541633403    小阳，我约踏青呢！
3823890558102894    今天我约了豆浆，油条。约了电饭煲几小时后饭就自动煮好，还想约豆浆机，让我早晨多睡一小时，不要那么辛苦，起床就就可以喝上香喷喷的豆浆。
3823890591683238    今天约孩子他干妈来我家做饼干
3823890621005004    今年的春天来得格外早，这么好的天气，当然要约上老公和闺女一起去踏春了 赏赏花 看看鱼 好不惬意的说
3823890629460577    路过，来冒个跑，加油
3823890629779115    我们约好了明天再去一次小学怀念一下校园生活
3823890642051141    约吃、约喝、约玩
3823890671749747    今天约了我家小女生去书店买文具咯~
3823890675629620    终于有点春天的感觉了！踏青约起来！

二、算法实现

  首先根据上面的公式，需要计算的值有三个：

TF（词频） 每个词条在单独文件中出现的次数, 注意此处是词条在每个微博的词频， 而不是所有文件中的词频

N 微博总数

DF （逆向文件频率） 词条在多少个文件中存在。

接下来进行算法的实现

2.1、第一个mapreduce

首先我们在第一个mapreduce中计算TF, N ，因为将原始数据进行遍历， 就可以得到微博微博总数， 不需要考虑各篇微博之间的关系， 对于每篇微博， 将他的内容提取出来， 使用词条分词器，进行统计TF

2.1.1 mapper

一个微博对应一个maptask， 进行处理， 计算出微博数， 及每篇微博的TF:

通过词条分词器，获取词条， 就可以在该篇微博中每个词条出现的次数。

为什么输出的可以是词条+微博ID:

  因为TF是计算词条在每篇微博中出现的次数，有区分在不同微博中出现的次数。

如果输出key不使用w_id， 而是使用w, 在shuffle阶段的分组就会将所有相同w分到一组（默认分组是按照key是否相同）,那么统计的TF就是所有微博中出现的总数,和实际不符合。

StringReader sr = new StringReader(content);
//词条拆分器
IKSegmenter ikSegmenter = new IKSegmenter(sr, true);
Lexeme word = null;

while((word = ikSegmenter.next()) != null){
String w = word.getLexemeText();
//统计词频
context.write(new Text(w+"_"+id), new IntWritable(1));
}

同时，要对微博的数量进行统计，输出key为”count”, value为1

在reduce中进行求和。

//每一行数据就是一个微博的id,  内容
context.write(new Text("count"), new IntWritable(1));

2.1.2 分区

因为我们的微博总数最终记录，只有一条数据，如果按照自定义分区， 按照key的hashcode值模reduceTaskNum, 那么微博总数的记录和TF的统计输出会在同一个文件中， 对我们后面的计算造成麻烦。

进行自定义分区， 将微博总数的统计送一个单独一个reduce中处理， 最终输出的微博总数只会在一个文件中， 有一条数据。

只要key是count的数据送到最后一个reuce中处理，

if (key.toString().equals("count")) {
return numReduceTask - 1;
}

TF统计送到其他的reuceTask中处理。本mapreuce中设置四个reduce, 前三个是统计TF值。

package com.chb.weibo1;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
/**
* 自定一个分区，
* 为了将TF和微博总数的统计生成在不同的文件中
* @author 12285
*/
public class FirstPartitioner extends HashPartitioner<Text, IntWritable>{
@Override
public int getPartition(Text key, IntWritable value, int numReduceTasks) {
if (key.toString().equals("count")) {
return numReduceTask - 1;
}else {
return super.getPartition(key, value, numReduceTasks - 1);
}
}

}

2.1.3 reducer

在reduce中就比较简单， TF和微博总数分在不同的reudce中处理， 只需要进行数据的累加。

package com.chb.weibo1;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class FirstReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable iw : values) {
sum += iw.get();
}
context.write(key, new IntWritable(sum));
}
}

2.1.4第一个mapreduce的执行。

package com.chb.weibo1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FirstRunJob {
public static void main(String[] args) throws Exception {
System.setProperty("HADOOP_USER_NAME", "chb");
Configuration conf = new Configuration();

FileSystem fs = FileSystem.get(conf);
Job job = Job.getInstance();
job.setJobName("First Job");
job.setJar("C:\\Users\\12285\\Desktop\\weibo.jar");
job.setJarByClass(FirstRunJob.class);

job.setMapperClass(FirstMapper.class);
job.setReducerClass(FirstReducer.class);
job.setPartitionerClass(FirstPartitioner.class);

job.setNumReduceTasks(4);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

job.setInputFormatClass(KeyValueTextInputFormat.class);

Path in = new Path("/user/chb/input/weibo/");
FileInputFormat.addInputPath(job, in);
Path out = new Path("/user/chb/output/outweibo1");
if (fs.exists(out)) {
fs.delete(out, true);
}
FileOutputFormat.setOutputPath(job, out);
boolean f = job.waitForCompletion(true);
if (f) {
System.out.println("job执行成功！");
}

}
}

结果

总共设置4个reduce，



TF为w_id, 词频数，中间以制表符分割

0.03_3824246315213843   2
0.25_3824246315213843   2
0.33斤_3824020946550570  2
0.83斤_3823927795601059  2
0.88斤_3823953501983192  2
002244_3824015577543437 2
002244_3824025388441161 2
002245_3824015577543437 2
002245_3824025388441161 2
002250_3824015577543437 2
002250_3824025388441161 2

最后一个为微博总数， 只有一条数据：