数据倾斜（spark中）_spark lateral view

一，什么是数据倾斜
数据倾斜即指在大数据计算任务中某个处理任务的进程（通常是一个JVM进程）被分配到的任务量过多，导致任务运行时间超长甚至最终失败，进而导致整个大任务超长时间运行或者失败。外部表现的话，在HiveSQL任务里看到map或者reduce的进度一直是99%持续数小时没有变化；在SparkSQL里则是某个stage里，正在运行的任务数量长时间是1或者2不变。总之如果任务进度信息一直在输出，但内容长时间没有任何变化的时候，大概率是出现数据倾斜了。有个特例需要注意，有时候大家会看到SparkSQL的任务信息也显示有1到2个任务在运行中，但进度信息不再刷新而表现为假死很久的时候，这通常是在进行最后阶段的文件操作，并不是数据倾斜（虽然这通常意味着小文件问题严重）。
再细分一下，倾斜可以分为以下四类：
1，读倾斜。即某个map（HiveSQL）或者task（SparkSQL）在读取数据阶段长期无法完成。这通常是因为文件分块过大或者此分块数据有异常。这种场景出现频率较小。
2，算倾斜。即在某个需要排序（如开窗函数或者非广播关联时）或者聚合操作的时候，同个一个key（通常是一个或者多个字段或者表达式的组合）的处理耗时过长。这通常是最多的情况，情况也较为复杂。
3，写倾斜。即某个操作需要输出大量的数据，比如超过几亿甚至几十亿行。主要出现在关联后数据膨胀及某些只能由一个task来操作（如limit）的情况。
4，文件操作倾斜。即数据生成在临时文件夹后，由于数量巨大，重命名和移动的操作非常耗时。这通常发生在动态分区导致小文件的情况。目前在国内和印度区域已经因为我们默认进行小文件合并而不再存在这个情况，新加坡还有（我们在推动解决）。

二，为什么会有数据倾斜
大数据计算依赖多种分布式系统，需要将所有的计算任务和数据经过一定的规则分发到集群中各个可用的机器和节点上去执行，最后可能还需要进行汇总到少数节点进行最后的聚合操作，以及数据写到HDFS/S3等分布式存储系统里以永储存。这个过程被设计来应对大多数情况，并不能应对所有的情况。它具有以下几个特点：
1，业务数据分布规律无法预知。比如系统无法不经过计算而提前知道某个表的某个字段的取值分布是否大致均匀。
2，计算结果数量无法预知。比如两表关联的结果对于某些key（关联的一个字段或者多个字段组合）的输出行数无法不经过计算而预知进而针对性处理；又比如对某个字段的值进行split操作或者explode等操作后产生的结果数量无法预知而进行针对性的应对。
3，某些操作只能由单一节点进行。一切需要维护一个全局状态的大多数操作，如排序，Limit，count distinct，全局聚合等，一般会安排到一个节点来执行。
上述三个主要特点导致单节点处理的数据量有概率出现巨量，造成了所谓的倾斜问题。当然，这些困难并不是不可解决的。随着时间的推移，越来越多的针对性的优化措施已逐渐出现，也许不久的将来业务同学不会再被倾斜问题烦恼

三，解决案例
1，事实表关联事实表数据膨胀
就是事实表关联事实表，其中有若干个key的输出达数十亿行，数据膨胀严重，造成数据计算和输出的倾斜
select a.userid,
b.pic_id
from id_userids a – xxID和userid的关系，一对多
left join id_pics b – xxID和图片ID的关系，一对多、
on a.options_id_modify=b.option_id;
这种事实表关联事实表的情况在非报表类的计算任务偶尔会遇到。平时我们解决数据倾斜主要是计算结果的过程涉及太多数据要处理导致慢，但通常输出的行数可能并不多，不存在写的困难，所以类似过滤异常数据或者广播关联等方法都不起作用。
这个问题的本质是一个task最多由一个进程来执行，而相同的key也必须在同一个task中处理，因此在无法改变这个机制的前提下，我们只有想办法减少一个task输出的行数。
那如何在不影响最终结果的前提下，减少单个task所需要处理数据行数呢？
其实网上也有许多建议，都是单独处理倾斜的key，通过加前缀后缀等方式打散key，再最后合并处理，但这样做法太麻烦了，不够优雅。我们要追求对业务同学更友好，代码更优雅的方式。
最后发现了collect_set/collect_list这个聚合函数，可以保证数据关系不丢失的前提下将数据收拢减少行数
最终代码如下

