sql调优.md

数据倾斜调优

1、数据倾斜发生时的现象

• 绝大多数task执行得都非常快，但个别task执行极慢。比如，总共有1000个task，997个task都在1分钟之 内执行完了，但是剩余两三个task却要一两个小时。这种情况很常见。
• 原本能够正常执行的Spark作业，某天突然报出OOM（内存溢出）异常，观察异常栈，是我们写的业务代码 造成的。这种情况比较少见。

2、数据倾斜发生的原理

在进行shuffle的时候，必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理，比如按照key进行聚合或join等操作。此时如果某个key对应的数据量特别大的话，就会发生数据倾斜。比如大部分key对应10条数据，但是个别key却对应了100万条数据，那么大部分task可能就只会分配到10条数据，然后1秒钟就运行完了；但是个别task可能分配到了100万数据，要运行一两个小时。因此，整个Spark作业的运行进度是由运行时间最长的那个task决定的。

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

3、如何定位导致数据倾斜的代码

数据倾斜只会发生在shuffle过程中。常用的并且可能会触发shuffle操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition 等。出现数据倾斜时，可能就是代码中使用了这些算子中的某一个所导致的。

只要看到Spark代码中出现了一个shuffle类算子或者是Spark SQL的SQL语句中出现了会导致shuffle的语句（比如group by语句），那么就可以判定，以那个地方为界限划分出了 前后两个stage。

知道数据倾斜发生在哪一个stage之后，接着我们就需要根据stage划分原理，推算出来发生倾斜的那个stage对应代码中的哪一部分，这部分代码中肯定会有一个shuffle类算子。

4、某个task莫名其妙内存溢出的情况

这种情况下去定位出问题的代码就比较容易了。我们建议直接看yarn-client模式下本地log的异 常栈，或者是通过YARN查看yarn-cluster模式下的log中的异常栈。一般来说，通过异常栈信息 就可以定位到代码中哪一行发生了内存溢出。然后在那行代码附近找找，一般也会有shuffle 类算子，此时很可能就是这个算子导致了数据倾斜。

但是也要注意，不能单纯靠偶然的内存溢出就判定发生了数据倾斜。因为自己编写的代码 的bug，以及偶然出现的数据异常，也可能会导致内存溢出。因此还是要按照上面所讲的方法，通过Spark Web UI查看报错的那个stage的各个task的运行时间以及分配的数据量，才能确定是否是由于数据倾斜才导致了这次内存溢出。

5、查看导致数据倾斜的key的数据分布情况

知道了数据倾斜发生在哪里之后，通常需要分析一下那个执行了shuffle操作并且导致了数据倾斜的RDD/Hive表，查看一下其中key的分布情况。这主要是为之后选择哪一种技术方案提供依据。 针对不同的key分布与不同的shuffle算子组合起来的各种情况，可能需要选择不同的技术方案来 解决。 此时根据你执行操作的情况不同，可以有很多种查看key分布的方式：

1. 如果是Spark SQL中的group by、join语句导致的数据倾斜，那么就查询一下SQL中使用的表的key分布情 况。 

2. 如果是对Spark RDD执行shuffle算子导致的数据倾斜，那么可以在Spark作业中加入查看key分布的代码，比如RDD.countByKey()。
   然后对统计出来的各个key出现的次数，collect/take到客户端打印一下，就可以 看到key的分布情况。

解决方案

我认为的调优分为两部分，一个是SQL调优，另一个属于使用层的调优

SQL调优

1. join时无效id产生的倾斜

这种方法不仅减少IO，还减少作业数

select a.detail, b.reg_date, b.uid 
from dwd.fact_log_detail a
left join dwd.dim_user_info b
     on case when a.useid is null then concat('hive', rand())
     else a.userid
     end = b.uid;

2. 不同数据类型关联时产生数据倾斜

对于日志表，如果id既有bigint，又有string类型。关联时做hash值，string先转为int再关联分配reducer。相同字符串的id还是会分到一起。但我个人认为，ETL中就应该把字段类型统一，这种方法也是不得已而为之。

