大数据技术之Hadoop（第二章 mapreduceshuffer过程）

雨点打透心脏的1/2处 2022-12-24 13:00 233阅读 0赞

### mapreduceshuffer过程 ###

*  3.4 Shuffle机制
 *   *   *  3.4.1 Shuffle机制
         *  3.4.2 Partition分区
         *  3.4.3 Partition分区案例实操
         *  3.4.4 WritableComparable排序
         *  3.4.5 WritableComparable排序案例实操
 *  案例一
 *  案例二
 *   *   *  3.4.6 Combiner合并
         *  3.4.7 Combiner合并案例实操
         *  3.4.8 GroupingComparator分组（辅助排序）
         *  3.4.9 GroupingComparator分组案例实操
     *  3.5 ReduceTask工作机制
     *  3.6 OutputFormat数据输出
     *   *  3.6.1 OutputFormat接口实现类
         *  3.6.2 自定义OutputFormat
         *  3.6.3 自定义OutputFormat案例实操
     *  3.7 Join多种应用
     *   *  3.7.1 Reduce join
         *  3.7.3 Map join

# 3.4 Shuffle机制 #

**shuffle: 洗牌，清洗。**  
// 源文件：  
public static void main hello  
hello static private asdfasdf ba c abc

public 1  
static 2  
hello 2  
**shuffle过程是MapReduce整个工作流程的核心环节**  
**map : mapShuffle  
reduce: reduceShuffle**  
**1.在Map端的shuffle过程：Map输出的结果首先被写入缓存，当缓存满时，启动溢写(环形缓冲区（100M。80M）)操作，把缓存写入磁盘文件，然后清空缓存，**

2.当启动溢写操作时，首先需要**把缓存中的数据进行分区**(partition)，**然后对每个分区进行排序**(sort)，**合并**(combine)，  
3.**之后再写入磁盘文件**，每次溢写操作会生成一个新的磁盘文件，随着Map任务的执行，会有越来越多的磁盘文件，然后通知相应的Reduce来领取属于自己的数据  
.在Reduce端的shuffle过程，Reduce任务领回属于自己的数据后，然后对数据进行归并（Merge）然后交给Reduce处理  
**2.Map端的shuffle**  
\*\*Map端的shuffle包括四个步骤

1.  **输入数据和执行Map任务**：Map输入的数据一般保存在HDFS文件块中，这些文件块的格式可以是任意的
2.  **写入缓存**：每个Map任务都会被分配一个缓存，Map输出的结果不是立即写入磁盘，而是首先写入缓存，积累一定缓存后，再一次性批量写入磁盘，这样可以大大减少对磁盘IO的影响，因为每次磁盘寻址开销很大，在写入缓存之前，键值对都会被序列化成字节数组
3.  **溢写（分区，排序，合并）**：提供给MapReduce的缓存容量是有限的，默认大小100MB，随着Map任务的执行，缓存会越来越多，这时就必须启动溢写操作（spill），溢写过程通常是由另一个单独的后台线程来完成，但是为了保证Map任务写入缓存不受溢写影响，一般设置一个溢写比例0.8，在溢写到磁盘之前，缓存中的数据首先会被分区（partition）（），MapReduce通过Partitioner接口对这些键值对进行分区，采用默认的分区方式是Hash函数对key进行哈希后再用Reduce任务的数量进行取模，这样就可以把Map输出结果均匀的分配给所有的Reduce去并行处理，当然也可以自定义分区机制，对于分区内的每个键值对，后台线程会根据key对他们进行排序（sort），排序是MapReduce的默认操作，排序结束后还包含一个可选的合并(combine)操作，如果用户事先没有定义combiner函数，就不用进行合并，如果定义了，这个操作会减少需要溢写到磁盘的数据量，合并就是把相同key的value加起来，减少键值对的数量，功能与Reduce相似，但是并不是所有的场合都可以使用combiner，因为combiner的输出是Reduce的输入，combiner绝不能改变Reduce的计算结果，一般在累加，最大值的这种场景使用，每次溢写操作都会生成一个新的溢写文件，所有写入这个文件中的数据都是经过分区和排序的。
4.  \*\*文件归并：\*\*在Map任务结束之前，系统会对所有的溢写文件进行归并（Merge），生成一个大的溢写文件，归并就是对具有相同key的键值对合并，形成一个新的键值对，另外，在文件归并时，如果磁盘中生成溢写文件的数量超过min.num.spills.for.combine的值时（默认是3，用户可以修改），那么就可以再次运行combiner对数据进行合并操作，从而减小写入磁盘的数据量  
    \*\*\*\*经过上述四个步骤后，Map端的shuffle过程全部完成，最终生成的大文件会被存放在本地磁盘中，这个大文件中的数据是被分区的，不同的分区分发到不同的Reduce任务进行并行处理，JobTracker会一直监测Map任务的执行，当监测到一个Map任务完成后，就会立刻通知相关的Reduce任务来领取数据，开始Reduce端的shuffle过程。  
    **3.Reduce端的shuffle**

