网站标题符号有哪些,个人网站尺寸,做网站大优惠,沧州市有建网站的吗一、简介
求TopK是算法中最常使用到的#xff0c;现在使用Mapreduce在海量数据中统计数据的求TopK。 二、例子
#xff08;1#xff09;实例描述 给出三个文件#xff0c;每个文件中都存储了若干个数值#xff0c;求所有数值中的求Top 5。
样例输入#xff1a; …一、简介
求TopK是算法中最常使用到的现在使用Mapreduce在海量数据中统计数据的求TopK。 二、例子
1实例描述 给出三个文件每个文件中都存储了若干个数值求所有数值中的求Top 5。
样例输入 1file1
1
2
3
7
9
-99
2 2file2
11
2
23
17
9
199
22 3file3
21
12
3
17
2
39
12 期望输出
199
39
23
22
21 2问题分析 实现统计海量数据的求TopK不能将所有的数据加载到内存计算只能使用类似外部排序的方式加载一部分数据统计求TopK接着加载另一部分进行统计TopK。
3实现步骤
1Map过程 首先使用默认的TextInputFormat类对输入文件进行处理得到文本中每行的偏移量及其内容。显然Map过程首先必须分析输入的key,value对得到数值然后在mapper中统计单个分块的求TopK。
2Reduce过程 经过map方法处理后Reduce过程将获取每个mapper的求TopK进行统计分行统计出总的TopK。 3关键代码
package com.mk.mapreduce;import org.apache.commons.lang.StringUtils;
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.NullWritable;
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;import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;public class TopK {public static class TopKMapper extends MapperLongWritable, Text, IntWritable, NullWritable {private ListInteger top5 new ArrayList(5);Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {if (StringUtils.isBlank(value.toString())) {System.out.println(空白行);return;}Integer v Integer.valueOf(value.toString().trim());if(top5.size()5){top5.add(v);}else{Integer min Collections.min(top5);if (min v) {top5.remove(min);top5.add(v);}}}Overrideprotected void cleanup(Context context) throws IOException, InterruptedException {for (Integer v : top5)context.write(new IntWritable(v), NullWritable.get());}}public static class TopKReducer extends Reducer IntWritable, NullWritable,IntWritable, NullWritable {private ListInteger top5 new ArrayList(5);Overrideprotected void reduce(IntWritable key, IterableNullWritable values, Context context) throws IOException, InterruptedException {Integer v key.get();if(top5.size()5){top5.add(v);}else{Integer min Collections.min(top5);if (min v) {top5.remove(min);top5.add(v);}}}Overrideprotected void cleanup(Context context) throws IOException, InterruptedException {top5.sort((a,b)-b-a);for (Integer v : top5)context.write(new IntWritable(v), NullWritable.get());}}public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {String uri hdfs://192.168.150.128:9000;String input /topk/input;String output /topk/output;Configuration conf new Configuration();if (System.getProperty(os.name).toLowerCase().contains(win))conf.set(mapreduce.app-submission.cross-platform, true);FileSystem fileSystem FileSystem.get(URI.create(uri), conf);Path path new Path(output);fileSystem.delete(path, true);Job job new Job(conf, TopK);job.setJar(./out/artifacts/hadoop_test_jar/hadoop-test.jar);job.setJarByClass(TopK.class);job.setMapperClass(TopKMapper.class);job.setReducerClass(TopKReducer.class);job.setMapOutputKeyClass(IntWritable.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(IntWritable.class);job.setOutputValueClass(NullWritable.class);FileInputFormat.addInputPaths(job, uri input);FileOutputFormat.setOutputPath(job, new Path(uri output));boolean ret job.waitForCompletion(true);System.out.println(job.getJobName() ----- ret);}
}
