大数据Hadoop学习之——TF-IDF算法实现

矫情吗；* 2022-11-29 05:48 196阅读 0赞

# 一、算法说明 #

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);
    
    	@Override
    	protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
    		//样本数据：5824431 我大剑三不负基三盛名，听的我都醉了
    
    		//计算词频IF
    		StringReader 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来分词，需要引用对应依赖

<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> {
    
    	@Override
    	public 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> {
    
    	@Override
    	protected 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);
    
    	@Override
    	protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
    		//数据样本
    		//台词_40270056	1
    		//10247087	48
    
    		FileSplit fs = (FileSplit) context.getInputSplit();
    		if (!fs.getPath().getName().contains("part-r-00003")) {
    			//part-r-00003的数据不处理，一个分区对应一个mapper task
    			StringTokenizer 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> {
    
    	@Override
    	public 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()) {
    				//分割后取最后一个，肯定是id
    				id = 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);
    	}
    
    	@Override
    	public 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);
    	}
    
    	@Override
    	public 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();
    
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    		// 输入数据样本，有三类数据：
    		// 第一组
    		// 10247087	48
    		// 能_10247087	1
    		// 相逢_10247087	1
    		// 第二组
    		//能	0
    		//能	0
    		//第三组
    		//count 	4340
    
    		if (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，做格式兼容，默认fase
    		job.getConfiguration().set("mapreduce.app-submission.cross-platform", "true");
    		//运行平台，这个可以不用设置，默认为yarn
    		conf.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());
    		//把词对应出现的文件数数据加载到job
    		job.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中需要加载上一步的两个结果集，如果是在hadoop集群中运行，这里需要设置两个结果集的路径，hadoop会在程序运行时将对应的文件发送到对应节点的主机上。并windows客户端提交job的话，需要本地先打好jar包，并制定jar包路径。如果实在本地运行程序，则可以再mapper中直接读本地的结果集文件，当然也可以读hdfs上的结果文件。

二、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();
    
    	@Override
    	protected 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();
    				}
    			}
    		}
    	}
    
    	@Override
    	protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
    		//输入样本数据
    		//好听_10033640	0.14285714285714285
    
    		FileSplit 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;
    		}
    		//计算idf
    		double 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);
    	}
    
    
    	@Override
    	public 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);
    	}
    
    	@Override
    	public 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> {
    
    	@Override
    	protected 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传送门][hadoop-test]

[hadoop-test]: https://gitee.com/jlzl/hadoop-test