\*\*Reduce端的shuffle包括三个步骤：

**1. 领取数据**：Map端的shuffle过程结束后，结果会保存在本地磁盘中，Reduce任务只需要把这些数据领取（Fetch）回来存放到自己机器所在的磁盘上，因此，每个Reduce任务在执行之前，大部分时间都在领数据，每一个Reduce会不断通过RPC向JobTracker询问Map任务是否已经完成  
\*\*2. 归并数据：\*\*领回数据后，会首先被存在Reduce任务所在机器的缓存中，如果缓存满了，就会像Map一样发生溢写操作，由于在shuffle阶段，真正的Reduce还没有执行，这时可以把内存的大部分空间分配给shuffle过程作为缓存，缓存中的数据是来自不同机器的，一般会存在很多可以合并(combine)的键值对，当溢写程序启动时，具有相同key的键值对会被归并，如果用户定义了combiner，则归并后还可以执行combiner，减少写入磁盘的数据量，每个溢写过程结束后，都会在磁盘生成一个溢写文件，当溢写文件过多时也会像Map一样被归并成一个大文件，归并是也会进行排序，当数据很少时，写入缓存就行，不需要溢写到磁盘，而是直接在内存中执行归并操作，直接输出给Reduce任务，需要说明，把这些溢写文件归并成一个大文件需要多轮归并操作，每轮归并操作的文件数量有参数io.sort.factor来决定（默认是10，可以修改）  
\*\*3. 把数据输入Reduce任务：\*\*磁盘中经过多轮归并后得到若干个大文件，不会继续归并成一个新的大文件，而是直接输出给Reduce任务，这样可以减少磁盘的读写开销，这样整个shuffle过就结束了，接下来Reduce任务会执行Reduce函数中定义的各种映射，输出最终结果，并保存到分布式文件系统中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 3]

**总结：**\[重点哦！！！！\]

> map task 从split中读取数据，进行处理后，输出key/value，对键值对进行Partitioner后，存入到缓存中，缓存默认大小是100M，当缓存内容达到80M时，启动溢写操作，把缓存区数据写入一个溢写文件，在写入文件之前，会对键值对进行分区排序和合并（如果设置的话）,当该map task处理完所有数据后，需要对该map生成的所有溢写文件进行merger操作，生成一个文件作为该maptask的成果，reduce task接受到通知后，就回拉取各个map task的成果数据，放到缓存区中，当缓存区内容达到阀值时，同样执行溢写操作，生成溢写文件，当把所有的map task的成果数据读取完毕后，会把生成的所有溢写文件进行merge操作，生成一个文件作为reduce task的输出数据。

### 3.4.1 Shuffle机制 ###

Mapreduce确保每个reducer的输入都是按key排序的。系统执行排序的过程（即将mapper输出作为输入传给reducer）称为shuffle，如图4-14所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 4]

### 3.4.2 Partition分区 ###

问题引出：要求将统计结果按照条件输出到不同文件中（分区）。比如：将统计结果按照手机归属地不同省份输出到不同文件中（分区）  
1．默认partition分区

> public class HashPartitioner<K, V> extends Partitioner<K, V> \{  
> public int getPartition(K key, V value, int numReduceTasks) \{  
> return (key.hashCode() & Integer.MAX\_VALUE) % numReduceTasks;  
> \}  
> \}

默认分区是根据key的hashCode对reduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区。  
2．自定义Partitioner步骤  
（1）自定义类继承Partitioner，重写getPartition()方法

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
        
    
    	@Override
    	public int getPartition(Text key, FlowBean value, int numPartitions) {
        
    
    // 1 获取电话号码的前三位
    		String preNum = key.toString().substring(0, 3);
    		
    		int partition = 4;
    		
    		// 2 判断是哪个省
    		if ("136".equals(preNum)) {
        
    			partition = 0;
    		}else if ("137".equals(preNum)) {
        
    			partition = 1;
    		}else if ("138".equals(preNum)) {
        
    			partition = 2;
    		}else if ("139".equals(preNum)) {
        
    			partition = 3;
    		}
    		return partition;
    	}
    }

（2）在job驱动中，设置自定义partitioner：

job.setPartitionerClass(CustomPartitioner.class);

（3）自定义partition后，要根据自定义partitioner的逻辑设置相应数量的reduce task

job.setNumReduceTasks(5);

3．注意

> 如果reduceTask的数量> getPartition的结果数，则会多产生几个空的输出文件part-r-000xx；  
> 如果1<reduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会Exception；  
> 如果reduceTask的数量=1，则不管mapTask端输出多少个分区文件，最终结果都交给这一个reduceTask，最终也就只会产生一个结果文件 part-r-00000；

