大数据技术-Hadoop（三）Mapreduce的介绍与使用

一、概念和定义

二、WordCount案例

1、WordCountMapper

2、WordCountReducer

3、WordCountDriver

三、序列化

1、为什么序列化

2、为什么不用Java的序列化

3、Hadoop序列化特点：

4、自定义bean对象实现序列化接口（Writable）

4.1、bean

4.2、FlowBeanMapper

4.3、FlowReducer

4.4、FlowDriver

四、MapReduce框架原理

1、mapreduce流程

2、Shuffle机制

3、Partion分区

3.1、默认分区方法

3.2、自定义分区

4、WritableComparable

5、Combiner合并

6、自定义FileOutputFormat

7、Reduce Join

8、数据清洗 ETL

五、数据压缩

1、参数说明

2、代码示例

六、完整代码

七、参考

一、概念和定义

请看 https://blog.csdn.net/weixin_48935611/article/details/137856999，这个文章概括的很全面，本文主要展示MapReduce的使用。

二、WordCount案例

1、WordCountMapper

package com.xiaojie.hadoop.mapreduce.wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: TODO
 * @date 2024/12/27 9:00
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    Text kOut = new Text();
    IntWritable vOut = new IntWritable(1);

    /**
     * @param key     偏移量
     * @param value   文本值
     * @param context 上下文
     * @description:
     * @return: void
     * @author 熟透的蜗牛
     * @date: 2024/12/27 9:01
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
//        hello world
//        hello mapreduce
//        hello haddop
//        hadoop
//        java
//        mysql
//        mysql orcale
        /**
         这里输出的结果为（hello,1)(world,1)(hello,1) (mapreduce,1)(hello,1)......
         */


        //获取一行,输入的内容
        String line = value.toString();
        //分隔
        String[] words = line.split(" ");
        for (String word : words) {
            kOut.set(word);
            //kout 即为单词 vout 单词出现的次数
            context.write(kOut, vOut);
        }

    }
}

2、WordCountReducer

package com.xiaojie.hadoop.mapreduce.wordcount;

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

import java.io.IOException;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: reduce把map的输出当作输入
 * @date 2024/12/27 9:17
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    int sum;
    IntWritable v = new IntWritable();

    /**
     * @param key     map 输出的key kOut
     * @param values  map输出的value Vout
     * @param context
     * @description:
     * @return: void
     * @author 熟透的蜗牛
     * @date: 2024/12/27 9:22
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //累加求和，合并map传递过来的值
        sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        //输出结果
        v.set(sum);
        context.write(key, v);
    }
}

3、WordCountDriver

package com.xiaojie.hadoop.mapreduce.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: TODO
 * @date 2024/12/27 9:23
 */
public class WordCountDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        // 1 获取配置信息以及获取job对象
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        // 2 关联本Driver程序的jar
        job.setJarByClass(WordCountDriver.class);
        // 3 关联Mapper和Reducer的jar
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

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

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 6 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path("D:\\hadoop\\hello.txt"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\wordcount"));

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

    }
}

三、序列化

1、为什么序列化

一般来说，“活的”对象只生存在内存里，关机断电就没有了。而且“活的”对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。然而序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。

2、为什么不用Java的序列化

Java的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。所以，Hadoop自己开发了一套序列化机制（Writable）。

3、Hadoop序列化特点：

（1）紧凑：高效使用存储空间。
（2）快速：读写数据的额外开销小。
（3）互操作：支持多语言的交互

4、自定义bean对象实现序列化接口（Writable）

4.1、bean

package com.xiaojie.hadoop.mapreduce.flow;

import org.apache.hadoop.io.Writable;

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

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: 定义一个bean 实现 writable接口
 * @date 2024/12/27 10:25
 */
public class FlowBean implements Writable {

    private long upFlow; //上行流量
    private long downFlow; //下行流量
    private long sumFlow; //总流量

    //创建无参构造函数
    public FlowBean() {
    }

    //创建gettter setter 方法
    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 getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }

    //重写setSumFlow 方法
    public void setSumFlow() {
        this.sumFlow = this.upFlow + this.downFlow;
    }

    //重写序列化方法，输出和输入的顺序要保持一致
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);

    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();

    }

    //结果显示在文本中，重写tostring 方法，
    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;

    }
}

4.2、FlowBeanMapper

package com.xiaojie.hadoop.mapreduce.flow;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: 流量mapper
 * @date 2024/12/27 10:32
 */
public class FlowBeanMapper extends Mapper<LongWritable, Text, Text, FlowBean> {

    //定义一个输出的key
    private Text outKey = new Text();
    //定义输出的value 即 FlowBean
    private FlowBean outValue = new FlowBean();

    /**
     * @param key     map的输入值偏移量
     * @param value   map 的输入value
     * @param context
     * @description:
     * @return: void
     * @author 熟透的蜗牛
     * @date: 2024/12/27 10:35
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {

        //获取一行数据
        String line = value.toString();
        //切割数据
        String[] split = line.split("\t");

        //抓取我们需要的数据:手机号,上行流量,下行流量
        String phone = split[1];  //手机号
        //上行流量 ，由于有的数据没有，这里从后面取值
        Long upFlow = Long.parseLong(split[split.length - 3]);
        Long downFlow = Long.parseLong(split[split.length - 2]);

        //封装输出结果
        //设置输出的key
        outKey.set(phone);
        //设置输出的value
        outValue.setUpFlow(upFlow);
        outValue.setDownFlow(downFlow);
        outValue.setSumFlow();
        //写出outK outV
        context.write(outKey, outValue);
    }
}

4.3、FlowReducer

package com.xiaojie.hadoop.mapreduce.flow;

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

import java.io.IOException;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: 定义流量输出的reduce
 * @date 2024/12/27 10:46
 */
public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

    private FlowBean finalOutV = new FlowBean();

    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {

        long totalUp = 0;
        long totalDown = 0;
        //遍历values,将其中的上行流量,下行流量分别累加
        for (FlowBean bean : values) {
            totalUp += bean.getUpFlow();
            totalUp += bean.getDownFlow();
        }
        //封装输出结果
        finalOutV.setUpFlow(totalUp);
        finalOutV.setDownFlow(totalDown);
        finalOutV.setSumFlow();
        //输出结果
        context.write(key, finalOutV);

    }
}

4.4、FlowDriver

package com.xiaojie.hadoop.mapreduce.flow;

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;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: 驱动
 * @date 2024/12/27 10:55
 */
public class FlowDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        //获取job对象
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        //设置jar
        job.setJarByClass(FlowDriver.class);

        //设置manpper 和reducer
        job.setMapperClass(FlowBeanMapper.class);
        job.setReducerClass(FlowReducer.class);

        //设置map输出kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //设置最终输出结果kv
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        //设置输入输出路径
        FileInputFormat.setInputPaths(job, new Path("d://hadoop//phone.txt"));
        FileOutputFormat.setOutputPath(job, new Path("d://hadoop//phone"));

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

    }
}

四、MapReduce框架原理

1、mapreduce流程

直观的效果，图片来自 https://blog.csdn.net/weixin_48935611/article/details/137856999

2、Shuffle机制

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle

（1）MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中

（2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件

（3）多个溢出文件会被合并成大的溢出文件

（4）在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序

（5）ReduceTask根据自己的分区号，去各个MapTask机器上拉取相应的结果分区数据

（6）ReduceTask会抓取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）

（7）合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）

注意：

（1）Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。

（2）缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb默认100M。