MapReduceTopK TreeMap

版权声明：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

 * [email protected]

 */

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

 *[email protected]

 */

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 + "\t" + key + "\r";

				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("\t");

			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);

		}

	}

}