9498HadoopMapReduce工作流程(重要)

Hadoop-MapReduce工作流程&＃xff08;重要&＃xff09;&＃xff1a;

上面的流程是整个 MapReduce 最全工作流程&＃xff0c;但是 Shuffle 过程只是从第 7 步开始到第

16 步结束&＃xff0c;具体 Shuffle 过程详解&＃xff0c;如下&＃xff1a;

&＃xff08;1&＃xff09;MapTask 收集我们的 map()方法输出的 kv 对&＃xff0c;放到内存缓冲区中

&＃xff08;2&＃xff09;从内存缓冲区不断溢出本地磁盘文件&＃xff0c;可能会溢出多个文件

&＃xff08;3&＃xff09;多个溢出文件会被合并成大的溢出文件

&＃xff08;4&＃xff09;在溢出过程及合并的过程中&＃xff0c;都要调用 Partitioner 进行分区和针对 key 进行排序

&＃xff08;5&＃xff09;ReduceTask 根据自己的分区号&＃xff0c;去各个 MapTask 机器上取相应的结果分区数据

&＃xff08;6&＃xff09;ReduceTask 会抓取到同一个分区的来自不同 MapTask 的结果文件&＃xff0c;ReduceTask 会

将这些文件再进行合并&＃xff08;归并排序&＃xff09;

&＃xff08;7&＃xff09;合并成大文件后&＃xff0c;Shuffle 的过程也就结束了&＃xff0c;后面进入 ReduceTask 的逻辑运算过

程&＃xff08;从文件中取出一个一个的键值对 Group&＃xff0c;调用用户自定义的 reduce()方法&＃xff09;

注&＃xff1a;

&＃xff08;1&＃xff09;Shuffle 中的缓冲区大小会影响到 MapReduce 程序的执行效率&＃xff0c;原则上说&＃xff0c;缓冲区

越大&＃xff0c;磁盘 io 的次数越少&＃xff0c;执行速度就越快。

&＃xff08;2&＃xff09;缓冲区的大小可以通过参数调整&＃xff0c;参数&＃xff1a;mapreduce.task.io.sort.mb 默认 100M。

Shuffle 机制

Map 方法之后&＃xff0c;Reduce 方法之前的数据处理过程称之为 Shuffle。

Partition 分区

1、问题引出

要求将统计结果按照条件输出到不同文件中&＃xff08;分区&＃xff09;。比如&＃xff1a;将统计结果按照手机

归属地不同省份输出到不同文件中&＃xff08;分区&＃xff09;

2、默认Partitioner分区

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


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

key存储到哪个分区。

3、自定义Partitioner步骤

&＃xff08;1&＃xff09;自定义类继承Partitioner&＃xff0c;重写getPartition()方法

public class CustomPartitioner extends Partitioner {
&＃64;Override
public int getPartition(Text key, FlowBean value, int numPartitions) {
// 控制分区代码逻辑
… …
return partition;
}
}


&＃xff08;2&＃xff09;在Job驱动中&＃xff0c;设置自定义Partitioner

job.setPartitionerClass(CustomPartitioner.class);

&＃xff08;3&＃xff09;自定义Partition后&＃xff0c;要根据自定义Partitioner的逻辑设置相应数量的ReduceTask

job.setNumReduceTasks(5);

4、分区总结

&＃xff08;1&＃xff09;如果ReduceTask的数量> getPartition的结果数&＃xff0c;则会多产生几个空的输出文件part-r-000xx&＃xff1b;

&＃xff08;2&＃xff09;如果1

&＃xff08;3&＃xff09;如 果ReduceTask的数量&＃61;1&＃xff0c;则不管MapTask端输出多少个分区文件&＃xff0c;最终结果都交给这一个

ReduceTask&＃xff0c;最终也就只会产生一个结果文件 part-r-00000&＃xff1b;

&＃xff08;4&＃xff09;分区号必须从零开始&＃xff0c;逐一累加。

Partition 分区案例实操 &＃xff08;尚硅谷&＃xff09;

1&＃xff09;需求

将统计结果按照手机归属地不同省份输出到不同文件中&＃xff08;分区&＃xff09;

&＃xff08;1&＃xff09;输入数据

&＃xff08;2&＃xff09;期望输出数据

手机号 136、137、138、139 开头都分别放到一个独立的 4 个文件中&＃xff0c;其他开头的放到

一个文件中。

2&＃xff09;需求分析

package com.zh.mapreduce.paritioner;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class ProvincePartitioner extends Partitioner {
&＃64;Override
public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
//获取手机号前三位 prePhone
String phone &＃61; text.toString();
String prePhone &＃61; phone.substring(0, 3);
//定义一个分区号变量 partition,根据 prePhone 设置分区号
int partition;
if("136".equals(prePhone)){
partition &＃61; 0;
}else if("137".equals(prePhone)){
partition &＃61; 1;
}else if("138".equals(prePhone)){
partition &＃61; 2;
}else if("139".equals(prePhone)){
partition &＃61; 3;
}else {
partition &＃61; 4;
}
//最后返回分区号 partition
return partition;
}
}


驱动修改

//8 指定自定义分区器
job.setPartitionerClass(ProvincePartitioner.class);
//9 同时指定相应数量的 ReduceTask
job.setNumReduceTasks(5);


学习路径&＃xff1a;https://space.bilibili.com/302417610/&＃xff0c;如有侵权&＃xff0c;请联系q进行删除&＃xff1a;3623472230