例如：假设自定义分区数为5，则  
（1）job.setNumReduceTasks(1);会正常运行，只不过会产生一个输出文件  
（2）job.setNumReduceTasks(2);会报错  
（3）job.setNumReduceTasks(6);大于5，程序会正常运行，会产生空文件

### 3.4.3 Partition分区案例实操 ###

1．需求  
将统计结果按照手机归属地不同省份输出到不同文件中（分区）  
2．数据准备  
![在这里插入图片描述][20201128153509319.png]  
3．分析  
（1）Mapreduce中会将map输出的kv对，按照相同key分组，然后分发给不同的reducetask。默认的分发规则为：根据key的hashcode%reducetask数来分发  
（2）如果要按照我们自己的需求进行分组，则需要改写数据分发（分组）组件Partitioner  
自定义一个CustomPartitioner继承抽象类：Partitioner  
（3）在job驱动中，设置自定义partitioner： job.setPartitionerClass(CustomPartitioner.class)  
4．在案例2.4的基础上，增加一个分区类

package com.bk.mapreduce.flowsum;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
        
    
    	@Override
    	public int getPartition(Text key, FlowBean value, int numPartitions) {
        
    		// 1 获取电话号码的前三位
    		String preNum = key.toString().substring(0, 3);
    		
    		int partition = 4;
    		
    		// 2 判断是哪个省
    		if ("136".equals(preNum)) {
        
    			partition = 0;
    		}else if ("137".equals(preNum)) {
        
    			partition = 1;
    		}else if ("138".equals(preNum)) {
        
    			partition = 2;
    		}else if ("139".equals(preNum)) {
        
    			partition = 3;
    		}
    
    		return partition;
    	}
    }

5．在驱动函数中增加自定义数据分区设置和reduce task设置

