MapReduce的详细使用

2023-05-16

MapReduce

文章目录

MapReduce
1、常用数据序列化类型
2、编程规范（三个阶段）
- Mapper阶段
- Reducer阶段
- Driver阶段
3、编程环境准备
4、简单案例(单词统计)
5、序列化
- 序列化概述
- 自定义 bean 对象实现序列化接口（Writable）
- - 步骤
  - 程序（序列化接口）
6、InputFormat 数据输入
- TextInputFormat（默认）
- CombineTextInputFormat
7、Shuffle 机制
- Partition 分区
- - 自定义Partitioner步骤
  - Job驱动类示例
  - 分区总结
- WritableComparable 排序
- - 自定义排序 WritableComparable
- Combiner 合并
- - 介绍
  - Combiner 合并方法一（自定义）
  - Combiner 合并方法二
8、OutputFormat 数据输出
- 自定义输入
9、数据压缩
- 概述
- MR 支持的压缩编码
- 压缩参数配置
- 程序实现压缩
- - Map 输出端采用压缩
  - Reduce 输出端采用压缩

1、常用数据序列化类型

Java 类型	Hadoop Writable 类型
Boolean	BooleanWritable
Byte	ByteWritable
Int	IntWritable
Float	FloatWritable
Long	LongWritable
Double	DoubleWritable
String	Text
Map	MapWritable
Array	ArrayWritable
Null	NullWritable

2、编程规范（三个阶段）

用户编写的程序分成三个部分：Mapper、Reducer 和 Driver

Mapper阶段

用户自定义的Mapper要继承自己的父类
Mapper的输入数据是KV对的形式（KV的类型可自定义）
Mapper中的业务逻辑写在map()方法中
Mapper的输出数据是KV对的形式（KV的类型可自定义）
map()方法（MapTask进程）对每一个<K,V>调用一次

Reducer阶段

用户自定义的Reducer要继承自己的父类
Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
Reducer的业务逻辑写在reduce()方法中
ReduceTask进程对每一组相同k的<k,v>组调用一次reduce()方法

Driver阶段

相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是封装了MapReduce程序相关运行参数的job对象

3、编程环境准备

添加依赖：创建Maven工程，添加依赖（pom.xml）

<dependency>
    <!-- 版本号跟自己的Hadoop版本对应 -->
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.1.4</version>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.30</version>
</dependency>

添加日志：log4j.properties

log4j.rootLogger=INFO, stdout 
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
log4j.appender.logfile=org.apache.log4j.FileAppender 
log4j.appender.logfile.File=target/spring.log 
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout 
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4、简单案例(单词统计)

Driver驱动类

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;

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //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(WordCountReduce.class);

        //4、设置 Mapper 输出的 key-value
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5、设置 Reducer 输出的 key-value
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6、设置 输入 和 输出 路径
        FileInputFormat.setInputPaths(job, new Path("E:\\test\\input\\input001"));
        FileOutputFormat.setOutputPath(job, new Path("E:\\test\\output\\output001"));

        //7、提交 job
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

Mapper类

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;

/**
 * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 *     KEYIN     输入key的类型
 *     VALUEIN   输入value的类型
 *     KEYOUT    输出key的类型
 *     VALUEOUT  输出value的类型
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private Text outKey = new Text();
    private IntWritable outValue = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1、获取一行数据
        String line = value.toString();

        //2、切割
        String[] words = line.split(" ");

        //3、输出
        for (String word : words) {
            outKey.set(word);
            context.write(outKey, outValue);
        }
    }
}

Reducer类

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

/**
 * Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
 *     KEYIN     输入key的类型
 *     VALUEIN   输入value的类型
 *     KEYOUT    输出key的类型
 *     VALUEOUT  输出value的类型
 */
public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    private int sum;
    private IntWritable outValue = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //1、累加求和
        sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        outValue.set(sum);

        //2、输出
        context.write(key, outValue);
    }
}

5、序列化

序列化概述

简介
- 序列化
  - 就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁盘（持久化）和网络传输
- 反序列化
  - 就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换成内存中的对象
序列化特点
- 紧凑：高效使用存储空间
- 快速：读写数据的额外开销小
- 互操作：支持多语言的交互

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

步骤

第一步：bean对象实现 Writable 接口
第二步：反序列化时，需要反射调用空参构造函数，所以必须有空参构造
第三步：重写序列化方法（write方法）
第四步：重写反序列化方法（readFields方法）
第五步：要想把结果显示在文件中，需要重写 toString()，可用"\t"分开，方便后续用
第六步：Driver、Mapper、Reducer 实现类

注意反序列化的顺序和序列化的顺序完全一致

程序（序列化接口）

import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//1、bean对象实现 Writable 接口
public class FlowBean implements Writable {

    private long upFlow;
    private long downFlow;
    private long sumFlow;

    //2、创建空参构造函数
    public FlowBean(){}

    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() {
        this.sumFlow = this.upFlow + this.downFlow;
    }

