版权声明： https://blog.csdn.net/zhangxiango/article/details/33319281
MapReduce TopK统计加排序中介绍的TopK在mapreduce的实现。
本案例省略的上面案例中的Sort步骤，改用TreeMap来实现获取前K个词
package TopK1;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
* 统计词频
* @author zx
* zhangxian1991@qq.com
*/
public class WordCount {
/**
* 读取单词
* @author zx
*
*/
public static class Map extends Mapper<Object,Text,Text,IntWritable>{
IntWritable count = new IntWritable(1);
@Override
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer st = new StringTokenizer(value.toString());
while(st.hasMoreTokens()){
String word = st.nextToken().replaceAll(""", "").replace("'", "").replace(".", "");
context.write(new Text(word), count);
}
}
}
/**
* 统计词频
* @author zx
*
*/
public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable>{
@SuppressWarnings("unused")
@Override
protected void reduce(Text key, Iterable<IntWritable> values,Context context)
throws IOException, InterruptedException {
int count = 0;
for (IntWritable intWritable : values) {
count ++;
}
context.write(key,new IntWritable(count));
}
}
@SuppressWarnings("deprecation")
public static boolean run(String in,String out) throws IOException, ClassNotFoundException, InterruptedException{
FileUtil.deleteFile(out);
Configuration conf = new Configuration();
Job job = new Job(conf,"WordCount1");
job.setJarByClass(WordCount.class);
job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
// 设置Map输出类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
// 设置Reduce输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 设置输入和输出文件夹
FileInputFormat.addInputPath(job, new Path(in));
FileOutputFormat.setOutputPath(job, new Path(out));
return job.waitForCompletion(true);
}
}
package TopK1;
import java.io.IOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
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.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
/**
*
* @author zx
*zhangxian1991@qq.com
*/
public class TopK {
public static class TopKMap extends Mapper<Object, Text, IntWritable, IntWritable>{
TreeMap<Integer,String> tm = new TreeMap<Integer,String>(new Comparator<Integer>() {
/**
* treeMap中的元素逆序排列
* @param o1
* @param o2
* @return
*/
@Override
public int compare(Integer o1, Integer o2) {
return o2.compareTo(o1);
}
});
int k = 0;
@Override
protected void cleanup(Context context) throws IOException,
InterruptedException {
Configuration conf = context.getConfiguration();
Path topKPath = new Path(conf.get("topKOut"));
FileSystem fs = topKPath.getFileSystem(conf);
FSDataOutputStream fsDOS = fs.create(topKPath);
Iterator<Integer> it = tm.keySet().iterator();
while(it.hasNext()){
Integer key = it.next();
String value = tm.get(key).toString();
String line = value + " " + key + "
";
fsDOS.write(line.getBytes(), 0, line.length());
}
fsDOS.flush();
fsDOS.close();
}
@Override
protected void setup(Context context) throws IOException,
InterruptedException {
k = Integer.parseInt(context.getConfiguration().get("K"));
}
@Override
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String[] parts = value.toString().split(" ");
tm.put(Integer.parseInt(parts[1]),parts[0]);
if(tm.size() > k){
tm.remove(tm.lastKey());
}
}
}
@SuppressWarnings("deprecation")
public static void main(String args[]) throws ClassNotFoundException, IOException, InterruptedException{
if(args.length < 4){
throw new IllegalArgumentException("要有4个參数:1，要统计的文本文件名称。2。统计后的结果路径。3，topK的结果文件夹，4，K");
}
FileUtil.deleteFile(args[2]);
//要统计字数的文本文件名称
String in = args[0];
//统计字数后的结果
String wordCount = args[1];
in = FileUtil.loadFile(wordCount, "TopK1", in);
//假设统计字数的job完毕后就開始求topK
if(WordCount.run(in, wordCount)){
int k = Integer.parseInt(args[3]);
Configuration conf = new Configuration();
FileUtil.deleteFile(args[2]);
conf.set("topKOut", args[2]);
conf.set("K", k+"");
Job job = new Job(conf,"TopK1");
job.setJarByClass(TopK.class);
job.setMapperClass(TopKMap.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(wordCount));
job.setOutputFormatClass(NullOutputFormat.class);
System.exit(job.waitForCompletion(true)?0:1);
}
}
}
package TopK1;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
/**
*
* @author zx
*
*/
public class FileUtil {
/**
* 上传数据文件到hdfs
* @param inputPath
* @param fileName
* @return
* @throws IOException
*/
public static String loadFile(String inputPath,String folder,String fileName) throws IOException{
//获取数据文件的全路径
if(null != folder && !"".equals(folder)){
folder = folder + "/";
}
String srcPathDir = FileUtil.class.getProtectionDomain().getCodeSource().getLocation()
.getFile() + folder + fileName;
Path srcpath = new Path("file:///" + srcPathDir);
Path dstPath = new Path(getJobRootPath(inputPath) + fileName);
Configuration conf = new Configuration();
FileSystem fs = dstPath.getFileSystem(conf);
fs.delete(dstPath, true);
fs.copyFromLocalFile(srcpath, dstPath);
fs.close();
return getJobRootPath(inputPath) + fileName;
}
/**
* 假设路径的最后不包哈“/”就加一个“/”
* @param path
* @return
*/
public static String getJobRootPath(String path){
if(path.lastIndexOf("/") == path.length()-1){
path = path.substring(0, path.lastIndexOf("/"));
}
return path.substring(0, path.lastIndexOf("/")+1);
}
public static void deleteFile(String ...filePath) throws IOException{
Configuration conf = new Configuration();
for (int i = 0; i < filePath.length; i++) {
Path path = new Path(filePath[i]);
FileSystem fs = path.getFileSystem(conf);
fs.delete(path,true);
}
}
}