Spark性能优化(shuffle调优)

面向大数据处理的Spark集群架构与性能优化在大数据时代，面向大数据处理的Spark集群架构和性能优化是一个重要的课题。

Spark作为一种快速、通用、可扩展的大数据处理框架，已经成为了许多公司和组织数据处理和分析的首选。

本文将深入探讨Spark集群架构的设计原则，以及如何通过性能优化来提高Spark的处理能力。

首先，我们来了解一下Spark集群架构的设计原则。

Spark集群通常包括一个主节点（Master）和多个工作节点（Worker）。

主节点负责整个集群的管理和调度工作，而工作节点则是实际执行任务的计算节点。

在正式使用Spark之前，我们需要在每个工作节点上安装和配置Spark，并确保它们可以与主节点进行通信。

Spark集群架构的设计原则是可伸缩性和容错性。

伸缩性使得Spark可以轻松地处理大规模数据集，而容错性则确保在节点故障的情况下能够提供可靠的数据处理。

为了实现这些原则，我们可以采取以下策略：1. 增加工作节点的数量：通过增加工作节点的数量，可以提高Spark集群的处理能力。

当负载增加时，可以动态地添加新的工作节点来分担负载，从而实现伸缩性。

同时，多个工作节点之间的数据可以并行处理，进一步提高处理效率。

2. 合理分配资源：在设计Spark集群架构时，需要根据任务的需求来合理分配资源。

可以通过调整每个工作节点的内存和CPU核数来优化性能。

对于内存密集型的任务，可以增加每个工作节点的内存大小；对于计算密集型的任务，可以增加每个工作节点的CPU核数。

此外，还可以通过动态分配资源的方式，根据当前任务的需求来动态调整分配的资源。

3. 数据本地性调度：在Spark中，数据本地性调度是一种重要的优化技术。

它可以将计算任务调度到存储有所需数据的工作节点上，从而避免数据传输的开销。

Spark提供了多种数据本地性调度策略，包括PROCESS_LOCAL、NODE_LOCAL、RACK_LOCAL等。

通过合理选择数据本地性调度策略，可以减少数据传输的开销，提高计算效率。

Spark Shuffle原理、Shuffle操作问题解决和参数调优Spark Shuffle原理、Shuffle操作问题解决和参数调优摘要：1 shuffle原理1.1 mapreduce的shuffle原理1.1.1 map task端操作1.1.2 reduce task端操作1.2 spark现在的SortShuffleManager2 Shuffle操作问题解决2.1 数据倾斜原理2.2 数据倾斜问题发现与解决2.3 数据倾斜解决方案3 spark RDD中的shuffle算子3.1 去重3.2 聚合3.3 排序3.4 重分区3.5 集合操作和表操作4 spark shuffle参数调优内容：1 shuffle原理概述：Shuffle描述着数据从map task输出到reduce task输入的这段过程。

在分布式情况下，reducetask需要跨节点去拉取其它节点上的maptask结果。

这一过程将会产生网络资源消耗和内存，磁盘IO的消耗。

1.1 mapreduce的shuffle原理1.1.1 map task端操作每个，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。

Spill过程：这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写。

整个缓冲区有个溢写的比例spill.percent（默认是memory写，同时溢写线程锁定已用memory，先对key(序列化的字节)做排序,如果client程序设置了Combiner，那么在溢写的过程中就会进行局部聚合。

Merge过程：每次溢写都会生成一个临时文件，在maptask真正完成时会将这些文件归并成一个文件，这个过程叫做Merge。

1.1.2 reducetask端操作当某台TaskTracker上的所有map task执行完成，对应节点的reduce task开始启动，简单地说，此阶段就是不断地拉取(Fetcher)每个maptask所在节点的最终结果，然后不断地做merge形成reducetask的输入文件。