package com.bk.mapreduce.flowsum;
    import java.io.IOException;
    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;
    
    public class FlowsumDriver {
        
    
    	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        
    		
    		// 1 获取配置信息，或者job对象实例
    		Configuration configuration = new Configuration();
    		Job job = Job.getInstance(configuration);
    
    		// 6 指定本程序的jar包所在的本地路径
    		job.setJarByClass(FlowsumDriver.class);
    
    		// 2 指定本业务job要使用的mapper/Reducer业务类
    		job.setMapperClass(FlowCountMapper.class);
    		job.setReducerClass(FlowCountReducer.class);
    
    		// 3 指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(FlowBean.class);
    
    		// 4 指定最终输出的数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(FlowBean.class);
    
    		// 8 指定自定义数据分区
    		job.setPartitionerClass(ProvincePartitioner.class);
    		// 9 同时指定相应数量的reduce task
    		job.setNumReduceTasks(5);
    		
    		// 5 指定job的输入原始文件所在目录
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

### 3.4.4 WritableComparable排序 ###

排序是MapReduce框架中最重要的操作之一。Map Task和Reduce Task均会对数据（按照key）进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。  
对于Map Task，它会将处理的结果暂时放到一个缓冲区中，当缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次排序，并将这些有序数据写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行一次合并，以将这些文件合并成一个大的有序文件。  
对于Reduce Task，它从每个Map Task上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则放到磁盘上，否则放到内存中。如果磁盘上文件数目达到一定阈值，则进行一次合并以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据写到磁盘上。当所有数据拷贝完毕后，Reduce Task统一对内存和磁盘上的所有数据进行一次合并。每个阶段的默认排序  
1．排序的分类  
（1）部分排序：  
MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部排序。  
（2）全排序：  
如何用Hadoop产生一个全局排序的文件？最简单的方法是使用一个分区。但该方法在处理大型文件时效率极低，因为一台机器必须处理所有输出文件，从而完全丧失了MapReduce所提供的并行架构。  
替代方案：首先创建一系列排好序的文件；其次，串联这些文件；最后，生成一个全局排序的文件。主要思路是使用一个分区来描述输出的全局排序。例如：可以为上述文件创建3个分区，在第一分区中，记录的单词首字母a-g，第二分区记录单词首字母h-n, 第三分区记录单词首字母o-z。  
（3）辅助排序：（GroupingComparator分组）  
Mapreduce框架在记录到达reducer之前按键对记录排序，但键所对应的值并没有被排序。甚至在不同的执行轮次中，这些值的排序也不固定，因为它们来自不同的map任务且这些map任务在不同轮次中完成时间各不相同。一般来说，大多数MapReduce程序会避免让reduce函数依赖于值的排序。但是，有时也需要通过特定的方法对键进行排序和分组等以实现对值的排序。  
（4）二次排序：  
在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。  
2．自定义排序WritableComparable  
（1）原理分析  
bean对象实现WritableComparable接口重写compareTo方法，就可以实现排序

@Override
    public int compareTo(FlowBean o) {
        
    	// 倒序排列，从大到小
    	return this.sumFlow > o.getSumFlow() ? -1 : 1;
    }

### 3.4.5 WritableComparable排序案例实操 ###

# 案例一 #

1．需求  
根据案例2.4产生的结果再次对总流量进行排序。  
2．数据准备  
![在这里插入图片描述][20201128153750313.png]  
3．分析  
（1）把程序分两步走，第一步正常统计总流量，第二步再把结果进行排序  
（2）context.write(总流量，手机号)  
（3）FlowBean实现WritableComparable接口重写compareTo方法

@Override
    public int compareTo(FlowBean o) {
        
    	// 倒序排列，从大到小
    	return this.sumFlow > o.getSumFlow() ? -1 : 1;
    }

4．代码实现  
（1）FlowBean对象在在需求1基础上增加了比较功能

package com.bk.mapreduce.sort;
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    import org.apache.hadoop.io.WritableComparable;
    
    public class FlowBean implements WritableComparable<FlowBean> {
        
    
    	private long upFlow;
    	private long downFlow;
    	private long sumFlow;
    
    	// 反序列化时，需要反射调用空参构造函数，所以必须有
    	public FlowBean() {
        
    		super();
    	}
    
    	public FlowBean(long upFlow, long downFlow) {
        
    		super();
    		this.upFlow = upFlow;
    		this.downFlow = downFlow;
    		this.sumFlow = upFlow + downFlow;
    	}
    
    	public void set(long upFlow, long downFlow) {
        
    		this.upFlow = upFlow;
    		this.downFlow = downFlow;
    		this.sumFlow = upFlow + downFlow;
    	}
    
    	public long getSumFlow() {
        
    		return sumFlow;
    	}
    
    	public void setSumFlow(long sumFlow) {
        
    		this.sumFlow = sumFlow;
    	}
    
    	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;
    	}
    
    	/**
    	 * 序列化方法
    	 * @param out
    	 * @throws IOException
    	 */
    	@Override
    	public void write(DataOutput out) throws IOException {
        
    		out.writeLong(upFlow);
    		out.writeLong(downFlow);
    		out.writeLong(sumFlow);
    	}
    
    	/**
    	 * 反序列化方法 注意反序列化的顺序和序列化的顺序完全一致
    	 * @param in
    	 * @throws IOException
    	 */
    	@Override
    	public void readFields(DataInput in) throws IOException {
        
    		upFlow = in.readLong();
    		downFlow = in.readLong();
    		sumFlow = in.readLong();
    	}
    
    	@Override
    	public String toString() {
        
    		return upFlow + "\t" + downFlow + "\t" + sumFlow;
    	}
    
    	@Override
    	public int compareTo(FlowBean o) {
        
    		// 倒序排列，从大到小
    		return this.sumFlow > o.getSumFlow() ? -1 : 1;
    	}
    }

（2）编写mapper

package com.bk.mapreduce.sort;
    import java.io.IOException;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{
        
    	FlowBean bean = new FlowBean();
    	Text v = new Text();
    
    	@Override
    	protected void map(LongWritable key, Text value, Context context)
    			throws IOException, InterruptedException {
        
    
    		// 1 获取一行
    		String line = value.toString();
    		
    		// 2 截取
    		String[] fields = line.split("\t");
    		
    		// 3 封装对象
    		String phoneNbr = fields[0];
    		long upFlow = Long.parseLong(fields[1]);
    		long downFlow = Long.parseLong(fields[2]);
    		
    		bean.set(upFlow, downFlow);
    		v.set(phoneNbr);
    		
    		// 4 输出
    		context.write(bean, v);
    	}
    }

（3）编写reducer

package com.bk.mapreduce.sort;
    import java.io.IOException;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{
        
    
    	@Override
    	protected void reduce(FlowBean key, Iterable<Text> values, Context context)
    			throws IOException, InterruptedException {
        
    		
    		// 循环输出，避免总流量相同情况
    		for (Text text : values) {
        
    			context.write(text, key);
    		}
    	}
    }

（4）编写driver

