如何理解TopK算法及其实现
今天就跟大家聊聊有关如何理解TopK算法及其实现,可能很多人都不太了解,为了让大家更加了解,小编给大家总结了以下内容,希望大家根据这篇文章可以有所收获。
1、问题描述
在大数据规模中,经常遇到一类需要求出现频率最高的K个数,这类问题称为“TOPK”问题!例如:统计歌曲中最热门的前10首歌曲,统计访问流量最高的前5个网站等。
2、例如统计访问流量最高的前5个网站:
数据test.data文件:
数据格式解释:域名 上行流量 下行流量
思路:
1、Mapper每解析一行内容,按照"\t"获取各个字段
2、因为URL有很多重复记录,所以将URL放到key(通过分析MapReduce原理),流量放在value
3、在reduce统计总流量,通过TreeMap进行对数据进行缓存,最后一并输出(值得注意的是要一次性输出必须要用到Reduce类的cleanup方法)
程序如下:
Mapper类:
package com.itheima.hadoop.mapreduce.mapper; import java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Counter; import com.itheima.hadoop.mapreduce.bean.FlowBean; public class TopKURLMapper extends Mapper<LongWritable, Text, Text, FlowBean> { /** * @param key * : 每一行偏移量 * @param value * : 每一行的内容 * @param context * : 环境上下文 */ @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { /** * 该计数器是org.apache.hadoop.mapreduce.Counter */ Counter counter = context .getCounter("ExistProblem", "ExistProblemLine"); // 自定义存在问题的行错误计数器 String line = value.toString(); // 读取一行数据 String[] fields = line.split("\t"); // 获取各个字段,按照\t划分 try { String url = fields[0]; // 获取URL字段 long upFlow = Long.parseLong(fields[1]); // 获取上行流量(upFlow)字段 long downFlow = Long.parseLong(fields[2]); // 获取下行流量(downFlow)字段 FlowBean bean = new FlowBean(upFlow, downFlow); // 将上行流量和下行流量封装到bean中 Text tUrl = new Text(url); // 将java数据类型转换hadoop数据类型 context.write(tUrl, bean); // 传递的数据较多,封装到bean进行传输(tips:bean传输时需要注意序列化问题) } catch (Exception e) { e.printStackTrace(); counter.increment(1); // 记录错误行数 } } }
Reduce类:
package com.itheima.hadoop.mapreduce.reducer; import java.io.IOException; import java.util.Map.Entry; import java.util.TreeMap; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import com.itheima.hadoop.mapreduce.bean.FlowBean; public class TopKURLReducer extends Reducer<Text, FlowBean, FlowBean, Text> { private TreeMap<FlowBean, Text> treeMap = new TreeMap<FlowBean, Text>(); /** * @param key * : 每一行相同URL * @param values * : 总流量bean */ @Override public void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { long countUpFlow = 0; long countDownFlow = 0; /* * 1、取出每个bean的总流量 2、统计多个bean的总流量 3、缓存到treeMap中 */ for (FlowBean bean : values) { countUpFlow += bean.getUpFlow(); // 统计上行流量 countDownFlow += bean.getDownFlow(); // 统计下行总流量 } // 封装统计的流量 FlowBean bean = new FlowBean(countUpFlow, countDownFlow); treeMap.put(bean, new Text(key)); // 缓存到treeMap中 } @Override public void cleanup(Context context) throws IOException, InterruptedException { //遍历缓存 for (Entry<FlowBean,Text> entry : treeMap.entrySet()) { context.write(entry.getKey(), entry.getValue()); } super.cleanup(context); // 不能动原本的销毁操作 } }
FlowBean类:
package com.itheima.hadoop.mapreduce.bean; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import org.apache.hadoop.io.Writable; public class FlowBean implements Writable, Comparable<FlowBean> { private long upFlow; private long downFlow; private long maxFlow; @Override public String toString() { return upFlow + "\t" + downFlow + "\t" + maxFlow; } /** * 1、序列化注意的问题,序列化需要默认的构造方法(反射) 2、在readFields()和write()方法中,应该遵循按照顺序写出和读入 */ public FlowBean() { } public FlowBean(long upFlow, long downFlow) { this.upFlow = upFlow; this.downFlow = downFlow; this.maxFlow = upFlow + downFlow; } public long getUpFlow() { return upFlow; } public void setUpFlow(long upFlow) { this.upFlow = upFlow; } public long getDownFlow() { return downFlow; } public void setDownFlow(long downFlow) { this.downFlow = downFlow; } public long getMaxFlow() { return maxFlow; } public void setMaxFlow(long maxFlow) { this.maxFlow = maxFlow; } @Override public void readFields(DataInput dataIn) throws IOException { upFlow = dataIn.readLong(); downFlow = dataIn.readLong(); maxFlow = dataIn.readLong(); } @Override public void write(DataOutput dataOut) throws IOException { dataOut.writeLong(upFlow); dataOut.writeLong(downFlow); dataOut.writeLong(maxFlow); } @Override public int compareTo(FlowBean o) { return this.maxFlow > o.maxFlow ? -1 : this.maxFlow < o.maxFlow ? 1 : 0; } }
驱动类:
package com.itheima.hadoop.drivers; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import com.itheima.hadoop.mapreduce.bean.FlowBean; import com.itheima.hadoop.mapreduce.mapper.TopKURLMapper; import com.itheima.hadoop.mapreduce.reducer.TopKURLReducer; public class TopKURLDriver extends Configured implements Tool{ @Override public int run(String[] args) throws Exception { /** * 1、创建job作业 * 2、设置job提交的Class * 3、设置MapperClass,设置ReduceClass * 4、设置Mapper和Reduce各自的OutputKey和OutputValue类型 * 5、设置处理文件的路径,输出结果的路径 * 6、提交job */ Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(TopKURLRunner.class); job.setMapperClass(TopKURLMapper.class); job.setReducerClass(TopKURLReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(FlowBean.class); job.setOutputKeyClass(FlowBean.class); job.setOutputValueClass(Text.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job,new Path(args[1])); //参数true为打印进度 return job.waitForCompletion(true)?0:1; } }
package com.itheima.hadoop.runner; import org.apache.hadoop.util.ToolRunner; import com.itheima.hadoop.runner.TopKURLRunner; public class TopKURLRunner { public static void main(String[] args) throws Exception { int res = ToolRunner.run(new TopKURLRunner(), args); System.exit(res); } }
运行命令:hadoop jar topkurl.jar com.itheima.hadoop.drives.TopKURLDriver /test/inputData /test/outputData
运行结果:
看完上述内容,你们对如何理解TopK算法及其实现有进一步的了解吗?如果还想了解更多知识或者相关内容,请关注行业资讯频道,感谢大家的支持。