Spark性能调优-RDD算子调优篇Spark调优之RDD算子调优不废话，直接进入正题！1. RDD复用在对RDD进行算子时，要避免相同的算子和计算逻辑之下对RDD进行重复的计算，如下图所示：对上图中的RDD计算架构进行修改，得到如下图所示的优化结果：2. 尽早filter获取到初始RDD后，应该考虑尽早地过滤掉不需要的数据，进而减少对内存的占用，从而提升Spark作业的运行效率。

3. 读取大量小文件-用wholeTextFiles当我们将一个文本文件读取为 RDD 时，输入的每一行都会成为RDD的一个元素。

也可以将多个完整的文本文件一次性读取为一个pairRDD，其中键是文件名，值是文件内容。

val input:RDD[String] = sc.textFile("dir/*.log")如果传递目录，则将目录下的所有文件读取作为RDD。

文件路径支持通配符。

但是这样对于大量的小文件读取效率并不高，应该使用wholeTextFiles返回值为RDD[(String, String)]，其中Key是文件的名称，Value是文件的内容。

def wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)])wholeTextFiles读取小文件:val filesRDD: RDD[(String, String)] =sc.wholeTextFiles("D:\\data\\files", minPartitions = 3)val linesRDD: RDD[String] = filesRDD.flatMap(_._2.split("\\r\\n"))val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(" "))wordsRDD.map((_, 1)).reduceByKey(_ + _).collect().foreach(println)4. mapPartition和foreachPartition•mapPartitionsmap(_….) 表示每一个元素mapPartitions(_….) 表示每个分区的数据组成的迭代器普通的map算子对RDD中的每一个元素进行操作，而mapPartitions算子对RDD中每一个分区进行操作。

Spark性能优化本文章来自于阿里云云栖社区摘要： Spark的性能分析和调优很有意思，今天再写一篇。

主要话题是shuffle，当然也牵涉一些其他代码上的小把戏。

以前写过一篇文章，比较了几种不同场景的性能优化，包括portal的性能优化，web service的性能优化，还有Spark j ob的性能优化。

Spark的性能优化有一些特殊的地方，比如Spark的性能分析和调优很有意思，今天再写一篇。

主要话题是shuffle，当然也牵涉一些其他代码上的小把戏。

以前写过一篇文章，比较了几种不同场景的性能优化（原文链接：/3658?spm=5176.100239.blogcont53509.5.zDcg U3），包括portal的性能优化，web service的性能优化，还有Spark job的性能优化。

Spark的性能优化有一些特殊的地方，比如实时性一般不在考虑范围之内，通常我们用Spark来处理的数据，都是要求异步得到结果的数据；再比如数据量一般都很大，要不然也没有必要在集群上操纵这么一个大家伙，等等。

事实上，我们都知道没有银弹，但是每一种性能优化场景都有一些特定的“大boss”，通常抓住和解决大boss以后，能解决其中一大部分问题。

比如对于portal来说，是页面静态化，对于web service来说，是高并发（当然，这两种可以说并不确切，这只是针对我参与的项目总结的经验而已），而对于Spark来说，这个大boss就是shuffle。

首先要明确什么是shuffle。

Shuffle指的是从map阶段到reduce阶段转换的时候，即map的output向着reduce的input映射的时候，并非节点一一对应的，即干map 工作的slave A，它的输出可能要分散跑到reduce节点A、B、C、D …… X、Y、Z 去，就好像shuffle的字面意思“洗牌”一样，这些map的输出数据要打散然后根据新的路由算法（比如对key进行某种hash算法），发送到不同的reduce节点上去。

Spark性能优化：开发调优篇1、前言在大数据计算领域，Spark已经成为了越来越流行、越来越受欢迎的计算平台之一。

Spark的功能涵盖了大数据领域的离线批处理、SQL类处理、流式/实时计算、机器学习、图计算等各种不同类型的计算操作，应用范围与前景非常广泛。

在美团?大众点评，已经有很多同学在各种项目中尝试使用Spark。

大多数同学(包括笔者在内)，最初开始尝试使用Spark的原因很简单，主要就是为了让大数据计算作业的执行速度更快、性能更高。