package com.bk.mapreduce.sort;
    import java.io.IOException;
    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;
    
    public class FlowCountSortDriver {
        
    
    	public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {
        
    		
    		// 1 获取配置信息，或者job对象实例
    		Configuration configuration = new Configuration();
    		Job job = Job.getInstance(configuration);
    
    		// 6 指定本程序的jar包所在的本地路径
    		job.setJarByClass(FlowCountSortDriver.class);
    
    		// 2 指定本业务job要使用的mapper/Reducer业务类
    		job.setMapperClass(FlowCountSortMapper.class);
    		job.setReducerClass(FlowCountSortReducer.class);
    
    		// 3 指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(FlowBean.class);
    		job.setMapOutputValueClass(Text.class);
    
    		// 4 指定最终输出的数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(FlowBean.class);
    
    		// 5 指定job的输入原始文件所在目录
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		
    		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

# 案例二 #

1．需求  
要求每个省份手机号输出的文件中按照总流量内部排序。  
2．分析  
基于前一个需求，增加自定义分区类即可。  
3．案例实操  
（1）增加自定义分区类

package com.bk.mapreduce.sort;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class ProvincePartitioner extends Partitioner<FlowBean, Text> {
        
    
    	@Override
    	public int getPartition(FlowBean key, Text value, int numPartitions) {
        
    		
    		// 1 获取手机号码前三位
    		String preNum = value.toString().substring(0, 3);
    		
    		int partition = 4;
    		
    		// 2 根据手机号归属地设置分区
    		if ("136".equals(preNum)) {
        
    			partition = 0;
    		}else if ("137".equals(preNum)) {
        
    			partition = 1;
    		}else if ("138".equals(preNum)) {
        
    			partition = 2;
    		}else if ("139".equals(preNum)) {
        
    			partition = 3;
    		}
    
    		return partition;
    	}
    }

（2）在驱动类中添加分区类

// 加载自定义分区类
    job.setPartitionerClass(FlowSortPartitioner.class);
    // 设置Reducetask个数
    job.setNumReduceTasks(5);

### 3.4.6 Combiner合并 ###

1）combiner是MR程序中Mapper和Reducer之外的一种组件。  
2）combiner组件的父类就是Reducer。  
3）combiner和reducer的区别在于运行的位置：  
Combiner是在每一个maptask所在的节点运行;  
Reducer是接收全局所有Mapper的输出结果；  
4）combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量。  
5）combiner能够应用的前提是不能影响最终的业务逻辑，而且，combiner的输出kv应该跟reducer的输入kv类型要对应起来。  
Mapper  
3 5 7 ->(3+5+7)/3=5  
2 6 ->(2+6)/2=4  
Reducer  
(3+5+7+2+6)/5=23/5 不等于 (5+4)/2=9/2  
6）自定义Combiner实现步骤：  
（1）自定义一个combiner继承Reducer，重写reduce方法

> public class WordcountCombiner extends Reducer<Text, IntWritable, Text,  
> IntWritable>\{  
> @Override  
> protected void reduce(Text key, Iterable values,  
> Context context) throws IOException, InterruptedException \{  
> // 1 汇总操作  
> int count = 0;  
> for(IntWritable v :values)\{  
> count += v.get();  
> \}  
> // 2 写出  
> context.write(key, new IntWritable(count));  
> \}  
> \}

（2）在job驱动类中设置：

> job.setCombinerClass(WordcountCombiner.class);

### 3.4.7 Combiner合并案例实操 ###

1．需求  
统计过程中对每一个maptask的输出进行局部汇总，以减小网络传输量即采用Combiner功能，如图4-15所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][202011281543240.png]  
方案一  
1）增加一个WordcountCombiner类继承Reducer

package com.bk.mr.combiner;
    import java.io.IOException;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{
        
    
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values,
    			Context context) throws IOException, InterruptedException {
        
            // 1 汇总
    		int count = 0;
    		for(IntWritable v :values){
        
    			count += v.get();
    		}
    		// 2 写出
    		context.write(key, new IntWritable(count));
    	}
    }

2）在WordcountDriver驱动类中指定combiner

> // 指定需要使用combiner，以及用哪个类作为combiner的逻辑  
> job.setCombinerClass(WordcountCombiner.class);

方案二  
1）将WordcountReducer作为combiner在WordcountDriver驱动类中指定

> // 指定需要使用combiner，以及用哪个类作为combiner的逻辑  
> job.setCombinerClass(WordcountReducer.class);

运行程序，如图4-16，4-17所示  
![在这里插入图片描述][20201128154421817.png]

### 3.4.8 GroupingComparator分组（辅助排序） ###

对reduce阶段的数据根据某一个或几个字段进行分组。

### 3.4.9 GroupingComparator分组案例实操 ###

1．需求  
有如下订单数据\](https://img-blog.csdnimg.cn/20201128154428730.png)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 6]  
现在需要求出每一个订单中最贵的商品。  
![在这里插入图片描述][20201128154522203.png]  
![在这里插入图片描述][20201128154534705.png]  
3．分析  
（1）利用“订单id和成交金额”作为key，可以将map阶段读取到的所有订单数据按照id分区，按照金额排序，发送到reduce。  
（2）在reduce端利用groupingcomparator将订单id相同的kv聚合成组，然后取第一个即是最大值，如图4-18所示  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 7]  
4．代码实现  
（1）定义订单信息OrderBean