    //3、重写序列化方法（write方法）
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(this.upFlow);
        out.writeLong(this.downFlow);
        out.writeLong(this.sumFlow);
    }

    //4、重写反序列化方法（readFields方法）
    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }

    //5、要想把结果显示在文件中，需要重写 toString()，可用"\t"分开，方便后续用
    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
}

6、InputFormat 数据输入

TextInputFormat（默认）

TextInputFormat 是默认的 FileInputFormat 实现类
- 按行读取每条记录。
- 键是存储该行在整个文件中的起始字节偏移量， LongWritable 类型。
- 值是这行的内容，不包括任何行终止符（换行符和回车符），Text 类型

CombineTextInputFormat

应用场景
- 用于小文件过多的场景，将多个小文件从逻辑上规划到一个切片中，多个小文件就交给一个 MapTask 处理
虚拟存储切片最大值和最小值设置
- 最大值：CombineTextInputFormat.setMinInputSplitSize(Job job, long size)
- 最小值：CombineTextInputFormat.setMaxInputSplitSize(Job job, long size)
- 注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值（size是）
- 示例（设置切片最大值为 4M）
  - CombineTextInputFormat.setMaxInputSplitSize(job, 4194304); // 4M

程序（在 Job驱动类中添加设置）

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.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //设置 InputFormat 使用 CombineTextInputFormat，它默认用的是 TextInputFormat
        job.setInputFormatClass(CombineTextInputFormat.class);
        //虚拟存储切片最大值设置 20M
        CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);
        
        FileInputFormat.setInputPaths(job, new Path("C:\\test\\input\\input001"));
        FileOutputFormat.setOutputPath(job, new Path("C:\\test\\output\\output001"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

7、Shuffle 机制

Partition 分区

默认分区是根据key的hashCode对ReduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区

自定义Partitioner步骤

第一步：自定义类继承Partitioner，重写getPartition()方法

package com.itfzk.mapreducer.partition;

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

public class PartitionTest extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text text, IntWritable intWritable, int i) {
        int partitioner;
        
        //逻辑处理，确定text写入到哪个分区中
        
        return partitioner;
    }
}

第二步：在Job驱动中，设置自定义Partitioner
- job.setPartitionerClass(Partition类名.class)
第三步：自定义Partition后，在Job驱动中根据自定义Partitioner的逻辑设置相应数量的ReduceTask
- job.setNumReduceTasks(ReduceTask的数量)

Job驱动类示例

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;

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //自定义Partitioner
        job.setPartitionerClass(PartitionTest.class);
        
        //设置相应数量的ReduceTask
        job.setNumReduceTasks(4);

        FileInputFormat.setInputPaths(job, new Path("C:\\test\\input\\input001"));
        FileOutputFormat.setOutputPath(job, new Path("C:\\test\\output\\output002"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

分区总结

如果 ReduceTask的数量 > getPartition的结果数，则会多产生几个空的输出文件part-r-000xx
如果 1 < ReduceTask的数量 < getPartition的结果数，则有一部分分区数据无处安放，会报错
如果 ReduceTask的数量 = 1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个ReduceTask，最终也就只会产生一个结果文件 part-r-00000
分区号必须从零开始，逐一累加

WritableComparable 排序

自定义排序 WritableComparable

创建bean 对象做为 Mapper的输出key 传输，需要实现 WritableComparable 接口重写 compareTo() 方法，就可以实现排序

Mapper的输出key：Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>中的 KEYOUT

import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//实现 WritableComparable
public class FlowBean implements WritableComparable<FlowBean> {
    
    //字段
    //setting、getting
    //toString

    //重写 compareTo()
    @Override
    public int compareTo(FlowBean o) {
        //需要按照哪个字段排序的逻辑
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
    }
}

Combiner 合并

介绍

Combiner是MR程序中Mapper和Reducer之外的一种组件
Combiner组件的父类就是Reducer
Combiner和Reducer的区别在于运行的位置
- Combiner是在每一个MapTask所在的节点运行
- Reducer是接收全局所有Mapper的输出结果
Combiner的意义就是对每一个MapTask的输出进行局部汇总，以减小网络传输量
Combiner应用的前提是不能影响最终的业务逻辑，Combiner的输出kv应该跟Reducer的输入kv类型要对应起来

Combiner 合并方法一（自定义）

第一步：自定义一个 Combiner类 继承 Reducer，重写 Reduce 方法
第二步：在 Job 驱动类中设置
- job.setCombinerClass(Combiner类名.class)

Combiner 合并方法二

当Reducer类的逻辑跟Combiner类的逻辑一样且Combiner的输出kv类型跟Reducer的输入kv类型一样

将 Reducer类作为 Combiner类
- job.setCombinerClass(Reducer类名.class)

8、OutputFormat 数据输出

自定义输入

第一步：创建 OutputFormat 类，继承 FileOutputFormat，重写 RecordWriter() 方法

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class WordCountOutPutFormat extends FileOutputFormat<Text, IntWritable> {
    @Override
    public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        //返回自定义RecordWriter类
        WordCountRecordWriter wordCountRecordWriter = new WordCountRecordWriter(taskAttemptContext);
        return wordCountRecordWriter;
    }
}

第二步：创建 RecordWriter 类，继承 RecordWriter，重写 write、close 方法

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import java.io.IOException;

public class WordCountRecordWriter extends RecordWriter<Text, IntWritable> {

    private FSDataOutputStream fsDataOutputStream1;
    private FSDataOutputStream fsDataOutputStream2;

    //构造函数
    public WordCountRecordWriter(TaskAttemptContext job) throws IOException {
        //按照自己的输出格式，例如：将数据输出到两个文件中(f.log, z.log)
        FileSystem fs = FileSystem.get(job.getConfiguration());
        fsDataOutputStream1 = fs.create(new Path("C:\\test\\f.log"));
        fsDataOutputStream2 = fs.create(new Path("C:\\test\\z.log"));
    }

    //输出方法
    @Override
    public void write(Text text, IntWritable intWritable) throws IOException, InterruptedException {
        //按照自己的逻辑将数据输出到不同的文件中
        if(text.toString().contains("f")){
            fsDataOutputStream1.writeBytes(text.toString() + "\t" + intWritable.get() + "\n");
        }else{
            fsDataOutputStream2.writeBytes(text.toString() + "\t" + intWritable.get() + "\n");
        }
    }

    //关闭
    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        fsDataOutputStream1.close();
        fsDataOutputStream2.close();
    }
}