select * from log left join dwd.fact_product_detail p on cast(log.id as int)=p.pid

3. 一个字段关联两列值

多次join转化成union，因为多个union all会优化成一个job 这样只有一个mr作业。两个表都只读一次。b表每读取一行就会打上一行标签，这样变成两个<key, value>对。

select * from log a
inner join
( 
   select price_id as price_id from fact_produce_sale_detail
   union all 
   select money_id as price_id from fact_produce_sale_detail
) b
on a.price_id = b.price_id;

4. 消灭子查询的group by 原写法:

select * from (
    select * from t1 group by c1, c2, c3
    union all
    select * from t2 group by c1, c2, c3) tmp
group by c1, c2, c3

优化:

select * from (
    select * from t1
    union all
    select * from t2) tmp
group by c1,c2,c3

从业务逻辑来将，子查询的group by功能和外层的group by重复，除非子查询有count(distinct)。 数据是一致的。MR的作业数由3减少到1。t1相当于一个目录，t2 相当于一个目录，对map reduce程序来说，t1，t2可以作为map reduce作业的mutli inputs。这可以通过一个map reduce来解决这个问题。 Hadoop计算框架，不怕数据多，怕作业数多。

5. 消灭子查询内的count(distinct), max, min

通过建立临时表，消除子查询内的count(distinct) 原写法:

select c1, c2, c3, sum(pv) from (
    select c1, c2, c3, count(c4) from t1 group by c1, c2, c3
    union all
    select c1, c2, c3, count(distinct c4) from t2 group by c1, c2, c3)t
group by c1, c2, c3

优化：

insert into t4
select c1, c2, c3, count(distinct c4) from t2 group by c1, c2, c3;

select c1, c2, c3, sum(pv) from (
    select c1, c2, c3, count(c4) from t1
    union all
    select * from t4)tmp
group by c1, c2, c3

6. 大表join

谓词下推，先把能过滤掉的主键过滤掉，减少表的连接操作。

8. 过多的where条件

有时候超级多的where条件来限制查询，这样子是非常低效的，主要原因是这个and条件在hive生成执行计划时，产生了一个嵌套层次很多的算子。 解决方案：

1. 把筛选条件对应的值写入一张小表，再一次join到主表。 
2. 利用udf处理。

业务层调优

一、 使用Hive ETL预处理数据

方案适用场景：导致数据倾斜的是Hive表。如果该Hive表中的数据本身很不均匀(比如某个key对应了100万数据，其他key才对应了10条数据)，而且业务场景需要频繁使用Spark对Hive表执行某个分析操作，那么比较适合使用这种技术方案。

方案实现思路：此时可以评估一下，是否可以通过Hive来进行数据预处理（即通过Hive ETL预先对数据按照key进行聚合，或者是预先和其他表进行join），然后在Spark作业中针对的数据源就不是原来的Hive表了，而是预处理后的Hive表。此时由于数据已经预先进行过聚合或join操作了，那么在Spark作业中也就不需要使用原先的shuffle类算子执行这类操作了。

方案实现原理：这种方案从根源上解决了数据倾斜，因为彻底避免了在Spark中执行shuffle类算子，那么肯定就不会有数据倾斜的问题了。但是这里也要提醒一下，这种方式属于治标不治本。因为毕竟数据本身就存在分布不均匀的问题，所以Hive ETL中进行group by或者join等shuffle操作时，还是会出现数据倾斜，导致Hive ETL的速度很慢。我们只是把数据倾斜的发生提 前到了Hive ETL中，避免Spark程序发生数据倾斜而已。

方案优点：实现起来简单便捷，效果还非常好，完全规避掉了数据倾斜，Spark作业的性能会大 幅度提升。

方案缺点：治标不治本，Hive ETL中还是会发生数据倾斜。

方案实践经验：在一些Java系统与Spark结合使用的项目中，会出现Java代码频繁调用Spark作业的场景，而且对Spark作业的执行性能要求很高，就比较适合使用这种方案。将数据倾斜提前到 上游的Hive ETL，每天仅执行一次，只有那一次是比较慢的，而之后每次Java调用Spark作业时，执行速度都会很快，能够提供更好的用户体验。

