不可不知的spark shuffle

shuffle概览

一个spark的RDD有一组固定的分区组成,每个分区有一系列的记录组成。对于由窄依赖变换(例如map和filter)返回的RDD,会延续父RDD的分区信息,以pipeline的形式计算。每个对象仅依赖于父RDD中的单个对象。诸如coalesce之类的操作可能导致任务处理多个输入分区,但转换仍然被认为是窄依赖的,因为一个父RDD的分区只会被一个子RDD分区继承。

Spark还支持宽依赖的转换,例如groupByKey和reduceByKey。在这些依赖项中,计算单个分区中的记录所需的数据可以来自于父数据集的许多分区中。要执行这些转换,具有相同key的所有元组必须最终位于同一分区中,由同一任务处理。为了满足这一要求,Spark产生一个shuffle,它在集群内部传输数据,并产生一个带有一组新分区的新stage。

可以看下面的代码片段:

sc.textFile("someFile.txt").map(mapFunc).flatMap(flatMapFunc).filter(filterFunc).count()

上面的代码片段只有一个action操作,count,从输入textfile到action经过了三个转换操作。这段代码只会在一个stage中运行,因为,三个转换操作没有shuffle,也即是三个转换操作的每个分区都是只依赖于它的父RDD的单个分区。

但是,下面的单词统计就跟上面有很大区别:

val tokenized = sc.textFile(args(0)).flatMap(_.split(' '))
val wordCounts = tokenized.map((_, 1)).reduceByKey(_ + _)
val filtered = wordCounts.filter(_._2 >= 1000)
val charCounts = filtered.flatMap(_._1.toCharArray).map((_, 1)).reduceByKey(_ + _)
charCounts.collect()

这段代码里有两个reducebykey操作,三个stage。

下面图更复杂,因为有一个join操作:

粉框圈住的就是整个DAG的stage划分。

在每个stage的边界,父stage的task会将数据写入磁盘,子stage的task会将数据通过网络读取。由于它们会导致很高的磁盘和网络IO,所以shuffle代价相当高,应该尽量避免。父stage的数据分区往往和子stage的分区数不同。触发shuffle的操作算子往往可以指定分区数的,也即是numPartitions代表下个stage会有多少个分区。就像mr任务中reducer的数据是非常重要的一个参数一样,shuffle的时候指定分区数也将在很大程度上决定一个应用程序的性能。

优化shuffle

通常情况可以选择使用产生相同结果的action和transform相互替换。但是并不是产生相同结果的算子就会有相同的性能。通常避免常见的陷阱并选择正确的算子可以显著提高应用程序的性能。

当选择转换操作的时候,应最小化shuffle次数和shuffle的数据量。shuffle是非常消耗性能的操作。所有的shuffle数据都会被写入磁盘,然后通过网络传输。repartition , join, cogroup, 和 *By 或者 *ByKey 类型的操作都会产生shuffle。我们可以对一下几个操作算子进行优化:

1. groupByKey某些情况下可以被reducebykey代替。

2. reduceByKey某些情况下可以被 aggregatebykey代替。

3. flatMap-join-groupBy某些情况下可以被cgroup代替。

具体细节,知识星球球友可以点击阅读原文进入知识星球阅读。

no shuffle

在某些情况下,前面描述的转换操作不会导致shuffle。当先前的转换操作已经使用了和shuffle相同的分区器分区数据的时候,spark就不会产生shuffle。

举个例子:

rdd1 = someRdd.reduceByKey(...)

rdd2 = someOtherRdd.reduceByKey(...)

rdd3 = rdd1.join(rdd2)

由于使用redcuebykey的时候没有指定分区器,所以都是使用的默认分区器,会导致rdd1和rdd2都采用的是hash分区器。两个reducebykey操作会产生两个shuffle过程。如果,数据集有相同的分区数,执行join操作的时候就不需要进行额外的shuffle。由于数据集的分区相同,因此rdd1的任何单个分区中的key集合只能出现在rdd2的单个分区中。 因此,rdd3的任何单个输出分区的内容仅取决于rdd1中单个分区的内容和rdd2中的单个分区,并且不需要第三个shuffle。