set spark.sql.adaptive.shuffle.targetPostShuffleInputSize=10000000;  
create table dc_tmp.mlwxh0328_t22 as 
select userid,
       pic_id,
       options_id_modify
  from (select /*+ repartition(1000) */
               options_id_modify,
               userids,
               pic_ids 
          from (select collect_set(userid) as userids,
                       options_id_modify    
                  from dc_tmp.mlwxh0328_t1   
                 where dayno='20210301'
                 group by options_id_modify,
                          ceil(rand()*100)
               )s1
          left join
               (select collect_list(pic_id) as pic_ids, 
                       option_id   
                  from cpc_tmp.temp_20210324_liuting_pic_optionid  
                 where option_id  is not null
                 group by option_id,
                          ceil(rand()*100)
               )s2
            on s1.options_id_modify=s2.option_id
       ) xh
       lateral view explode(userids) t1 as userid
       lateral view explode(pic_ids) t1 as pic_id 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

代码里的hint（repartition(1000)）的作用是考虑到经过collect_list聚合后的数据单行携带的数据经过一行变多行的展开操作后会膨胀很多倍，因此单个任务处理的数据量必须很小，才能保证处理速度够快。这个hint的作用是告诉系统将上一阶段关联后的结果分成1000份，交给下游处理；
group by语句里的ceil(rand()*N)作用是将一个key分成最多N行，这样可以限制最后按key关联后生成的行数的上限；
通过spark.sql.files.maxPartitionBytes参数控制单个任务处理的数据量，进一步拆分单个任务需要处理的数据。事实上如果第1点里文件足够小。

2，避免排序
有一些算法基础的同学都知道排序操作在软件领域是开销非常大的操作，目前大规模应用的几大排序算法的时间复杂度中最好的也是O(nlogn)，即随着数据量的增长而非线性的增长。这就是说，大规模数据量的排序往往意味着巨大的时间消耗。然而这在大数据SQL中却是常见的情况，从而引发倾斜。一旦有了排序的需求，什么优化参数都不好使了， 一般来说只有进行改写代码。幸运的是，在绝大多数大数据场景下，排序是不必要的，很多时候只是业务同学不解排序在大数据场景下的开销很大而信手写下了排序代码。下面介绍2个改写代码从而避免排序的案例。
1）用max函数替换排序。
需要对某个业务的埋点数据做一次样本展示，要在约1200亿行数据中，捞出约1万条数据。很简单的一个SQL如下
SQL的意思：希望取出上报数据里针对某个维度组合的一条内容较为丰富的样本数据，因此以某字段的size做为降序排序并取结果的第一条。
这个SQL当然跑失败了。我对partition by的字段集合（后续简称key）进行了统计，最大的key有137亿行，另外还有至少10个key的数据量超过20亿行。这样executor的内存加得再大都无法跑成功了。
这个问题的本质还是对大数据做了不必要的排序（大数据架构里对排序暂无非常高效的处理办法）。因此优化的思路还是想办法减少这种不必要排序。
既然用户只需要排序后的最大的一条，本质上不就是取某个key的最大值嘛。取出这个最大值，最后再跟源表进行关联，就可以取出最大值对应的那一条数据。
这里有个前提条件，要想在第二步关联回源表数据的时候干掉排序，我们只有走一条路：广播关联（如果走sort-meger关联，还是会避免不了sort步骤）。这就要求我们的小表（key-最大值）要足够小。通常这个条件都会满足的，因为如果不满足的话，说明key值非常多，非常稀疏，也不会产生倾斜的困境了。如开始就说明了，最后Key的去重数据量不到1万条，完全可以走广播关联。
我们使用了semi join，这在日常代码中比较少见。它的意思是，左表去匹配右表，如果一旦发现左表的某条数据的关联key在右表，便保留此条左表的数据，不再继续在右表里查找了。这样做有两个结果：1）速度更快；2）不会把右表的数据放到结果里）。它等价于 select * from left_table where key in (select key from right_table)。但大数据发展过程中一度不支持in的用法（现在部分支持了），因此有这种语法，从效率上看，一般认为这样更高效。
因为能匹配到最大值的数据可能有许多条，所以对最后结果再做一次row_number的开窗并取其中一条即可。这个时候由于size(xxxx)的值都是一样的，因此任意取一条均符合业务需求。
2）用分位函数替换排序
问题点：代码是想做一个全局排序，然后使用其序号所在位置来进行分类打标。上述代码在排序数据小于5亿5千万行的情况勉强能运行出结果。但在某一天数据量到了5亿5千万行后就死活跑不出来，加了reducer的内存到10G也不行
新思路：虽然可能还有一些参数能调整，停止了研究，把方向转为干掉全局排序。在和一位前辈沟通的时候，突然意识到，既然业务是想做一个分档，本质上并不需要具体的排序号，所以理论上完全的排序是可以省掉的。于是自然想到了分位数函数，立马想到了新方案。分位函数计算出数据必须大于或者等于某个值才能处于整个数据排序的某个位置。详情请大家自行搜索。

****对于排序函数的使用尽量避免