Hive和Spark SQL优化

Posted 2022-10-17 strongyoung88

优化步骤

• 问题定位，通过日志和代码分析，定位任务耗时长的原因

  • hive任务，对于hive任务，可以通过日志查看哪个application、以及是哪个阶段耗时较长，另外，可以通过stage编号，结合explain查看执行计划，可以得知耗时较长的application对应的是sql中的哪个操作，便于具体分析原因。如果是map阶段较长，很可能是并发不够，如果是reduce较长，且长时间留停在99%，则很可能发生了数据倾斜，如果是map和reduce的耗时都不长，但整个application的耗时却很长，则很有可能是大量小文件的读写造成的I/O耗时较长，当然，也不排除集群本身的一些因素，比如网络等问题。

  • spark任务，对于spark任务，可以在spark UI页面上查看作业的dag图，每个stage耗费的时长，每个stage启有的executor数量，每个任务耗费的时长，最长任务耗时，最短任务耗时，以及任务的中位数耗时等等，通过这些信息，我们可以判断出当前作业耗时较长的原因。

• 对症下药
  大多数情况，运行慢的原因很可能是并行度不够，或者数据处理发生了倾斜，找到任务耗时长的根源之后，根据不同的原因，作出相应的调优策略，例如，如果是并发不够，则适当提高并发，如果是数据发生倾斜，则考虑如何解决倾斜。

优化策略

• 剪枝
• 提高并行度
• join优化
• 避免生成太多小文件
• 数据倾斜
• 尽量减少shuffle

剪枝

• 合理利用分区
  对于全量计算的表，最好考虑拆分，减少数据输入量

• 合理利用分桶

  https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL+BucketedTables

• 谓词下推，能减少处理的数据，就尽量提前减少，保证数据准确性的前提下，尽量使用谓词下推操作

  注意：有些条件放在on后和where后结果不一样，需要留意

  	Join(inner join)		Left Outer Join		Right Outer Join		Full Outer Join
  	Left Table	Right Table	Left Table	Right Table	Left Table	Right Table	Left Table	Right Table
  Join Predicate	Pushed	Pushed	Not Pushed	Pushed	Pushed	Not Pushed	Not Pushed	Not Pushed
  Where Predicate	Pushed	Pushed	Pushed	Not Pushed	Not Pushed	Pushed	Not Pushed	Not Pushed

提高并行度

• 对于hive程序，可以调整提高map或reduce的任务的个数，从而一定程度上提高并行度
  具体调整方法可参考：Hive中Map任务和Reduce任务数计算原理

• 对于可以并发执行的，可以设置并发，让多个没有依赖关系的application并发执行，配置如下：
  set hive.exec.parallel=true

• 对于spark程序，可以适当调整提高executor的数量

Join优化

• 对于hive作业，有小表或比较小的表，尽量走mapjoin

  set hive.auto.convert.join=true; -- 版本0.11.0之后，默认是开启状态的，但猛犸时不时会把这个配置关闭，比较坑，所以最好还是手动配置一下
  set hive.mapjoin.smalltable.filesize = 25000000; -- 默认是25Mb开启mapjoin，对于稍微超过这大小的，可以适当调大，但不能太大
  

  另外，对于join操作中有子查询，且子查询中的某个表原始数据大于25M，但经过子查询之后，数据量小于25M的，可以采用临时表落地的方式，让此join走mapJoin，如果不落地，暂时无法走mapJoin，spark中，自适应框架中加入这个feature，自动判别，但目前还未上线。

• 对于spark作业，如果join时有小表的情况，尽量走broadcasthashjoin，该默认是开启的，小表大小默认10M，如果小表小于10M，但仍然没有走broadcasthashjoin，可以检查一下相关配置参数。

spark.sql.autoBroadcastJoinThreshold，默认值：10485760（10MB)，但要注意，只支持运行了ANALYZE TABLE <tableName> COMPUTE STATISTICS noscan命令的hive元数据表，所以，如果是刚刚创建的表想走broadcasthashjoin，还需要运行一下此条命令。

BroadcastHashJoin：spark 如果判断一张表存储空间小于 broadcast 阈值时（Spark 中使用参数 spark.sql.autoBroadcastJoinThreshold 来控制选择 BroadcastHashJoin 的阈值，默认是 10MB），就是把小表广播到 Executor， 然后把小表放在一个 hash 表中作为查找表，通过一个 map 操作就可以完成 join 操作了，避免了性能代码比较大的 shuffle 操作，不过要注意， BroadcastHashJoin 不支持 full outer join， 对于 right outer join， broadcast 左表，对于 left outer join，left semi join，left anti join ，broadcast 右表， 对于 inner join，那个表小就 broadcast 哪个。

• 小文件问题

  大量的小文件不仅会给namenode增加压力，也极易产生数据性能问题

  • 对于hive作业

  对于有产生很多小文件的任务，尤其是动态分区，可以通过以下配置，在任务结束自动合并小文件
  set hive.merge.mapfiles = true; – 开启map端合并小文件，默认开启，会另外运行一个只有map的applicaiton来合并小文件
  set hive.merge.mapredfiles = true; – 开启reduce端合并小文件
  set hive.merge.smallfiles.avgsize = 16000000 – 平均文件大小，默认16M，满足条件则自动合并，只有在开启以上两个开关才有效

  • 对于spark作业

  如果每个分区很小，但产生的文件很多，可以在sql最后加上以下，来合并同一个分区(ds)里的多个小文件：

  distribute by ds
  

• 存储格式
  尽量不要使用text文件存储，不仅存储量大，还可能引发bug，例如文本存储的时候，如查某个字段的值存在\\n换行符时，在数据读取时就会有异常。

• 倾斜问题

  如果是倾斜问题，可参考文章：

• 减少shuffle次数

  shuffle会造成数据在集群中传输，并且伴随着读和写，很影响任务的执行性能

  union与union all，union去重，有shuffle，union all不去重，无shuffle
  尽量使用union all，例如有以下sql:

  select c1, c2
  from (
  	select c1, c2 from t1
  	union
  	select c1, c2 from t2
  	union
  	select c1, c2 from t3
  ) t4
  

  可换成以下，将两次shuffle降低到一个shuffle

  select c1, c2
  from (
  	select c1, c2 from t1
  	union all
  	select c1, c2 from t2
  	union all
  	select c1, c2 from t3
  ) t4
  group by c1, c2