然而，通过Spark开发出高性能的大数据计算作业，并不是那么简单的。

如果没有对Spark作业进行合理的调优，Spark作业的执行速度可能会很慢，这样就完全体现不出Spark 作为一种快速大数据计算引擎的优势来。

因此，想要用好Spark，就必须对其进行合理的性能优化。

Spark的性能调优实际上是由很多部分组成的，不是调节几个参数就可以立竿见影提升作业性能的。

我们需要根据不同的业务场景以及数据情况，对Spark作业进行综合性的分析，然后进行多个方面的调节和优化，才能获得最佳性能。

笔者根据之前的Spark作业开发经验以及实践积累，总结出了一套Spark作业的性能优化方案。

整套方案主要分为开发调优、资源调优、数据倾斜调优、shuffle调优几个部分。

开发调优和资源调优是所有Spark作业都需要注意和遵循的一些基本原则，是高性能Spark 作业的基础;数据倾斜调优，主要讲解了一套完整的用来解决Spark作业数据倾斜的解决方案;shuffle调优，面向的是对Spark的原理有较深层次掌握和研究的同学，主要讲解了如何对Spark作业的shuffle运行过程以及细节进行调优。

本文作为Spark性能优化指南的基础篇，主要讲解开发调优以及资源调优。

2、开发调优Spark性能优化的第一步，就是要在开发Spark作业的过程中注意和应用一些性能优化的基本原则。

开发调优，就是要让大家了解以下一些Spark基本开发原则，包括：RDD lineage设计、算子的合理使用、特殊操作的优化等。

【Spark调优】：尽量避免使⽤shuffle类算⼦ 如果有可能的话，尽量避免使⽤shuffle类算⼦。

因为Spark作业运⾏过程中，最消耗性能的地⽅就是shuffle过程。

shuffle过程，就是将分布在集群中多个节点上的同⼀个key，拉取到同⼀个节点上，进⾏聚合或join等操作。

⽐如reduceByKey、join等算⼦，都会触发shuffle操作。

shuffle过程中，各个节点上的相同key都会先写⼊本地磁盘⽂件中，然后其他节点需要通过⽹络传输拉取各个节点上的磁盘⽂件中的相同key。

⽽且相同key都拉取到同⼀个节点进⾏聚合操作时，还有可能会因为⼀个节点上处理的key过多，导致内存不够存放，进⽽溢写到磁盘⽂件中。

因此在shuffle过程中，可能会发⽣⼤量的磁盘⽂件读写的IO操作，以及数据的⽹络传输操作。

会引起⼤量磁盘IO和⽹络数据传输也是shuffle性能较差的主要原因。

因此在⽣产环境开发过程中，能避免则尽可能避免使⽤reduceByKey、join、distinct、repartition等会进⾏shuffle的算⼦，尽量使⽤map类的⾮shuffle算⼦。

这样的话，没有shuffle操作或者仅有较少shuffle操作的Spark作业，可以⼤⼤减少性能开销。

spark中会导致shuffle操作的有以下⼏种算⼦：重分区类操作： ⽐如repartition、repartitionAndSortWithinPartitions、coalesce(shuffle=true)等。

重分区⼀般会shuffle，因为需要在整个集群中，对之前所有的分区的数据进⾏随机，均匀的打乱，然后把数据放⼊下游新的指定数量的分区内。

聚合，byKey类操作： ⽐如reduceByKey、groupByKey、sortByKey等。

byKey类的操作要对⼀个key，进⾏聚合操作，那么肯定要保证集群中，所有节点上的相同的key，移动到同⼀个节点上进⾏处理。

