MapReduce编程案例五

2019-08-22

Mapreduce经典案例之倒排索引创建

题目描述

案例需求实现统计某目录下单词在不同文件的出现的次数如下的结果

数据样例：
需求：有大量的文本文档，如下所示：
a.txt
hello tom
hello jim
hello kitty
hello rose
b.txt
hello jerry
hello jim
hello kitty
hello jack
c.txt
hello jerry
hello java
hello c++
hello c++
需要得到以下结果：
hello a.txt–>4 b.txt–>4 c.txt–>4
java c.txt–>1
jerry b.txt–>1 c.txt–>1
….

处理逻辑：
1 第一步：先求出每个文件中每个单词出现的次数，使用”单词-文件名”作为key,1为value 输出为总次数。
2 第二步：使用上一次结果将单词的作为一组key 文件名–>次数作为value。

vworker在调map方法时，会传入一个context，而context中包含了这个worker所读取的数据切片信息，而切片信息又包含这个切片所在的文件信息
那么，就可以在map中：
FileSplit split = context.getInputSplit();
String fileName = split.getpath().getName();

处理逻辑代码

第一步

package cn.aparke.mr.index;

import java.io.IOException;

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.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * /**
 * 倒排索引创建
 * 案例需求实现统计目录下单词在不同文件的出现的次数 如下的结果
 *c++	c.txt-->2	
  hello	a.txt-->4	b.txt-->4	c.txt-->4	
  jack	b.txt-->1	
  java	c.txt-->1	
  jerry	b.txt-->1	c.txt-->1	
  jim	a.txt-->1	b.txt-->1	
  kitty	b.txt-->1	a.txt-->1	
  rose	a.txt-->1	
  tom	a.txt-->1	
  
 * 实现思路
 * 第一步：先求出每个文件中每个单词出现的次数
 * "单词-文件名"作为key,1为value 输出
 *reduce集合每个单词-文件名 在文件目录下的所有次数
 * 第二步：通过上一次结果 将单词的作为一组key 文件名次数作为value
 *
 * @author aparke
 *
 *补充
 *InputSplit是指分片，在MapReduce当中作业中，作为map task最小输入单位。分片是基于文件基础上出来的而来的概念，
 *通俗的理解一个文件可以切分为多少个片段，每个片段包括了<文件名，开始位置，长度，位于哪些主机>等信息。
 *map task的数量由输入文件总大小和分片大小确定的；hadoop2.2版本hdfs的数据块默认是128M。
 *若一个文件大于128M，通过将大文件分解得到若干个数据块；若一个文件小于128M，则按它的实际大小组块存储；
 */

/*MapReduce Input Split（输入分/切片）如果maptask读的是文件：划分范围应该是文件路径、偏移量范围
 * 
 * map输入   LongWritable  Text
 * 
 * 输出		Text		IntWritable
 * "单词-文件名"作为key,1为value 输出
 */
class IndexStepOneMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	
	Text outKey=new Text();
	IntWritable intWritable=new IntWritable(1);

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		
		// 从输入切片信息中获取当前正在处理的一行数据所属文件
		//context.getInputSplit(); 获取map输入切片
		FileSplit inputSplit = (FileSplit) context.getInputSplit();
		String fileName = inputSplit.getPath().getName();
		
		/*InputSplit类
		 是抽象类，因为他不知道你要实现什么切分方法所以定义为抽象类，选中InputSplit然后按ctrl+t可查看他的全部子类，
		找到适合自己用的方法的子类，然后进行强转成子类，实现子类的方法,拿到文件名
		*/
		String[] word = value.toString().split(" ");
		for (String words : word) {
			outKey.set(words+"-"+fileName);//设置key为文件中的每个单词和 所在的文件名
			context.write(outKey, intWritable);
		}
		
	}

}

class IndexStepOneReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
	IntWritable outValue=new IntWritable();
