Spark性能优化(数据倾斜调优)

1、Spark性能优化：数据倾斜曹征1.数据倾斜曹征调整简介在某些情况下，大数据计算中最棘手的问题之一数据可能会倾斜。此时，火花工作的性能会远远低于预期。调整数据倾斜是通过使用多种技术解决方案解决不同类型的数据倾斜问题，以确保火花操作的性能。数据倾斜时的现象大多数task的运行速度非常快，但单个task的运行速度非常慢。例如，您总共有1000个task，997个task在一分钟内运行，但剩馀的2-3个task需要1-2个小时。这种情况很常见。原来正常运行的火花作业有一天突然报告内存溢出(OOM)异常，观察异常堆栈是因为我们编写的业务代码。这种情况比较罕见。数据倾斜的发生原理数据倾斜的原理很简单。在

2、shuffle过程中，必须将每个节点上的相同关键点拖动到节点上的一个task中进行处理。例如，按key聚集或执行join。此时，如果特定键的数据杨怡特别多，则会发生数据倾斜。例如，大部分密钥对应于10个数据，而单个密钥对应于100万个数据，则大部分task只能分配给10个数据，并且可以在一秒内运行。但是，单个task可能分配给运行1到2个小时的100万个数据。因此，整个Spark作业的执行进度由运行时间最长的相应task确定。因此，如果数据倾斜，Spark操作看起来运行得很慢，甚至可能因特定task处理的数据量过多而发生内存泄漏。下图是一个明显的示例。名为hello的key在三个节点上共对应于

3、7个数据，所有这些数据都被拉至同一task进行处理。World和you分别对应于一个数据，因此其他两个task分别处理一个数据就可以了。第一个task的运行时间是其他两个task的7倍，并且整个stage的运行速度由最慢的task决定。如何查找导致数据倾斜的代码数据倾斜仅发生在shuffle进程中。以下是可以触发shuffle操作的常见运算符：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup和repartition。如果数据倾斜，则可能是在代码中使用这些运算符之一的结果。Task运行特别慢第一个问题是，数据倾斜发生在第几个s

4、tage上。如果以Yarn-client模式提交，则可以在本地查看日志，并且日志中当前运行的stage中选择所需的构件。在Yarn-cluster模式下提交时，Spark Web UI可以查看当前是否正在运行到第几个stage。您还可以在Spark Web UI中详细查看当前stage中的单个task分配的数据量，无论您是在yarn-client模式还是yarn-cluster模式下，都可以进一步了解task分配的数据不均匀是否导致了数据倾斜。例如，在下图中，最后第三列显示了每个task的运行时间。您可能会发现，某些task的运行速度非常快，只需几秒钟即可运行。某些task的运行时间非常慢，运

5、行需要几分钟，此时只能在运行时间确定数据的斜率。倒计时的第一列还显示了每个task处理的数据的杨怡，运行时间特别短的task只需要处理数百KB的数据，运行时间特别长的task需要处理数千KB的数据，处理的数据量减少了10倍。您可以更好地知道此时是否发生了数据倾斜。知道了发生数据倾斜的stage后，必须根据stage拆分原理计算发生倾斜的stage的相应代码中具有shuffle类运算符的部分。正确计算需要深入了解火花源代码的stage和代码的对应关系，当火花代码出现一个shuffle类运算符或火花Spark SQL语句出现导致shuffle的语句(例如group by语句)时，您可以根据该位置确

6、定两个stage是分开的。这里，我们以火花的基本入门级程序字数为例，看一下如何用最简单的方法粗略计算对应于Spark的代码。在下面的示例中，整个代码中只有一个reduceByKey是引发shuffle的运算符，因此，您可以将此运算符作为边界，将前后两个stage拆分开。Stage0主要在textFile中使用map，并执行shuffle write操作。这可以理解为Shuffle write操作，简单地将pairs RDD上的数据分区，每个task处理的数据中的相同密钥都写入到同一磁盘文件中。Stage1主要执行从reduceByKey到collect的任务，在stage1中的每个task从头

