TopK算法及其實現

一、問題描述

在大數據規模中，常常遇到一類須要求出現頻率最高的K個數，這類問題稱爲「TOPK」問題！例如：統計歌曲中最熱門的前10首歌曲，統計訪問流量最高的前5個網站等。

二、例如統計訪問流量最高的前5個網站：

數據test.data文件：

數據格式解釋：域名    上行流量    下行流量

思路：

一、Mapper每解析一行內容,按照"\t"獲取各個字段

二、由於URL有不少重複記錄,因此將URL放到key(經過分析MapReduce原理),流量放在value

三、在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;
        /*
         * 一、取出每一個bean的總流量 二、統計多個bean的總流量 三、緩存到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;
    }
    /**
     * 一、序列化注意的問題,序列化須要默認的構造方法(反射) 二、在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 {
        
        /**
         * 一、建立job做業
         * 二、設置job提交的Class
         * 三、設置MapperClass,設置ReduceClass
         * 四、設置Mapper和Reduce各自的OutputKey和OutputValue類型
         * 五、設置處理文件的路徑,輸出結果的路徑
         * 六、提交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

運行結果：