项目实践经验：在自主研发的报表下载引擎使用了这种方案，该系统主要是允许用户通过Java Web系统提交数据统计任务，后端通过Python提交Spark作业进行数据分析统计。 要求Spark作业速度必须要快，尽量在5分钟以内，否则速度太慢，用户体验会很差。所以我们将有些Spark作业的shuffle操作提前到了Hive ETL中，从而让Spark直接使用预处理的Hive中间表，尽可能地减少Spark的shuffle操作，大幅度提升了性能，将部分作业的性能提升了n倍以上。

二、 过滤少数导致倾斜的key和无效的key

方案适用场景：如果发现导致倾斜的key就少数几个，而且对计算本身的影响并不大的话，那么很适合使用这种方案。比如99%的key就对应10条数据，但是只有一个key对应了100万数据，从而导致了数据倾斜。

方案实现思路：如果判断出导致shuffle产生的key对作业的执行和计算结果不是特别重要的话，那么干脆就直接过滤掉那少数几个key。比如，在Spark SQL中可以使用where子句过滤掉这些key或者在Spark Core中对RDD执行filter算子过滤掉这些key。如果需要每次作业执行时，动态判定哪些key的数据量最多然后再进行过滤，那么可以使用sample算子对RDD进行采样，然后计算出每个key的数量，取数据量最多的key过滤掉即可。

方案实现原理:将导致数据倾斜的key给过滤掉之后，这些key就不会参与计算了，自然不可能产生数据倾斜。

方案优点：实现简单，而且效果也很好，可以完全规避掉数据倾斜。

方案缺点：适用场景不多，大多数情况下，导致倾斜的key还是很多的，并不是只有少数几个。

方案实践经验：在项目中我们也采用过这种方案解决数据倾斜。有一次发现某一天Spark作业在运行的时候突然OOM了，追查之后发现，是Hive表中的某一个key在那天数据异常，导致数据量暴增。因此就采取每次执行前先进行采样，计算出样本中数据量最大的几个key之后，直接在程序中将那些key给过滤掉

三、 提高shuffle操作的并行度

方案适用场景：如果我们必须要对数据倾斜迎难而上，那么建议优先使用这种方案，因为这是处 理数据倾斜最简单的一种方案。

方案实现思路：在对RDD执行shuffle算子时，给shuffle算子传入一个参数，比如 reduceByKey(1000)，该参数就设置了这个shuffle算子执行时shuffle read task的数量。对于 Spark SQL中的shuffle类语句，比如group by、join等，需要设置一个参数，即 spark.sql.shuffle.partitions，该参数代表了shuffle read task的并行度，该值默认是200，对于很多场景来说都有点过小。

方案实现原理：增加shuffle read task的数量，可以让原本分配给一个task的多个key分配给多个task，从而让每个task处理比原来更少的数据。举例来说，如果原本有5个key，每个key对应10条数据，这5个key都是分配给一个task的，那么这个task就要处理50条数据。而增加了shuffle read task以后，每个task就分配到一个key，即每个task就处理10条数据，那么自然每个task的执行时间都会变短了。

方案优点：实现起来比较简单，可以有效缓解和减轻数据倾斜的影响。

方案缺点：只是缓解了数据倾斜而已，没有彻底根除问题，根据实践经验来看，其效果有限。

方案实践经验：该方案通常无法彻底解决数据倾斜，因为如果出现一些极端情况，比如某个key对应的数据量有100万，那么无论你的task数量增加到多少，这个对应着100万数据的key肯定还是会分配到一个task中去处理，因此注定还是会发生数据倾斜的。所以这种方案只能说是在发现数据倾斜时尝试去用最简单的方法缓解数据倾斜而已，或者是和其他方案结合起来使用。

四、 两阶段聚合（局部聚合+全局聚合）

方案适用场景：对RDD执行reduceByKey等聚合类shuffle算子或者在Spark SQL中使用group by语句进行分组聚合时，比较适用这种方案。

