Hadoop MapReduce--实现获取最大值和最小值

根据txt文档，获取age的最值

前言

例如：随着大数据的不断发展，hadoop这门技术也越来越重要，很多人都开启了学习大数据，本文就如何在海量数据中获取最值提供了思路。

提示：以下是本篇文章正文内容，下面案例可供参考

一、txt数据准备

python中有random和faker包（外部）给我们提供假的数据。我们使用python创建一个小型的txt文档，其中包括姓名，年龄，score(1分制)

以下是创建的txt文档（按照\t分行）：

rose    27    0.6270426084076096
lisa    27    0.7321873119280536
black    22    0.8129257516728405
jack    27    0.8328363834853498
rose    22    0.9685109182738815
black    30    0.7727688951446979
lisa    21    0.24477509026919908
black    25    0.8785838355517068
rose    30    0.8901461639299959
black    26    0.5460709396256507

1.代码设计

1.创建对象并对其序列化。对象属性有min,max

2.mapper阶段对数据进行分割处理，context写入（StuBean,Text）的数据.

3.reduce阶段对mapper阶段的数据进行处理：获取最值。context写入（StuBean，Text）的数据

4.Driver实现整个流程

2.代码实现

StuBean代码如下（示例）：

package com.atguigu.mapreduce.MaxMinscore;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//input--排序
public class StuBean implements Writable {

    public int max;
    public int min;

    public int getMax() {
        return max;
    }

    public void setMax(int max) {
        this.max = max;
    }

    public int getMin() {
        return min;
    }

    public void setMin(int min) {
        this.min = min;
    }

    @Override
    public String toString() {
        return  max+"\t"+min;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(max);
        out.writeInt(min);
    }
    //     序列化读取
    @Override
    public void readFields(DataInput in) throws IOException {
        this.max =in.readInt();
        this.min =in.readInt();
    }

}

2.mapper阶段代码

import java.io.IOException;

//根据input类型，输出Text的主键,stuBean的数据
public class MinMaxMapper extends Mapper<Object,Text, Text,StuBean> {

    private StuBean sb=new StuBean();

    @Override
    protected void map(Object key, Text value, Mapper<Object, Text, Text, StuBean>.Context context) throws IOException, InterruptedException {
        //        获取一行
        String strs=value.toString();
        String[] split = strs.split("\t");
        String data=split[0];
        if (data==null){
            return ;
        }
        sb.setMin(Integer.parseInt(split[1]));
        sb.setMax(Integer.parseInt(split[1]));

        context.write(new Text(data),sb);
    }
}

3.reduce实现

package com.atguigu.mapreduce.MaxMinscore;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
//将mapper阶段的数据转化成视图bean和int
public class StuReducer extends Reducer<Text,StuBean,Text,StuBean> {

    private StuBean sb=new StuBean();
    @Override
    protected void reduce(Text key, Iterable<StuBean> values, Reducer<Text, StuBean, Text, StuBean>.Context context) throws IOException, InterruptedException {

        sb.setMax(0);
        sb.setMin(0);
        for (StuBean value:values){
            if (sb.getMin() == 0 ||value.getMin() <sb.getMin() ){
                    sb.setMin(value.getMin());
            }
            if (sb.getMax() == 0 ||value.getMax() >sb.getMax() ){
                 sb.setMax(value.getMax());
            }
        }
        context.write(key,sb);
    }
}

4.driver实现

package com.atguigu.mapreduce.MaxMinscore;

import org.apache.hadoop.conf.Configuration;
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 java.io.IOException;
/*
只排序
 */
public class StuDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 1 获取配置信息以及获取job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 2 关联本Driver程序的jar
        job.setJarByClass(StuDriver.class);

        // 3 关联Mapper和Reducer的jar
        job.setMapperClass(MinMaxMapper.class);
        job.setReducerClass(StuReducer.class);

        // 4 设置Mapper输出的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(StuBean.class);

        // 5 设置最终输出kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(StuBean.class);

        // 6 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path("D:\\desk\\ac.txt"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\desk\\ac2"));

        // 7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);

    }
    }

总结

提示：这里对文章进行总结：
例如：以上就是今天要讲的内容，本文仅仅简单介绍了hadoop mapreduce操作最值的基本使用，而mapre提供了大量能使我们快速便捷地处理数据的函数和方法。