例如,如果someRdd有四个分区,someOtherRdd有两个分区,而reduceByKeys都使用三个分区,运行的任务集如下所示:

如果rdd1和rdd2使用不同的分区器或者相同的分区器不同的分区数,仅仅一个数据集在join的过程中需要重新shuffle

在join的过程中为了避免shuffle,可以使用广播变量。当executor内存可以存储数据集,在driver端可以将其加载到一个hash表中,然后广播到executor。然后,map转换可以引用哈希表来执行查找。

增加shuffle

有时候需要打破最小化shuffle次数的规则。

当增加并行度的时候,额外的shuffle是有利的。例如,数据中有一些文件是不可分割的,那么该大文件对应的分区就会有大量的记录,而不是说将数据分散到尽可能多的分区内部来使用所有已经申请cpu。在这种情况下,使用reparition重新产生更多的分区数,以满足后面转换算子所需的并行度,这会提升很大性能。

使用reduce和aggregate操作将数据聚合到driver端,也是修改区数的很好的例子。

在对大量分区执行聚合的时候,在driver的单线程中聚合会成为瓶颈。要减driver的负载,可以首先使用reducebykey或者aggregatebykey执行一轮分布式聚合,同时将结果数据集分区数减少。实际思路是首先在每个分区内部进行初步聚合,同时减少分区数,然后再将聚合的结果发到driver端实现最终聚合。典型的操作是treeReduce 和 treeAggregate。

当聚合已经按照key进行分组时,此方法特别适用。例如,假如一个程序计算语料库中每个单词出现的次数,并将结果使用map返回到driver。一种方法是可以使用聚合操作完成在每个分区计算局部map,然后在driver中合并map。可以用aggregateByKey以完全分布的方式进行统计,然后简单的用collectAsMap将结果返回到driver。

原文发布于微信公众号 - Spark学习技巧(bigdatatip)

原文发表时间:2018-08-23

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏美图数据技术团队

Spark Streaming VS Flink

本文从编程模型、任务调度、时间机制、Kafka 动态分区的感知、容错及处理语义、背压等几个方面对比 Spark Stream 与 Flink,希望对有实时处理...

1651
来自专栏cloudskyme

hadoop使用(五)

第1章 引言 1.1 编写目的 对关于hadoop的文档及资料进行进一步的整理。 1.2 相关网站    毋庸置疑 http://hadoop.apache.o...

3515
来自专栏我是攻城师

理解Spark的运行机制

4639
来自专栏恰童鞋骚年

Hadoop学习笔记—10.Shuffle过程那点事儿

  在第四篇博文《初识MapReduce》中,我们认识了MapReduce的八大步骤,其中在Reduce阶段总共三个步骤,如下图所示:

1062

将R与Cloudera Impala集成,以实现Hadoop上的实时查询

Cloudera Impala支持Hadoop数据集上的低延迟交互式查询,这些数据集可以存储在Hadoop分布式文件系统(HDFS)或Hadoop的分布式NoS...

6777
来自专栏牛肉圆粉不加葱

YARN资源调度器

1716
来自专栏Hadoop实操

如何通过CM禁用Federation

本文主要讲述如何通过CM禁用Federation。我前面写过两篇关于Federation的文章,可参看《HDFS Federation(联邦)简介》和《如何通过...

1344
来自专栏美团技术团队

【技术博客】Spark性能优化指南——高级篇

前言 继基础篇讲解了每个Spark开发人员都必须熟知的开发调优与资源调优之后,本文作为《Spark性能优化指南》的高级篇,将深入分析数据倾斜调优与shuffle...

5696
来自专栏企鹅号快讯

大数据入门基础系列之浅谈Hive和HBase的区别

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。 在前面的博文里,我已经介绍了 Hive和HBase分别是什么? Apache Hive是一个构建...

2386
来自专栏星回的实验室

打造自己的MapReduce[一]:Hadoop集群搭建

那就是我在这里的每一篇文开头都必然是:最近工作好忙,又断更很久了……Anyway,这也不能成为偷懒的理由。我可能对记录技术有些固执的误解,总认为是要待到整理出一...

1111

扫码关注云+社区

领取腾讯云代金券