package com.bk.mapreduce.order;
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    import org.apache.hadoop.io.WritableComparable;
    
    public class OrderBean implements WritableComparable<OrderBean> {
        
    
    	private int order_id; // 订单id号
    	private double price; // 价格
    
    	public OrderBean() {
        
    		super();
    	}
    
    	public OrderBean(int order_id, double price) {
        
    		super();
    		this.order_id = order_id;
    		this.price = price;
    	}
    
    	@Override
    	public void write(DataOutput out) throws IOException {
        
    		out.writeInt(order_id);
    		out.writeDouble(price);
    	}
    
    	@Override
    	public void readFields(DataInput in) throws IOException {
        
    		order_id = in.readInt();
    		price = in.readDouble();
    	}
    
    	@Override
    	public String toString() {
        
    		return order_id + "\t" + price;
    	}
    
    	public int getOrder_id() {
        
    		return order_id;
    	}
    
    	public void setOrder_id(int order_id) {
        
    		this.order_id = order_id;
    	}
    
    	public double getPrice() {
        
    		return price;
    	}
    
    	public void setPrice(double price) {
        
    		this.price = price;
    	}
    
    	// 二次排序
    	@Override
    	public int compareTo(OrderBean o) {
        
    
    		int result;
    
    		if (order_id > o.getOrder_id()) {
        
    			result = 1;
    		} else if (order_id < o.getOrder_id()) {
        
    			result = -1;
    		} else {
        
    			// 价格倒序排序
    			result = price > o.getPrice() ? -1 : 1;
    		}
    
    		return result;
    	}
    }

（2）编写OrderSortMapper

package com.bk.mapreduce.order;
    import java.io.IOException;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {
        
    	OrderBean k = new OrderBean();
    	
    	@Override
    	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        
    		
    		// 1 获取一行
    		String line = value.toString();
    		
    		// 2 截取
    		String[] fields = line.split("\t");
    		
    		// 3 封装对象
    		k.setOrder_id(Integer.parseInt(fields[0]));
    		k.setPrice(Double.parseDouble(fields[2]));
    		
    		// 4 写出
    		context.write(k, NullWritable.get());
    	}
    }

（3）编写OrderSortPartitioner

package com.bk.mapreduce.order;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class OrderPartitioner extends Partitioner<OrderBean, NullWritable> {
        
    
    	@Override
    	public int getPartition(OrderBean key, NullWritable value, int numReduceTasks) {
        
    		
    		return (key.getOrder_id() & Integer.MAX_VALUE) % numReduceTasks;
    	}
    }

（4）编写OrderSortGroupingComparator

package com.bk.mapreduce.order;
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    public class OrderGroupingComparator extends WritableComparator {
        
    
    	protected OrderGroupingComparator() {
        
    		super(OrderBean.class, true);
    	}
    
    	@SuppressWarnings("rawtypes")
    	@Override
    	public int compare(WritableComparable a, WritableComparable b) {
        
    
    		OrderBean aBean = (OrderBean) a;
    		OrderBean bBean = (OrderBean) b;
    
    		int result;
    		if (aBean.getOrder_id() > bBean.getOrder_id()) {
        
    			result = 1;
    		} else if (aBean.getOrder_id() < bBean.getOrder_id()) {
        
    			result = -1;
    		} else {
        
    			result = 0;
    		}
    
    		return result;
    	}
    }

（5）编写OrderSortReducer

package com.bk.mapreduce.order;
    import java.io.IOException;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {
        
    
    	@Override
    	protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context)
    			throws IOException, InterruptedException {
        
    		
    		context.write(key, NullWritable.get());
    	}
    }

（6）编写OrderSortDriver