方案实现思路：这个方案的核心实现思路就是进行两阶段聚合。第一次是局部聚合，先给每个key都打上一个随机数，比如10以内的随机数，此时原先一样的key就变成不一样的了，比如 (hello, 1) (hello, 1) (hello, 1) (hello, 1)，就会变成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着对打上随机数后的数据，执行reduceByKey等聚合操作，进行局部聚合，那么 局部聚合结果，就会变成了(1_hello, 2) (2_hello, 2)。然后将各个key的前缀给去掉，就会变成 (hello,2)(hello,2)，再次进行全局聚合操作，就可以得到最终结果了，比如(hello, 4)。

方案实现原理：将原本相同的key通过附加随机前缀的方式，变成多个不同的key，就可以让原本 被一个task处理的数据分散到多个task上去做局部聚合，进而解决单个task处理数据量过多的问 题。接着去除掉随机前缀，再次进行全局聚合，就可以得到最终的结果。具体原理见下图。

方案优点：对于聚合类的shuffle操作导致的数据倾斜，效果是非常不错的。通常都可以解决掉数 据倾斜，或者至少是大幅度缓解数据倾斜，将Spark作业的性能提升数倍以上。

方案缺点：仅仅适用于聚合类的shuffle操作，适用范围相对较窄。如果是join类的shuffle操作，还得用其他的解决方案。

五、 将reduce join转为map join

方案适用场景：在对RDD使用join类操作，或者是在Spark SQL中使用join语句时，而且join操作 中的一个RDD或表的数据量比较小（比如几百M或者一两G），比较适用此方案。

方案实现思路：不使用join算子进行连接操作，而使用Broadcast变量与map类算子实现join操作，进而完全规避掉shuffle类的操作，彻底避免数据倾斜的发生和出现。将较小RDD中的数据直 接通过collect算子拉取到Driver端的内存中来，然后对其创建一个Broadcast变量；接着对另外一个RDD执行map类算子，在算子函数内，从Broadcast变量中获取较小RDD的全量数据，与当 前RDD的每一条数据按照连接key进行比对，如果连接key相同的话，那么就将两个RDD的数据 用你需要的方式连接起来。

方案实现原理：普通的join是会走shuffle过程的，而一旦shuffle，就相当于会将相同key的数据 拉取到一个shuffle read task中再进行join，此时就是reduce join。但是如果一个RDD是比较小 的，则可以采用广播小RDD全量数据+map算子来实现与join同样的效果，也就是map join，此时就不会发生shuffle操作，也就不会发生数据倾斜。具体原理如下图。

方案优点：对join操作导致的数据倾斜，效果非常好，因为根本就不会发生shuffle，也就根本不 会发生数据倾斜。

方案缺点：适用场景较少，因为这个方案只适用于一个大表和一个小表的情况。毕竟我们需要将小表进行广播，此时会比较消耗内存资源，driver和每个Executor内存中都会驻留一份小RDD的 全量数据。如果我们广播出去的RDD数据比较大，比如10G以上，那么就可能发生内存溢出了。 因此并不适合两个都是大表的情况。

六、 采样倾斜key并分拆join操作

方案适用场景：两个RDD/Hive表进行join的时候，如果数据量都比较大，无法采用“解决方案 五”，那么此时可以看一下两个RDD/Hive表中的key分布情况。如果出现数据倾斜，是因为其中 某一个RDD/Hive表中的少数几个key的数据量过大，而另一个RDD/Hive表中的所有key都分布 比较均匀，那么采用这个解决方案是比较合适的。

方案实现思路：

1. 对包含少数几个数据量过大的key的那个RDD，通过sample算子采样出一份样本来，然后统计一下每个key 的数量，计算出来数据量最大的是哪几个key。
2. 然后将这几个key对应的数据从原来的RDD中拆分出来，形成一个单独的RDD，并给每个key都打上n以内的 随机数作为前缀，而不会导致倾斜的大部分key形成另外一个RDD。
3. 接着将需要join的另一个RDD，也过滤出来那几个倾斜key对应的数据并形成一个单独的RDD，将每条数据 膨胀成n条数据，这n条数据都按顺序附加一个0~n的前缀，不会导致倾斜的大部分key也形成另外一个RDD。
4. 再将附加了随机前缀的独立RDD与另一个膨胀n倍的独立RDD进行join，此时就可以将原先相同的key打散 成n份，分散到多个task中去进行join了。
5. 而另外两个普通的RDD就照常join即可。
6. 最后将两次join的结果使用union算子合并起来即可，就是最终的join结果。