/*
reduce 输入key 单词-文件名  value 1
		输出     单词-文件名   value 次数
 */
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
		int count=0;
		for (IntWritable value : values) {
			count+=value.get();
			
		}
		outValue.set(count);
		context.write(key, outValue);
	}
	
	
}


public  class IndexStepOne {
	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(IndexStepOne.class);

		job.setMapperClass(IndexStepOneMapper.class);
		job.setReducerClass(IndexStepOneReduce.class);

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

		Path inpath=new Path("d:/testdata/index/input");
		Path outpath=new Path("d:/testdata/index/output");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job, outpath);
		
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion?0:1);
		
	}
	
}

第二步

package cn.aparke.mr.index;

import java.io.IOException;

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.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * /**
 * 倒排索引创建
 * 案例需求实现统计目录下单词在不同文件的出现的次数 如下的结果
 *c++	c.txt-->2	
  hello	a.txt-->4	b.txt-->4	c.txt-->4	
  jack	b.txt-->1	
  java	c.txt-->1	
  jerry	b.txt-->1	c.txt-->1	
  jim	a.txt-->1	b.txt-->1	
  kitty	b.txt-->1	a.txt-->1	
  rose	a.txt-->1	
  tom	a.txt-->1	
  
 * 实现思路
 * 第一步：先求出每个文件中每个单词出现的次数
 * "单词-文件名"作为key,1为value 输出
 *reduce集合每个单词-文件名 在文件目录下的所有次数
 * 第二步：通过上一次结果 将单词的作为一组key 文件名次数作为value
 *
 * @author aparke
 *
 *补充
 *InputSplit是指分片，在MapReduce当中作业中，作为map task最小输入单位。分片是基于文件基础上出来的而来的概念，
 *通俗的理解一个文件可以切分为多少个片段，每个片段包括了<文件名，开始位置，长度，位于哪些主机>等信息。
 *map task的数量由输入文件总大小和分片大小确定的；hadoop2.2版本hdfs的数据块默认是128M。
 *若一个文件大于128M，通过将大文件分解得到若干个数据块；若一个文件小于128M，则按它的实际大小组块存储；
 */

/*MapReduce Input Split（输入分/切片）如果maptask读的是文件：划分范围应该是文件路径、偏移量范围
 * 
 * map输入   LongWritable  Text
 * 
 * 输出		Text		IntWritable
 * "单词-文件名"作为key,1为value 输出
 */
class IndexStepOneMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	
	Text outKey=new Text();
	IntWritable intWritable=new IntWritable(1);

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		
		// 从输入切片信息中获取当前正在处理的一行数据所属文件
		//context.getInputSplit(); 获取map输入切片
		FileSplit inputSplit = (FileSplit) context.getInputSplit();
		String fileName = inputSplit.getPath().getName();
		
		/*InputSplit类
		 是抽象类，因为他不知道你要实现什么切分方法所以定义为抽象类，选中InputSplit然后按ctrl+t可查看他的全部子类，
		找到适合自己用的方法的子类，然后进行强转成子类，实现子类的方法,拿到文件名
		*/
		String[] word = value.toString().split(" ");
		for (String words : word) {
			outKey.set(words+"-"+fileName);//设置key为文件中的每个单词和 所在的文件名
			context.write(outKey, intWritable);
		}
		
	}

}

class IndexStepOneReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
	IntWritable outValue=new IntWritable();
/*
reduce 输入key 单词-文件名  value 1
		输出     单词-文件名   value 次数
 */
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
		int count=0;
		for (IntWritable value : values) {
			count+=value.get();
			
		}
		outValue.set(count);
		context.write(key, outValue);
	}
	
	
}


public  class IndexStepOne {
	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(IndexStepOne.class);

		job.setMapperClass(IndexStepOneMapper.class);
		job.setReducerClass(IndexStepOneReduce.class);

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

		Path inpath=new Path("d:/testdata/index/input");
		Path outpath=new Path("d:/testdata/index/output");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job, outpath);
		
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion?0:1);

	}
	
}

}

项目地址

Mapreduce经典案例