package com.bk.mapreduce.order;
    import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    public class OrderDriver {
        
    
    	public static void main(String[] args) throws Exception, IOException {
        
    
    		// 1 获取配置信息
    		Configuration conf = new Configuration();
    		Job job = Job.getInstance(conf);
    
    		// 2 设置jar包加载路径
    		job.setJarByClass(OrderDriver.class);
    
    		// 3 加载map/reduce类
    		job.setMapperClass(OrderMapper.class);
    		job.setReducerClass(OrderReducer.class);
    
    		// 4 设置map输出数据key和value类型
    		job.setMapOutputKeyClass(OrderBean.class);
    		job.setMapOutputValueClass(NullWritable.class);
    
    		// 5 设置最终输出数据的key和value类型
    		job.setOutputKeyClass(OrderBean.class);
    		job.setOutputValueClass(NullWritable.class);
    
    		// 6 设置输入数据和输出数据路径
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		// 10 设置reduce端的分组
    		job.setGroupingComparatorClass(OrderGroupingComparator.class);
    
    		// 7 设置分区
    		job.setPartitionerClass(OrderPartitioner.class);
    
    		// 8 设置reduce个数
    		job.setNumReduceTasks(3);
    
    		// 9 提交
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

## 3.5 ReduceTask工作机制 ##

1．设置ReduceTask并行度（个数）  
reducetask的并行度同样影响整个job的执行并发度和执行效率，但与maptask的并发数由切片数决定不同，Reducetask数量的决定是可以直接手动设置：  
![在这里插入图片描述][20201128154911564.png]  
2．注意  
（1）reducetask=0 ，表示没有reduce阶段，输出文件个数和map个数一致。  
（2）reducetask默认值就是1，所以输出文件个数为一个。  
（3）如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜  
（4）reducetask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个reducetask。  
（5）具体多少个reducetask，需要根据集群性能而定。  
（6）如果分区数不是1，但是reducetask为1，是否执行分区过程。答案是：不执行分区过程。因为在maptask的源码中，执行分区的前提是先判断reduceNum个数是否大于1。不大于1肯定不执行。  
3．实验：测试reducetask多少合适。  
（1）实验环境：1个master节点，16个slave节点：CPU:8GHZ，内存: 2G  
（2）实验结论：  
![在这里插入图片描述][20201128154925421.png]  
4．ReduceTask工作机制  
ReduceTask工作机制，如图4-19所示  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 8]

> （1）Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。  
> （2）Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。  
> （3）Sort阶段：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。  
> （4）Reduce阶段：reduce()函数将计算结果写到HDFS上。

## 3.6 OutputFormat数据输出 ##

### 3.6.1 OutputFormat接口实现类 ###

OutputFormat是MapReduce输出的基类，所有实现MapReduce输出都实现了 OutputFormat接口。下面我们介绍几种常见的OutputFormat实现类。  
1．文本输出TextOutputFormat  
默认的输出格式是TextOutputFormat，它把每条记录写为文本行。它的键和值可以是任意类型，因为TextOutputFormat调用toString()方法把它们转换为字符串。  
2．SequenceFileOutputFormat  
SequenceFileOutputFormat将它的输出写为一个顺序文件。如果输出需要作为后续 MapReduce任务的输入，这便是一种好的输出格式，因为它的格式紧凑，很容易被压缩。  
3．自定义OutputFormat  
根据用户需求，自定义实现输出。

### 3.6.2 自定义OutputFormat ###

为了实现控制最终文件的输出路径，可以自定义OutputFormat。  
要在一个mapreduce程序中根据数据的不同输出两类结果到不同目录，这类灵活的输出需求可以通过自定义outputformat来实现。  
1．自定义OutputFormat步骤  
（1）自定义一个类继承FileOutputFormat。  
（2）改写recordwriter，具体改写输出数据的方法write()。

### 3.6.3 自定义OutputFormat案例实操 ###

1．需求  
过滤输入的log日志中是否包含bk  
（1）包含bk的网站输出到e:/bk.log  
（2）不包含bk的网站输出到e:/other.log  
2．输入数据  
![在这里插入图片描述][20201128155026927.png]  
3．具体程序  
（1）自定义一个outputformat

package com.bk.mapreduce.outputformat;
    import java.io.IOException;
    import org.apache.hadoop.io.NullWritable;
    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;
    
    public class FilterOutputFormat extends FileOutputFormat<Text, NullWritable>{
        
    
    	@Override
    	public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job)
    			throws IOException, InterruptedException {
        
    
    		// 创建一个RecordWriter
    		return new FilterRecordWriter(job);
    	}
    }

（2）具体的写数据RecordWriter

package com.bk.mapreduce.outputformat;
    import java.io.IOException;
    import org.apache.hadoop.fs.FSDataOutputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.RecordWriter;
    import org.apache.hadoop.mapreduce.TaskAttemptContext;
    
    public class FilterRecordWriter extends RecordWriter<Text, NullWritable> {
        
    	FSDataOutputStream bkOut = null;
    	FSDataOutputStream otherOut = null;
    
    	public FilterRecordWriter(TaskAttemptContext job) {
        
    		// 1 获取文件系统
    		FileSystem fs;
    
    		try {
        
    			fs = FileSystem.get(job.getConfiguration());
    
    			// 2 创建输出文件路径
    			Path bkPath = new Path("e:/bk.log");
    			Path otherPath = new Path("e:/other.log");
    
    			// 3 创建输出流
    			bkOut = fs.create(bkPath);
    			otherOut = fs.create(otherPath);
    		} catch (IOException e) {
        
    			e.printStackTrace();
    		}
    	}
    
    	@Override
    	public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        
    
    		// 判断是否包含“bk”输出到不同文件
    		if (key.toString().contains("bk")) {
        
    			bkOut.write(key.toString().getBytes());
    		} else {
        
    			otherOut.write(key.toString().getBytes());
    		}
    	}
    
    	@Override
    	public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        
    		// 关闭资源
    		if (bkOut != null) {
        
    			bkOut.close();
    		}
    		
    		if (otherOut != null) {
        
    			otherOut.close();
    		}
    	}
    }

