shuffle是MapReduce的核心，map和reduce的中间过程。
Map负责过滤分发，reduce归并整理，从map输出到reduce输入就是shuffle过程。
实现的功能
分区
决定当前key交给哪个reduce处理
默认：按照key的hash值对reduce的个数取余进行分区
分组
将相同key的value合并
排序
按照key对每一个keyvalue进行排序，字典排序
过程
map端shuffle
spill阶段：溢写
每一个map task处理的结果会进入环形缓冲区（内存100M）
分区
对每一条key进行分区（标上交给哪个reduce）
hadoop 1 reduce0 hive 1 reduce0 spark 1 reduce1 hadoop 1 reduce0 hbase 1 reduce1
排序
按照key排序，将相同分区的数据进行分区内排序
hadoop 1 reduce0 hadoop 1 reduce0 hive 1 reduce0 hbase 1 reduce1 spark 1 reduce1
溢写
当整个缓冲区达到阈值80%，开始进行溢写
将当前分区排序后的数据写入磁盘变成一个文件file1最终生成多个spill小文件
可以在mapred-site.xml中设置内存的大小和溢写的阈值
在mapred-site.xml中设置内存的大小 ? <property> ? <name>mapreduce.task.io.sort.mb</name> ? <value>100</value> ? </property> ? 在mapred-site.xml中设置内存溢写的阈值 ? <property> ? <name>mapreduce.task.io.sort.spill.percent</name> ? <value>0.8</value> ? </property>
merge:合并
将spill生成的多个小文件进行合并
排序：将相同分区的数据进行分区内排序，实现comparator比较器进行比较。最终形成一个文件。
file1 hadoop 1 reduce0 hadoop 1 reduce0 hive 1 reduce0 hbase 1 reduce1 spark 1 reduce1 ? file2 hadoop 1 reduce0 hadoop 1 reduce0 hive 1 reduce0 hbase 1 reduce1 spark 1 reduce1 ? end_file: hadoop 1 reduce0 hadoop 1 reduce0 hadoop 1 reduce0 hadoop 1 reduce0 hive 1 reduce0 hive 1 reduce0 hbase 1 reduce1 hbase 1 reduce1 spark 1 reduce1 spark 1 reduce1
map task 结束，通知app master，app master通知reduce拉取数据
reduce端shuffle
map task1
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hive 1 reduce0
hive 1 reduce0
hbase 1 reduce1
hbase 1 reduce1
spark 1 reduce1
spark 1 reduce1
map task2
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hive 1 reduce0
hive 1 reduce0
hbase 1 reduce1
hbase 1 reduce1
spark 1 reduce1
spark 1 reduce1
reduce启动多个线程通过http到每台机器上拉取属于自己分区的数据
reduce0：
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hadoop 1 reduce0
hive 1 reduce0
hive 1 reduce0
hive 1 reduce0
hive 1 reduce0
merge:合并，将每个map task的结果中属于自己的分区数据进行合并
排序：对整体属于我分区的数据进行排序
分组：对相同key的value进行合并，使用comparable完成比较。
hadoop，list<1,1,1,1,1,1,1,1> hive,list<1,1,1,1>
优化
combine
在map阶段提前进行一次合并。一般等同于提前执行reduce
job.setCombinerClass(WCReduce.class);
compress
压缩中间结果集，减少磁盘IO以及网络IO
压缩配置方式
1.default：所有hadoop中默认的配置项 2.site：用于自定义配置文件，如果修改以后必须重启生效 3.conf对象配置每个程序的自定义配置 4.运行时通过参数实现用户自定义配置 bin/yarn jar xx.jar -Dmapreduce.map.output.compress=true -Dmapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.Lz4Codec main_class input_path ouput_path
查看本地库支持哪些压缩
bin/hadoop checknative
通过conf配置对象配置压缩
public static void main(String[] args) {
Configuration configuration = new Configuration();
//配置map中间结果集压缩
configuration.set("mapreduce.map.output.compress","true");
configuration.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.Lz4Codec");
//配置reduce结果集压缩
configuration.set("mapreduce.output.fileoutputformat.compress","true");
configuration.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.Lz4Codec");
try {
int status = ToolRunner.run(configuration, new MRDriver(), args);
System.exit(status);
} catch (Exception e) {
e.printStackTrace();
}
}
原文地址：https://www.cnblogs.com/whcwkw1314/p/8970985.html
时间： 2024-10-10 23:12:34