7、开始运行时，将首先执行shuffle read任务。执行Shuffle read操作的task在stage0的每个task节点上拖动其处理的关键点，然后对同一关键点执行完整集合或join等操作。其中累计key的value值。Stage1运行reduceByKey运算符后，计算最终的wordCounts RDD，运行collect运算符，将所有数据拖到Driver中，遍历输出并打印输出。plain view plain copy1.valconf=newparkconf()2.valsc=newparkcontext (conf)3.4.vallines=sc . textfile( HDFS

8、:/. ）5.valwords=lines.flatmap (_ .split(“”)6.val pairs=words . map(_，1)7.valword counts=pairs . reduce bykey(_ _ _)8.9.wordcounts.collect()。foreach (println (_)我想分析字数程序，告诉您stage分割的最基本原理，以及stage分割后shuffle操作在两个stage的边界上是如何执行的。然后，您可以了解如何快速找到发生数据倾斜的stage的相应代码部分。例如，如果Spark Web UI或本地log确定stage1中的某些task的执行

9、速度特别慢，并且stage1具有数据倾斜，则可以通过确认stage1主要包含shuffle类运算符(例如reduceByKey)来确定educeByKey运算符导致的数据倾斜问题。例如，如果一个单词出现了100万次，另一个单词出现了10次，则stage1中的一个task处理100万个数据时，整个stage的速度会被此task减慢。任何task以某种方式导致内存溢出在这种情况下，查找问题的代码更容易。建议您直接查看YARN-client模式下本地日志的异常错误堆栈，或通过YARN查看yarn-cluster模式下log中的异常错误堆栈。通常，异常堆栈信息允许代码跳转到发生内存溢出的行。然后浏览代

10、码行，通常还会有shuffle类运算符。此操作符很可能导致数据倾斜。但是要注意，不能因为不小心泄漏了内存就判断出发生了数据倾斜。因为您编写的代码中的错误和意外发生的数据异常会导致内存溢出。如上所述，您必须查看通过Spark Web UI报告错误的stage的单独task运行时间和分配的数据量，以确定是否发生了内存泄漏。查看key中数据倾斜的分布知道数据倾斜发生的位置后，通常会执行shuffle操作，分析数据倾斜的RDD/Hive表以查看关键点分布。这主要是为以后选择什么技术方案提供了依据。徐璐对于不同的关键点分布和徐璐其他“shuffle”操作符组合在一起的不同情况，您可能需要徐璐选择不同的技

11、术解决方案。根据您正在执行的任务，有多种查看关键点分布的方法。1.Spark SQL中的group by，join语句导致数据倾斜时，查询SQL中使用的表的key分布。2.如果在Spark RDD上运行shuffle运算符导致数据倾斜，则可以在Spark操作中添加代码以查看键分布，如RDD.countByKey()。然后，对于收集的每个密钥的出现次数，可以通过collect/take打印到客户端来检查密钥分布。例如，对于上面提到的字数程序，如果确定stage1的reduceByKey运算符会导致数据倾斜，则需要查看执行reduceByKey操作的RDD的键分配。在这种情况下，例如pairs R

12、DD。在以下示例中，可以首先对pairs的10%采样数据进行采样，然后使用countByKey运算符计算每个关键点的发生次数，最后在客户端上遍历和打印采样数据中单个关键点的发生次数。plain view plain copy1.valsampledairs=pairs . sample(false，0.1)2.valsampledordcounts=sampled pairs . count bykey()3.sampledordcounts . foreach(print ln(_)2.数据倾斜解决方案解决方案1:使用蜂巢ETL预处理数据案例套用案例：资料倾斜是Hive表格。如果Hive表中

13、的数据本身不统一(例如，关键字对应于百万数据，其他关键字对应于10个数据)，则此技术适用于业务场景中需要经常使用火花对hit表进行分析的情况。方案实施思路：此时，评估是否可以通过Hive预处理数据(即，是否可以通过Hive ETL将数据预聚合到key或预联接到其他表)后，火花操作中要作为目标的数据源是预处理的Hive表，而不是原始Hive表。现在，Spark作业不再需要使用原始shuffle类运算符执行这些操作，因为已收集数据或执行了join操作。方案实施原则：由于此方案完全避免在Spark中运行shuffle类运算符，因此从源角度解决了数据倾斜。但是要提醒你，这种方式是消除根本。最终，数据本

14、身存在不均匀的散布问题，因此在Hive ETL上执行组by或join等shuffle操作时，数据仍然倾斜，从而降低Hive ETL的速度。为了避免火花程序中的数据倾斜，我们只能将数据倾斜提前到蜂巢ETL。方案优势：可以简单方便地实施，从而完全消除数据倾斜并显着提高火花工作的性能。方案缺点：数据在hitetl(而不是根本原因)中倾斜。案例实践经验：在与某些Java系统搭配使用Spark的专案中，如果Java程式码经常呼叫Spark动作，并且Spark动作的效能需求很高，则此案例更为适用。将数据提前到上游Hive ETL，每天运行一次更慢，然后每次Java调用Spark作业时执行速度更快，从而提供

15、更好的用户体验。项目实践经验：此方案主要用于美国集团评论的交互式用户行为分析系统，该系统允许用户通过Java web系统提交数据分析统计作业，后端通过Java提交Spark作业以分析数据分析统计。Spark操作必须快，并且必须在可能的10分钟内完成。否则，速度太慢，用户环境可能会很差。因此，我们将一些火花工作的shuffle工作提前到了Hive ETL。因此，通过直接使用预处理spark的Hive临时表，将spark的shuffle操作降至最低，显着提高了性能，并将某些操作的性能提高了6倍以上。解决方案2:过滤导致倾斜的少数关键点案例套用案例：如果导致斜度的索引键较少，对计算本身影响不大，则此

16、案例适用。例如，99%的密钥对应于10个数据，但只有1个密钥对应于100万个数据，因此数据倾斜。方案实施思路：判断数据杨怡，特别是许多关键点，如果任务执行和计算结果不是特别重要，则简单地过滤掉其中的几个关键点。例如，在Spark SQL中，您可以使用where子句过滤这些关键字，或在Spark Core中对RDD运行filter运算符以过滤这些关键字。要在每次作业运行时过滤之前动态确定数据杨怡中最多的关键点，可以使用sample运算符对RDD进行采样，然后计算每个关键点的数量，最后使用数据杨怡中最多的关键点进行过滤。方案实施原则：在过滤了数据倾斜的键后，不能生成数据倾斜，因为不参与计算。方案优势：简单有效，完全避免了数据倾斜。场景缺点：适用的场景不多，在大多数情况下，导致倾斜的关键点仍然很多，并且不仅仅是几个。情景实践经验：使用此解决方案解决了项目中的数据坡率。一天，Spark作业运行时突然发现了OOM，跟踪发现蜂巢表的键在当天异常地增加了数据杨怡爆炸。因此