【Spark篇】---Spark中Shuffle机制,SparkShuffle和SortShuffle

一、前述

Spark中Shuffle的机制可以分为HashShuffle,SortShuffle。

SparkShuffle概念

reduceByKey会将上一个RDD中的每一个key对应的所有value聚合成一个value,然后生成一个新的RDD,元素类型是<key,value>对的形式,这样每一个key对应一个聚合起来的value。

问题:聚合之前,每一个key对应的value不一定都是在一个partition中,也不太可能在同一个节点上,因为RDD是分布式的弹性的数据集,RDD的partition极有可能分布在各个节点上。

如何聚合?

– Shuffle Write:上一个stage的每个map task就必须保证将自己处理的当前分区的数据相同的key写入一个分区文件中,可能会写入多个不同的分区文件中。

 – Shuffle Read:reduce task就会从上一个stage的所有task所在的机器上寻找属于己的那些分区文件,这样就可以保证每一个key所对应的value都会汇聚到同一个节点上去处理和聚合。

Spark中有两种Shuffle类型,HashShuffle和SortShuffle,Spark1.2之前是HashShuffle默认的分区器是HashPartitioner,Spark1.2引入SortShuffle默认的分区器是RangePartitioner。

二、具体

1、HashShuffle

  1) 普通机制

  • 普通机制示意图
  • 执行流程

       a) 每一个map task将不同结果写到不同的buffer中,每个buffer的大小为32K。buffer起到数据缓存的作用。新写的磁盘小文件会追加内容。

       b) 每个buffer文件最后对应一个磁盘小文件。

       c) reduce task来拉取对应的磁盘小文件。

  • 总结

     a) maptask的计算结果会根据分区器(默认是hashPartitioner)来决定写入到哪一个磁盘小文件中去。ReduceTask会去Map端拉取相应的磁盘小文件。

     b)产生的磁盘小文件的个数:M(map task的个数)*R(reduce task的个数)

  • 存在的问题

      产生的磁盘小文件过多,会导致以下问题:

      a) 在Shuffle Write过程中会产生很多磁盘小文件的对象。

      b) 在Shuffle Read过程中会产生很多取磁盘小文件的对象。

      c) 在JVM堆内存中对象过多会造成频繁的gc,gc还无法解决运行所需要的内存 的话,就会OOM。gc工作的时候是不提供工作的。

      d) 在数据传输过程中会有频繁的网络通信,频繁的网络通信出现通信故障的可能性大大增加一旦网络通信出现了故障会导致shuffle file cannot find 由于这个错误导致的task失败,TaskScheduler不负责重试,由DAGScheduler负责重试Stage。变相的延长执行时间

    1) 合并机制

  • 合并机制示意图

一个core 一般运行一个task,图中即便一个executor有两个task,也是串行执行的!!!!

  • 总结

产生磁盘小文件的个数:C(core的个数)*R(reduce的个数)

2、SortShuffle

1) 普通机制

  • 普通机制示意图
  • 执行流程

       a) map task 的计算结果会写入到一个内存数据结构里面,内存数据结构默认是5M

       b) 在shuffle的时候会有一个定时器,不定期的去估算这个内存结构的大小,当内存结构中的数据超过5M时,比如现在内存结构中的数据为5.01M,那么他会申请5.01*2-5=5.02M内存给内存数据结构。

       c) 如果申请成功不会进行溢写,如果申请不成功,这时候会发生溢写磁盘。

       d) 在溢写之前内存结构中的数据会进行排序分区

       e) 然后开始溢写磁盘,写磁盘是以batch的形式去写,一个batch是1万条数据,

       f) map task执行完成后,会将这些磁盘小文件合并成一个大的磁盘文件(有序),同时生成一个索引文件。

      g) reduce task去map端拉取数据的时候,首先解析索引文件,根据索引文件再去拉取对应的数据。

  • 总结

    产生磁盘小文件的个数: 2*M(map task的个数)索引文件-和磁盘文件

2) bypass机制(比如wordcount)不需要排序时使用

  • bypass机制示意图
  • 总结

          a) bypass运行机制的触发条件如下:

                shuffle reduce task的数量小于spark.shuffle.sort.bypassMergeThreshold的参数值。这个值默认是200。

          b)产生的磁盘小文件为:2*M(map task的个数)

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏owent

pbc的proto3接入

Protobuf 的 proto3发布也有挺长一段时间了。现在很多新项目慢慢转变用proto3来开发。这篇文章主要记录一下我在给pbc写对proto3支持时的一...

26310
来自专栏SDNLAB

OpenFlow协议库开发者指南

介绍 OpenFlow协议库是OpenDaylight的一个组件,调解OpenDaylight controller和支持OpenFlow协议的硬件设备之间通...

63080
来自专栏Albert陈凯

Spark详解07广播变量BroadcastBroadcast

Broadcast 顾名思义,broadcast 就是将数据从一个节点发送到其他各个节点上去。这样的场景很多,比如 driver 上有一张表,其他节点上运行的 ...

41060
来自专栏Spark学习技巧

必读:Spark与kafka010整合

SparkStreaming与kafka010整合 读本文之前,请先阅读之前文章: 必读:再讲Spark与kafka 0.8.2.1+整合 Spark Str...

83170
来自专栏钟绍威的专栏

java.io.StreamCorruptedException: invalid type code: AC错误的解决方法

问题描述: 在向一个文件写入可序列化对象时,每次只想向文件的末尾添加一个可序列化的对象,于是使用了FileOutputStream(文件名,true)间接的构建...

230100
来自专栏云计算教程系列

如何在CVM上同步自建数据库的数据?

Transporter是一种用于在不同数据存储之间移动数据的开源工具。开发人员经常为诸如跨数据库移动数据,将数据从文件移动到数据库或反之亦然等任务编写一次性脚本...

222120
来自专栏Spark学习技巧

用java提交一个Spark应用程序

第一种方式 经常有人在公众号留言或者在群里问浪尖,如何使用java代码提交一个Spark应用程序。在这里浪尖介绍一种简单的方法来实现这个功能。 首先用vim打开...

424100
来自专栏一名合格java开发的自我修养

RPC原来就是Socket——RPC框架到dubbo的服务动态注册,服务路由,负载均衡演化

  序:RPC就是使用socket告诉服务端我要调你的哪一个类的哪一个方法然后获得处理的结果。服务注册和路由就是借助第三方存储介质存储服务信息让服务消费者调用。...

16320
来自专栏祝威廉

快速开发基于Yarn的分布式应用

mammuthus-yarn-client使得基于Yarn开发分布式应用变得非常容易,基本蔽掉了Yarn的API。 我们后面会把mammuthus-yarn-...

20780
来自专栏比原链

剥开比原看代码12:比原是如何通过/create-account-receiver创建地址的?

Gitee地址:https://gitee.com/BytomBlockchain/bytom

16810

扫码关注云+社区

领取腾讯云代金券