如何理解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算法及其实现有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注亿速云行业资讯频道，感谢大家的支持。