eg:

num表是一张一列30行的表，对应1-30的正整数。这样做可以把users表膨胀成N份，然后根据 log表的id和login_time分配到不同的reduce里，保证数据均匀。

select * from log a 
left join 
(
    select uid, number from users d 
    join num e
) b 
on a.uid = b.uid 
and mod(a.login_time, 30)+1 = b.number

方案实现原理：对于join导致的数据倾斜，如果只是某几个key导致了倾斜，可以将少数几个key分拆成独立RDD，并附加随机前缀打散成n份去进行join，此时这几个key对应的数据就不会集中 在少数几个task上，而是分散到多个task进行join了。具体原理见下图。

方案优点：对于join导致的数据倾斜，如果只是某几个key导致了倾斜，采用该方式可以用最有效的方式打散key进行join。而且只需要针对少数倾斜key对应的数据进行扩容n倍，不需要对全量数据进行扩容。避免了占用过多内存。

方案缺点：如果导致倾斜的key特别多的话，比如成千上万个key都导致数据倾斜，那么这种方式也不适合。

七、 使用随机前缀和扩容RDD进行join

方案适用场景：如果在进行join操作时，RDD中有大量的key导致数据倾斜，那么进行分拆key也 没什么意义，此时就只能使用最后一种方案来解决问题了。

方案实现思路：

1. 该方案的实现思路基本和“解决方案六”类似，首先查看RDD/Hive表中的数据分布情况，找到那个造成数 据倾斜的RDD/Hive表，比如有多个key都对应了超过1万条数据。
2. 然后将该RDD的每条数据都打上一个n以内的随机前缀。
3. 同时对另外一个正常的RDD进行扩容，将每条数据都扩容成n条数据，扩容出来的每条数据都依次打上一个 0~n的前缀。
4. 最后将两个处理后的RDD进行join即可。

方案实现原理：将原先一样的key通过附加随机前缀变成不一样的key，然后就可以将这些处理后的“不同key”分散到多个task中去处理，而不是让一个task处理大量的相同key。该方案与方案六的不同之处在于，上一种方案是尽量只对少数倾斜key对应的数据进行特殊处理，由于处理过程需要扩容RDD，因此上一种方案扩容RDD后对内存的占用并不大；而这一种方案是 针对有大量倾斜key的情况，没法将部分key拆分出来进行单独处理，因此只能对整个RDD进行数据扩容，对内存资源要求很高。

方案优点：对join类型的数据倾斜基本都可以处理，而且效果也相对比较显著，性能提升效果非常不错。

方案缺点：该方案更多的是缓解数据倾斜，而不是彻底避免数据倾斜。而且需要对整个RDD进行扩容，对内存资源要求很高。

方案实践经验：曾经开发一个数据需求的时候，发现一个join导致了数据倾斜。优化之前，作业的执行时间大约是60分钟左右；使用该方案优化之后，执行时间缩短到10分钟左右，性能提升了6倍。

八、 多种方案组合使用

在实践中发现，很多情况下，如果只是处理较为简单的数据倾斜场景，那么使用上述方案中的某一种基本就可以解决。但是如果要处理一个较为复杂的数据倾斜场景，那么可能需要将多种方案 组合起来使用。

比如说，针对出现了多个数据倾斜环节的Spark作业：

• 可以先运用解决方案一和二，预处理一部分数据，并过滤一部分数据来缓解；
• 其次可以对某些shuffle操作提升并行度，优化其性能；
• 最后还可以针对不同的聚合或join操作，选择一种方案来优化其性能。

说到底还是需要对这些方案的思路和原理都透彻理解之后，在实践中根据各种不同的情况，灵活运用多种方案，来解决自己的数据倾斜问题。