（3）编写FilterMapper

package com.bk.mapreduce.outputformat;
    import java.io.IOException;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class FilterMapper extends Mapper<LongWritable, Text, Text, NullWritable>{
        
    	
    	Text k = new Text();
    	
    	@Override
    	protected void map(LongWritable key, Text value, Context context)
    			throws IOException, InterruptedException {
        
    		// 1 获取一行
    		String line = value.toString();
    		
    		k.set(line);
    		
    		// 3 写出
    		context.write(k, NullWritable.get());
    	}
    }

（4）编写FilterReducer

package com.bk.mapreduce.outputformat;
    import java.io.IOException;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class FilterReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
        
    
    	@Override
    	protected void reduce(Text key, Iterable<NullWritable> values, Context context)
    			throws IOException, InterruptedException {
        
    
    		String k = key.toString();
    		k = k + "\r\n";
    
    		context.write(new Text(k), NullWritable.get());
    	}
    }

（5）编写FilterDriver

package com.bk.mapreduce.outputformat;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    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;
    
    public class FilterDriver {
        
    	public static void main(String[] args) throws Exception {
        
    
    args = new String[] {
         "e:/input/inputoutputformat", "e:/output2" };
    
    		Configuration conf = new Configuration();
    
    		Job job = Job.getInstance(conf);
    
    		job.setJarByClass(FilterDriver.class);
    		job.setMapperClass(FilterMapper.class);
    		job.setReducerClass(FilterReducer.class);
    
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(NullWritable.class);
    		
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(NullWritable.class);
    
    		// 要将自定义的输出格式组件设置到job中
    		job.setOutputFormatClass(FilterOutputFormat.class);
    
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    
    		// 虽然我们自定义了outputformat，但是因为我们的outputformat继承自fileoutputformat
    		// 而fileoutputformat要输出一个_SUCCESS文件，所以，在这还得指定一个输出目录
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

## 3.7 Join多种应用 ##

### 3.7.1 Reduce join ###

1．原理  
Map端的主要工作：为来自不同表(文件)的key/value对打标签以区别不同来源的记录。然后用连接字段作为key，其余部分和新加的标志作为value，最后进行输出。  
Reduce端的主要工作：在reduce端以连接字段作为key的分组已经完成，我们只需要在每一个分组当中将那些来源于不同文件的记录(在map阶段已经打标志)分开，最后进行合并就ok了。  
2．该方法的缺点  
这种方式的缺点很明显就是会造成map和reduce端也就是shuffle阶段出现大量的数据传输，效率很低。

### 3.7.3 Map join ###

1．使用场景  
一张表十分小、一张表很大。  
2．解决方案  
在map端缓存多张表，提前处理业务逻辑，这样增加map端业务，减少reduce端数据的压力，尽可能的减少数据倾斜。  
3．具体办法：采用distributedcache(缓存)  
（1）在mapper的setup阶段，将文件读取到缓存集合中。  
（2）在驱动函数中加载缓存。  
job.addCacheFile(new URI(“file:/e:/mapjoincache/pd.txt”));// 缓存普通文件到task运行节点，如图4-21所示  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 9]

Reduce join案例实操 Map join案例实操: [https://blog.csdn.net/wzq18845280610/article/details/110284771][https_blog.csdn.net_wzq18845280610_article_details_110284771].

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70]: /images/20221120/f3781223d380431a8dd5b56a2c263df1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 1]: /images/20221120/1c16f815a3b34964918b133ffaa3539e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 2]: /images/20221120/f6001697c00d45e78ec5ae3cdd53c2e2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201128153111681.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20201128152631926.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[20201128153509319.png]: https://img-blog.csdnimg.cn/20201128153509319.png
[20201128153750313.png]: https://img-blog.csdnimg.cn/20201128153750313.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20201128154314667.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[202011281543240.png]: https://img-blog.csdnimg.cn/202011281543240.png
[20201128154421817.png]: https://img-blog.csdnimg.cn/20201128154421817.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/2020112815450593.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[20201128154522203.png]: https://img-blog.csdnimg.cn/20201128154522203.png
[20201128154534705.png]: https://img-blog.csdnimg.cn/20201128154534705.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20201128154557947.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[20201128154911564.png]: https://img-blog.csdnimg.cn/20201128154911564.png
[20201128154925421.png]: https://img-blog.csdnimg.cn/20201128154925421.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20201128154936211.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[20201128155026927.png]: https://img-blog.csdnimg.cn/20201128155026927.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20201128155445725.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_wzq18845280610_article_details_110284771]: https://blog.csdn.net/wzq18845280610/article/details/110284771