第三步：在Job驱动中，设置自定义的 OutputFormat
- job.setOutputFormatClass(自定义的OutputFormat类名.class);

9、数据压缩

概述

优点：以减少磁盘 IO、减少磁盘存储空间。
缺点：增加 CPU 开销。
压缩原则
- 运算密集型的 Job，少用压缩
- IO 密集型的 Job，多用压缩

MR 支持的压缩编码

压缩算法对比介绍

压缩格式	Hadoop 自带？	算法	文件扩展名	是否可切片	换成压缩格式后，原来的程序是否需要修改
DEFLATE	是，直接使用	DEFLATE	.deflate	否	和文本处理一样，不需要修改
Gzip	是，直接使用	DEFLATE	.gz	否	和文本处理一样，不需要修改
bzip2	是，直接使用	bzip2	.bz2	是	和文本处理一样，不需要修改
LZO	否，需要安装	LZO	.lzo	是	需要建索引，还需要指定输入格式
Snappy	是，直接使用	Snappy	.snappy	否	和文本处理一样，不需要修改

压缩性能的比较

压缩算法	原始文件大小	压缩文件大小	压缩速度	解压速度
gzip	8.3GB	1.8GB	17.5MB/s	58MB/s
bzip2	8.3GB	1.1GB	2.4MB/s	9.5MB/s
LZO	8.3GB	2.9GB	49.3MB/s	74.6MB/s
Snappy			250MB/s	500MB/s

压缩算法的优缺点比较

压缩算法	优点	缺点
Gzip	压缩率比较高	不支持 Split；压缩/解压速度一般
Bzip2	压缩率高；支持 Split	压缩/解压速度慢
Lzo	压缩/解压速度比较快；支持 Split	压缩率一般；想支持切片需要额外创建索引
Snappy	压缩和解压缩速度快	不支持 Split；压缩率一般

压缩参数配置

Hadoop 引入了编码/解码器

压缩格式	对应的编码/解码器
DEFLATE	org.apache.hadoop.io.compress.DefaultCodec
gzip	org.apache.hadoop.io.compress.GzipCodec
bzip2	org.apache.hadoop.io.compress.BZip2Codec
LZO	com.hadoop.compression.lzo.LzopCodec
Snappy	org.apache.hadoop.io.compress.SnappyCodec

配置参数

参数	默认值	阶段	建议
io.compression.codecs （在 core-site.xml 中配置）	无，这个需要在命令行输入hadoop checknative 查看	输入压缩	Hadoop 使用文件扩展名判断是否支持某种编解码器
mapreduce.map.output.compress （在 mapred-site.xml 中配置）	false	mapper 输出	这个参数设为 true 启用压缩
mapreduce.output.fileoutputformat.compress （在mapred-site.xml 中配置）	false	reducer 输出	这个参数设为 true 启用压缩
mapreduce.output.fileoutputformat.compress.codec （在mapred-site.xml 中配置）	org.apache.hadoop.io. compress.DefaultCodec	reducer 输出	使用标准工具或者编解码器，如 gzip 和bzip2

程序实现压缩

Map 输出端采用压缩

在Driver类中配置，开启压缩和设置压缩方式（以BZip压缩为例）

Configuration configuration = new Configuration();

// 开启 map 端输出压缩
configuration.setBoolean("mapreduce.map.output.compress", true);
// 设置 map 端输出压缩方式
configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

Reduce 输出端采用压缩

在Driver类中配置，开启压缩和设置压缩方式（以BZip压缩为例）

// 设置 reduce 端输出压缩开启
FileOutputFormat.setCompressOutput(job, true);
// 设置压缩的方式
FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

MapReduce

详细使用