前言

reduceByKey会将上一个RDD中的每一个key对应的所有value聚合成一个value，然后生成一个新的RDD，元素类型是<key,value>对的形式，这样每一个key对应一个聚合起来的value。聚合之前，每一个key对应的value不一定都是在一个partition中，也不太可能在同一个节点上，因为RDD是分布式的弹性的数据集，RDD的partition极有可能分布在各个节点上。在1.6版本中有两种管理机制：hashShuffleManager、sortShuffleManager。在2.3版本中砍掉了hashShuffleManager，只留下了sortShuffleManager。

hashShuffleManager

1.普通机制

上图中1个cpu core，2个task，必定串行运行。一个Task的数据写往内存Buffer缓冲区，每个Buffer的大小是32k，有几个reduceTask就写往几个buffer。每个buffer对应一个磁盘小文件。然后reduce端就从各个节点拉取属于自己的磁盘小文件。小文件个数=map个数*reduce个数

缺陷

产生的磁盘小文件过多，会导致：
1.shuffle write过程会产生很多写磁盘小文件的对象
2.shuffle read过程会产生很多读取磁盘小文件的对象
3.在JVM堆内存中对象过多就会造成频繁GC，如果GC都解决不了运行所需内存问题，就会OOM
4.数据传输过程中会有频繁的网络通信，频繁的网络通信出现通信故障的可能性大大增加，一旦网络通信出现了故障会导致shuffle file cannot find 由于这个错误导致的task失败，TaskScheduler不负责重试，由DAGScheduler负责重试Stage。

2.合并机制-优化

task共用一个buffer缓冲区，小文件个数=core个数*reduce个数

sortShuffleManager

能够替代hashShuffleManager，主要就是为了解决磁盘I/O问题。

1.普通机制

小文件个数：2map （2包含索引文件和磁盘文件）
1.map task 的计算结果会写入到一个内存数据结构里面，内存数据结构默认是5M
2.在shuffle的时候会有一个定时器，不定期的去估算这个内存结构的大小，当内存结构中的数据超过5M时，比如现在内存结构中的数据为5.01M，那么他会申请5.012-5=5.02M内存给内存数据结构。
3.如果申请成功不会进行溢写（继续写往内存），如果申请不成功，这时候会发生溢写磁盘。
4.在溢写之前内存结构中的数据会根据分区号进行排序（0、1、2）
5.然后开始溢写磁盘，写磁盘是以batch的形式去写，一个batch是1万条数据，
6.map task执行完成后，会将这些磁盘小文件合并成一个大的磁盘文件（0号在一块，1号在一块…），同时生成一个索引文件。
7.reduce task去map端拉取数据的时候，首先解析索引文件，根据索引文件再去拉取对应的数据。

2.byPass机制

相比于普通模式，丢弃了排序。诸如：repartition之类的算子涉及到了重新分区，所有排序在此处就毫无意义了。

总结：

① bypass运行机制的触发条件如下：
shuffle reduce task的数量小于spark.shuffle.sort.bypassMergeThreshold的参数值。这个值默认是200。
产生的磁盘小文件为：2*M（map task的个数）
②map端不能有预聚合

Spark Shuffle两种Manager相关推荐

深入理解spark两种调度模式：FIFO，FAIR模式
深入理解spark-两种调度模式FIFO,FAIR模式前面我们应知道了一个任务提交会由DAG拆分为job,stage,task,最后提交给TaskScheduler,在提交taskscheduler ...
Java Spark之创建RDD的两种方式和操作RDD
首先看看思维导图,我的spark是1.6.1版本,jdk是1.7版本 spark是什么? Spark是基于内存计算的大数据并行计算框架.Spark基于内存计算,提高了在大数据环境下数据处理的实时性, ...
Spark 基础 —— RDD（创建 RDD）的两种方式
弹性分布式数据集(Resilient Distributed Dataset),简称 RDD,RDD 是 Spark 所提供的最基本的数据抽象,代表分布在集群中多台机器上的对象集合.Spark 有两种 ...
submit按钮html,html的两种提交按钮submit和button
转自:http://baiying.blog.51cto.com/1068039/1319784 html按钮有两种: 和登录这两种在页面上看起来好像一样没什么区别,但是这两种是有区别的. 1,如果 ...
RDD转换为DataFrame的两种方式详解
Spark支持两种方法将存在的RDD转换为DataFrame(SchemaRDD),后面附完整样例代码元数据:person.txt 1 zhangsan 20 2 lisi 29 3 wangwu ...
Spark内核（上）——附：两种Yarn模式源码解析
文章目录一.Spark内核概述 1.1 Spark核心组件回顾 1.1.1 Driver 1.1.2 Executor 1.2 Spark通用运行流程概述二.Spark通信架构概述 2.1 Spa ...
string 中的offset_Kafka+Spark Streaming管理offset的两种方法
Kafka配合Spark Streaming是大数据领域常见的黄金搭档之一,主要是用于数据实时入库或分析. 为了应对可能出现的引起Streaming程序崩溃的异常情况,我们一般都需要手动管理好Kaf ...
两种类型的Spark RDD task解析以及iterator解析 -- （视频笔记）
spark 每次提交一个作业,都会划分成不同的stage,除了最后一个stage(shuffle map task),前边的stage属于一种类型(result task). spark中的job,内 ...
Spark资源调度机制源码分析--基于spreadOutApps及非spreadOutApps两种资源调度算法
Spark资源调度机制源码分析--基于spreadOutApps及非spreadOutApps两种资源调度算法 1.spreadOutApp尽量平均分配到每个executor上: 2.非spreadO ...

Spark Shuffle两种Manager

文章目录

前言

hashShuffleManager

1.普通机制

缺陷

2.合并机制-优化

sortShuffleManager

1.普通机制

2.byPass机制

总结：

Spark Shuffle两种Manager相关推荐

最新文章

热门文章