spark性能优化（⼀）本⽂内容说明初始化配置给rdd和dataframe带来的影响repartition的相关说明cache&persist的相关说明性能优化的说明建议以及实例配置说明spark：2.4.0服务器：5台（8核32G）初始化配置项%%init_sparklauncher.master = "yarn" = "BDP-xw"launcher.conf.spark.driver.cores = 1launcher.conf.spark.driver.memory = '1g'launcher.conf.spark.executor.instances = 3launcher.conf.spark.executor.memory = '1g'launcher.conf.spark.executor.cores = 2launcher.conf.spark.default.parallelism = 5launcher.conf.spark.dynamicAllocation.enabled = Falseimport org.apache.spark.sql.SparkSessionvar NumExecutors = spark.conf.getOption("spark.num_executors").reprvar ExecutorMemory = spark.conf.getOption("spark.executor.memory").reprvar AppName = spark.conf.getOption("").reprvar max_buffer = spark.conf.getOption("spark.kryoserializer.buffer.max").reprimport org.apache.spark.rdd.RDDimport org.apache.spark.sql.{DataFrame, Row}import org.apache.spark.sql.SparkSessionimport org.apache.spark.sql.functions.monotonically_increasing_idimport org.apache.log4j.{Level, Logger}import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}import org.apache.spark.sql.functions.{udf, _}import org.apache.spark.{SparkConf, SparkContext}object LoadingData_from_files{def main(args: Tuple2[String, Array[String]]=Tuple2(hdfs_file, etl_date:Array[String])): Unit = {for( a <- etl_date){val hdfs_file_ = s"$hdfs_file" + aval rdd_20210113 = spark.sparkContext.textFile(hdfs_file_).cache()val num1 = rdd_20210113.countprintln(s"加载数据啦:$a RDD的数据量是$num1")}val rdd_20210113_test = spark.sparkContext.textFile(hdfs_file + "20210328").cache()var num1 = rdd_20210113_test.count()println(s"加载数据啦:20210113 RDD的数据量是$num1")rdd_20210113_test.unpersist() // 解除持久化val df_20210420 = spark.sparkContext.textFile(hdfs_file + "20210113").toDF.cache()num1 = df_20210420.count() // 指定memory之后，cache的数量太多之前cache的结果会被⼲掉println(s"加载数据啦:20210420 DataFrame的数据量是$num1")}}// 配置参数multiple_duplicatedval hdfs_file = "hdfs://path/etl_date="val etl_date = Array("20210113","20210112","20210112","20210112","20210112","20210112", "20210113")LoadingData_from_files.main(hdfs_file, etl_date)得到结果如下：结果分析可以看到默认情况下，RDD的缓存⽅式都是到Memory的，⽽DataFrame的缓存⽅式都是Memory and Disk的指定memory之后，cache的数量太多之前cache的结果会被⼲掉⽆特定配置项import org.apache.spark.sql.SparkSessionvar NumExecutors = spark.conf.getOption("spark.num_executors").reprvar ExecutorMemory = spark.conf.getOption("spark.executor.memory").reprvar AppName = spark.conf.getOption("").reprobject LoadingData_from_files{def main(args: Tuple2[String, Array[String]]=Tuple2(hdfs_file, etl_date:Array[String])): Unit = {for( a <- etl_date){val hdfs_file_ = s"$hdfs_file" + aval rdd_20210113 = spark.sparkContext.textFile(hdfs_file_).cache()val num1 = rdd_20210113.countprintln(s"加载数据啦:$a RDD的数据量是$num1")}val rdd_20210113_test = spark.sparkContext.textFile(hdfs_file + "20210328").cache()var num1 = rdd_20210113_test.count()println(s"加载数据啦:20210328 RDD的数据量是$num1")rdd_20210113_test.unpersist() // 解除持久化val df_20210420 = spark.sparkContext.textFile(hdfs_file + "20210113").toDF.cache()num1 = df_20210420.count() // 指定memory之后，cache的数量太多之前cache的结果会被⼲掉println(s"加载数据啦:20210420 DataFrame的数据量是$num1 \n当前环境下cache的个数及id为：")spark.sparkContext.getPersistentRDDs.foreach(i=>println("cache的id：" + i._1))}}// ⽆配置参数multiple_duplicatedval hdfs_file = "hdfs://path/etl_date="val etl_date = Array("20210113","20210112","20210112","20210112","20210112","20210112", "20210113" )LoadingData_from_files.main(hdfs_file, etl_date)得到结果如下：结果分析spark的配置⽂件中，设置的也是动态分配内存；cache的结果也是到达memory限制的时候，已经cache的结果会⾃动消失；上述例⼦中，我们增加了8个⽂件，但最终只保留了5个cache的结果；通过for重复从⼀个⽂件取数，并val声明给相同变量并cache，结果是会被多次保存在memory或者Disk中的；查看当前服务下的所有缓存并删除spark.sparkContext.getPersistentRDDs.foreach(i=>println(i._1))spark.sparkContext.getPersistentRDDs.foreach(i=>{i._2.unpersist()})repartitionrepartition只是coalesce接⼝中shuffle为true的实现repartition 可以增加和减少分区，⽽使⽤ coalesce 则只能减少分区每个block的⼤⼩为默认的128M//RDDrdd.getNumPartitionsrdd.partitions.lengthrdd.partitions.size// For DataFrame, convert to RDD firstdf.rdd.getNumPartitionsdf.rdd.partitions.lengthdf.rdd.partitions.sizeRDD默认cache的级别是Memoryval hdfs_file = "hdfs://path1/etl_date="val rdd_20210113_test = spark.sparkContext.textFile(hdfs_file + "20210113").cache()// ⽂件⼤⼩为1.5Grdd_20210113_test.getNumPartitions// res2: Int = 13val rdd_20210113_test_par1 = rdd_20210113_test.repartition(5)rdd_20210113_test_par1.partitions.size// res9: Int = 5val rdd_20210113_test_par2 = rdd_20210113_test_par1.coalesce(13)rdd_20210113_test_par2.partitions.length// res14: Int = 5 增加分区没⽣效val rdd_20210113_test_par3 = rdd_20210113_test_par1.coalesce(3)rdd_20210113_test_par3.partitions.length// res16: Int = 3 增加分区⽣效DataFrame默认cache的级别是Memory and Diskval hdfs_file = "hdfs://path1/etl_date="val df_20210420 = spark.sparkContext.textFile(hdfs_file + "20210113").toDF().cache()df_20210420.rdd.getNumPartitions// res18: Int = 13val df_20210420_par1 = df_20210420.repartition(20)df_20210420_par1.rdd.getNumPartitions// res19: Int = 20 增加分区⽣效val df_20210420_par2 = df_20210420_par1.coalesce(5)df_20210420_par2.rdd.getNumPartitions// res20: Int = 5cache&persist对⽐cache调⽤的是⽆参数版本的persist()persist的说明import org.apache.spark.storage.StorageLevel._// MEMORY_AND_DISKval hdfs_file = "hdfs://path1/etl_date="var etl_date = "20210113"var hdfs_file_ = s"$hdfs_file" + etl_dateval rdd_20210113_DISK_MEMORY = spark.sparkContext.textFile(hdfs_file_).persist(MEMORY_AND_DISK)println("DISK_ONLY数据量为" + rdd_20210113_DISK_MEMORY.count())// MEMORY_ONLYetl_date = "20210112"hdfs_file_ = s"$hdfs_file" + etl_dateval rdd_20210113_MEMORY_ONLY = spark.sparkContext.textFile(hdfs_file_).persist(MEMORY_ONLY)println("MEMORY_ONLY数据量为" + rdd_20210113_MEMORY_ONLY.count())// DISK_ONLYetl_date = "20210328"hdfs_file_ = s"$hdfs_file" + etl_dateval rdd_20210113_DISK_ONLY = spark.sparkContext.textFile(hdfs_file_).persist(DISK_ONLY)println("DISK_ONLY数据量为" + rdd_20210113_DISK_ONLY.count())// DISK_ONLY数据量为4298617// MEMORY_ONLY数据量为86340// DISK_ONLY数据量为20000性能优化参数说明参数配置建议优化⽅⾯说明tipsyarn集群中⼀般有资源申请上限，如，executor-memory*num-executors < 400G 等，所以调试参数时要注意这⼀点如果GC时间较长，可以适当增加--executor-memory的值或者减少--executor-cores的值yarn下每个executor需要的memory = spark-executor-memory + spark.yarn.executor.memoryOverhead.⼀般需要为，后台进程留下⾜够的cores(⼀般每个节点留⼀个core)。

Spark性能优化指南——⾼级篇前⾔继讲解了每个Spark开发⼈员都必须熟知的开发调优与资源调优之后，本⽂作为《Spark性能优化指南》的⾼级篇，将深⼊分析数据倾斜调优与shuffle调优，以解决更加棘⼿的性能问题。

数据倾斜调优调优概述有的时候，我们可能会遇到⼤数据计算中⼀个最棘⼿的问题——数据倾斜，此时Spark作业的性能会⽐期望差很多。

数据倾斜调优，就是使⽤各种技术⽅案解决不同类型的数据倾斜问题，以保证Spark作业的性能。

数据倾斜发⽣时的现象绝⼤多数task执⾏得都⾮常快，但个别task执⾏极慢。

⽐如，总共有1000个task，997个task都在1分钟之内执⾏完了，但是剩余两三个task却要⼀两个⼩时。

这种情况很常见。

原本能够正常执⾏的Spark作业，某天突然报出OOM（内存溢出）异常，观察异常栈，是我们写的业务代码造成的。

这种情况⽐较少见。

数据倾斜发⽣的原理数据倾斜的原理很简单：在进⾏shuffle的时候，必须将各个节点上相同的key拉取到某个节点上的⼀个task来进⾏处理，⽐如按照key进⾏聚合或join等操作。

此时如果某个key对应的数据量特别⼤的话，就会发⽣数据倾斜。

⽐如⼤部分key对应10条数据，但是个别key却对应了100万条数据，那么⼤部分task可能就只会分配到10条数据，然后1秒钟就运⾏完了；但是个别task可能分配到了100万数据，要运⾏⼀两个⼩时。

因此，整个Spark作业的运⾏进度是由运⾏时间最长的那个task决定的。

因此出现数据倾斜的时候，Spark作业看起来会运⾏得⾮常缓慢，甚⾄可能因为某个task处理的数据量过⼤导致内存溢出。

下图就是⼀个很清晰的例⼦：hello这个key，在三个节点上对应了总共7条数据，这些数据都会被拉取到同⼀个task中进⾏处理；⽽world和you这两个key分别才对应1条数据，所以另外两个task只要分别处理1条数据即可。

此时第⼀个task的运⾏时间可能是另外两个task的7倍，⽽整个stage的运⾏速度也由运⾏最慢的那个task所决定。

SparkSubmit参数及参数性能调优⾸先摆出常⽤的参数设定bin/spark-submit \--class com.xyz.bigdata.calendar.PeriodCalculator \--master yarn \--deploy-mode cluster \--queue default_queue \--num-executors 50 \--executor-cores 2 \--executor-memory 4G \--driver-memory 2G \--conf "spark.default.parallelism=250" \--conf "spark.shuffle.memoryFraction=0.3" \--conf "spark.storage.memoryFraction=0.5" \--conf "spark.driver.extraJavaOptions=-XX:+UseG1GC" \--conf "spark.executor.extraJavaOptions=-XX:+UseG1GC" \--verbose \${PROJECT_DIR}/bigdata-xyz-0.1.jar关于spark-submit的执⾏过程，读Spark Core的源码能够获得⼀个⼤致的印象。

今天事情⽐较多，所以之后会另写⽂章专门叙述关于Spark on YARN的事情（⼜挖了⼀个坑，上⼀个坑是关于Java String和JVM的，需要尽快填上了）。

num-executors含义：设定Spark作业要⽤多少个Executor进程来执⾏。

设定⽅法：根据我们的实践，设定在30~100个之间为最佳。

如果不设定，默认只会启动⾮常少的Executor。

如果设得太⼩，⽆法充分利⽤计算资源。