欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Hadoop入门(二十三)Mapreduce的求TopK程序

程序员文章站 2022-07-07 21:33:50
...

一、简介

求TopK是算法中最常使用到的,现在使用Mapreduce在海量数据中统计数据的求TopK。

 

二、例子

(1)实例描述
给出三个文件,每个文件中都存储了若干个数值,求所有数值中的求Top 5。

样例输入:                                            
1)file1:  

1
2
3
7
9
-99
2


2)file2:  

11
2
23
17
9
199
22


3)file3:  

21
12
3
17
2
39
12


 期望输出:

199
39
23
22
21

 

(2)问题分析
实现统计海量数据的求TopK,不能将所有的数据加载到内存,计算只能使用类似外部排序的方式,加载一部分数据统计求TopK,接着加载另一部分进行统计TopK。

(3)实现步骤

1)Map过程 
    首先使用默认的TextInputFormat类对输入文件进行处理,得到文本中每行的偏移量及其内容。显然,Map过程首先必须分析输入的<key,value>对,得到数值,然后在mapper中统计单个分块的求TopK。

2)Reduce过程 
    经过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 Mapper<LongWritable, Text, IntWritable, NullWritable> {

        private List<Integer> top5 = new ArrayList<>(5);

        @Override
        protected 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);
                }

            }
        }

        @Override
        protected 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 List<Integer> top5 = new ArrayList<>(5);
        @Override
        protected void reduce(IntWritable key, Iterable<NullWritable> 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);
                }

            }

        }

        @Override
        protected 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);
    }
}