13.108.Spark 优化、Spark优化与hive的区别、SparkSQL启动参数调优、四川任务优化实践：执行效率提升50%以上

13.108.Spark 优化
1.1.25.Spark优化与hive的区别
1.1.26.SparkSQL启动参数调优
1.1.27.四川任务优化实践：执行效率提升50%以上

13.108.Spark 优化：

1.1.25.Spark优化与hive的区别

先理解spark与mapreduce的本质区别，算子之间（map和reduce之间多了依赖关系判断，即宽依赖和窄依赖。）
优化的思路和hive基本一致，比较大的区别就是mapreduce算子之间都需要落磁盘，而spark只有宽依赖才需要落磁盘，窄依赖不落磁盘。

1.1.26.SparkSQL启动参数调优

1)先对比结果：executors优化
Hive执行了30分钟（1800秒）的sql，没有优化过的SparkSQL执行需要，
最少化的Executor执行需要640秒（提高了Executor的并行度，牺牲了HDFS的吞吐量：5个core最合适），
最大化的Executor 281.634秒（最大限度的利用HDFS的吞吐量，牺牲Executor的并行度），
优化取中间值，253.189秒。

方案1：最少化 Fat executors

---------------------------------	Fat executors	--------------------------------------------------------------------------------
./spark-sql --master yarn \	# Fat executors (每个节点一个Executor)【优势：最佳吞吐量】
--num-executors 3 \			# 集群中的节点的数目 = 3
--executor-memory 30G \	# 每个节点的内存/每个节点的executor数目 = 30GB/1 = 30GB
--executor-cores 16 \		# 每个executor独占节点中所有的cores = 节点中的core的数目 = 16
--driver-memory 1G			# AM大约需要1024MB的内存和一个Executor
耗时：Time taken: 640 seconds

方案2：最大化Tiny executors

---------------------------------	Tiny executors	--------------------------------------------------------------------------------
./spark-sql --master yarn \	# Tiny executors [每个Executor一个Core]【优势：并行性】
--num-executors 48 \		# 集群中的core的总数 = 每个节点的core数目 * 集群中的节点数 = 16*3
--executor-memory 1.6G \	# 每个节点的内存/每个节点的executor数目 = 30GB/16 = 1.875GB
--executor-cores 1 \			# 每个executor一个core
--driver-memory 1G			# AM大约需要1024MB的内存和一个Executor
耗时：Time taken: 281.634 seconds
executor并发度只有45，task的并发度，1个executor 50左右，总数 18382

方案3：折中方案

---------------------------------	Balance between Fat (vs) Tiny	--------------------------------------------------------------------------------
./spark-sql --master yarn \	# Balance between Fat (vs) Tiny
--num-executors 8 \			# (16-1)*3/5 = 9 留一个executor给ApplicationManager => --num-executors = 9-1 = 8# 每个节点的executor的数目 = 9 / 3 = 3
--executor-memory 10G \	# 每个executor的内存 = 30GB / 3 = 10GB【默认分配的是8G，需要修改配置文件支持到10G。】# 计算堆开销 = 7% * 10GB = 0.7GB。因此，实际的 --executor-memory = 10 - 0.7 = 9.3GB
--executor-cores 5 \			# 给每个executor分配5个core，保证良好的HDFS吞吐。# 每个节点留一个core给Hadoop/Yarn守护进程 => 每个节点可用的core的数目= 16 - 1
--driver-memory 1G			
耗时：Time taken: 253 seconds

Task并行度优化
1.调整 Executors 下 每个stage的默认task数量，即设置Task 的并发度：

【当集群数量比较大时】
很多人常犯的一个错误就是不去设置这个参数，那么此时就会导致Spark自己根据底层HDFS的block数量来设置task的数量，
！【默认是一个HDFS block对应一个task（如果不设置那么可以通过第三种方案来优化！）】。
通常来说，Spark默认设置的数量是偏少的（比如就几十个task），
如果task数量偏少的话，就会导致你前面设置好的Executor的参数都前功尽弃。
试想一下，无论你的Executor进程有多少个，内存和CPU有多大，但是task只有1个或者10个，
那么90%的Executor进程可能根本就没有task执行，也就是白白浪费了资源！
因此Spark官网建议的设置原则是，设置该参数为num-executors * executor-cores的2~3倍较为合适，
比如Executor的总CPU core数量为300个，那么设置1000个task是可以的，此时可以充分地利用Spark集群的资源。

30 G 16 core

/home/admin/bigdata/spark-2.2.0-bin-hadoop2.6/bin/spark-sql \
--master yarn \
--num-executors 16 \
--executor-memory 1G \
--executor-cores 10 \
--driver-memory 1G \
--conf spark.default.parallelism=450 \
--conf spark.storage.memoryFraction=0.5 \
--conf spark.shuffle.memoryFraction=0.3	

1.1.27.四川任务优化实践：执行效率提升50%以上

一、四川的信息
账号：xxxxxx 密码： xxxxxxxx

一、事实表优化
1、**优化结果： 20 分钟左右，优化完成后 5 分钟左右。**数据量：5.8亿

2、原SQL：（spark不一定快）

drop table if exists dc_f_organization;
create table if not exists dc_f_organization (orgid int,orgcode string,yearmonth string ,zzdate string,orgname string,orglevel int,id int,orgtagging int, createdate timestamp
);insert into dc_f_organization
select a.orgid, .orgcode, a.yearmonth, a.zzdate, n.orgname, n.orglevel, n.id, n.orgtagging, n.createdate
from ( select o.id orgid, o.orgcode, d.yearmonth, d.zzdate from dc_d_organization o, dc_d_wddate ) aleft join dc_d_organization n on to_date(n.createdate)=a.zzdate and n.orgcode = a.orgcode;

3、优化方案：
– ############################## HIVE 执行：增加 block 的数量，提高Spark的并发度（当前任务文件比较小，设置了26；一般参考数量：300左右；） #################################
– (1) 单独执行笛卡尔积，
– 先拆分文件：（改用hive，拆分文件，增加并行度）
– 【耗时：101.586 seconds；结果文件数量 26】
– 检查文件块数量：hadoop fs -ls /user/hive/warehouse/test.db/dc_d_org_date 26 个block

set mapreduce.map.memory.mb=1024;
set mapred.max.split.size=524288;
set mapred.min.split.size.per.node=524288;
set mapred.min.split.size.per.rack=524288;
drop table if exists dc_d_org_date;
create table dc_d_org_date as select o.id orgid,o.orgcode,d.yearmonth,d.zzdate from dc_d_organization o CROSS JOIN dc_d_wddate d;

-- ##############################	SPARK 执行；参数：spark-sql --master yarn --num-executors 100 --executor-memory 5G --executor-cores 3 --driver-memory 3G	#################################
-- (2)【Spark：Time taken: 115.78 seconds；】
set spark.shuffle.consolidateFiles=true;
drop table if exists dc_f_organization;
create table if not exists dc_f_organization
(orgid int,orgcode string,YEARMONTH string ,ZZDATE string,ORGNAME string,orglevel int,id int,ORGTAGGING int, createdate timestamp);insert into dc_f_organization
select a.orgid,a.orgcode,a.YEARMONTH,a.ZZDATE,n.ORGNAME,n.orglevel,n.id,n.ORGTAGGING,n.createdate
from dc_d_org_date a
left join DC_D_ORGANIZATION n on to_date(n.CREATEDATE)=a.ZZDATE and n.orgcode = a.orgcode;

– ############################## 持续优化方向：将上述两者合并到一起在 spark 中执行 ##############################
问题：可能是因为文件太小，spark 分区命令没有生效。set spark.sql.shuffle.partitions=300;
注意：SPARK中笛卡尔积需要改成 CROSS JOIN，否则语法报错。

二、优化CUBE表

  1、优化结果：原来1小时左右，优化后26分钟。总结：shuffle时间：16分钟，数据量	35.2亿任务含有宽依赖（group）被分成2个stage✔采用方案 1：改用spark执行。提高并行度。执行参数：spark-sql --master yarn --num-executors 100 --executor-memory 5G --executor-cores 3 --driver-memory 3Gstage 1 执行时间：11（partitions=300）stage 2 执行时间：15（partitions=200）设置分区数量，默认是200：set spark.sql.shuffle.partitions=300;（理论上可以提高 stage 2 30%的速度，实际运行的时候可能会丢失executor，运行不稳定，不建议设置。）（原因可能是设置了虚拟核心数量。）方案 2：将case when的操作独立出一张表，去除部分重复扫描计算，减少cube阶段的计算量。抽取的时间增加了2分钟，节省的 shuffle 时间也是2分钟。没有意义。预处理时间：2-3分钟stage 1 执行时间：11stage 2 执行时间：13（节省的时间也是2-3分钟）方案 3：提高 shuffle 使用内存的占比 设置为60%执行参数：spark-sql --master yarn --num-executors 100 --executor-memory 5G --executor-cores 3 --driver-memory 3G --conf spark.storage.memoryFraction=0.3 --conf spark.shuffle.memoryFraction=0.5执行结果：效果不明显，多次执行时间也不太一致。方案 4：减少CUBE的维度数量， orgid 和 orgcode是一对一关系，可以去掉1个维度，计算完成之后再join执行结果：join 消耗的时间更久。2、采用的方案1：SPARK执行-- 执行参数 spark-sql --master yarn --num-executors 100 --executor-memory 5G --executor-cores 3 --driver-memory 3G-- set spark.sql.shuffle.partitions=300;drop table  if  exists dc_c_organization;create table if not exists dc_c_organization(YEARMONTH string,ZZDATE string,orgid int ,orgcode string,total int,provinceNum int,cityNum int,districtNum int, newDistrictNum int,townNum int,streetNum int,otherNum int,communityNum int,villageNum int,gridNum int);-- 如果用 hive 执行可以开启 combiner，map端先预聚合，减少reduce端的数据量和计算量，减少磁盘的IO和网络传输时间。-- set hive.map.aggr = true;-- set hive.groupby.mapaggr.checkinterval = 10000;-- ##############################	SPARK	#################################-- set spark.sql.shuffle.partitions=300;insert into dc_c_organizationselect  n.YEARMONTH,n.ZZDATE,n.orgid,n.orgcode,count(n.id) total,nvl(SUM(case when pt.displayname='省' then 1  else 0 end),0) AS provinceNum,nvl(SUM(case when pt.displayname='市' then 1  else 0 end),0) as cityNum,nvl(SUM(case when pt.displayname='县（区）' then 1  else 0 end),0) AS districtNum,(nvl(SUM(case when pt.displayname='县（区）'  then 1  else 0 end),0) -nvl(SUM(case when pt.displayname='县（区）' AND n.ORGTAGGING= 31 then 1  else 0 end),0)) as newDistrictNum,nvl(SUM(case when  ((n.ORGNAME LIKE '%乡%' OR n.ORGNAME LIKE '%镇%' OR n.ORGNAME LIKE '%乡镇%')) AND pt.displayname='乡镇（街道）' then 1  else 0 end),0) townNum,nvl(SUM(case when (n.ORGNAME LIKE '%街道%') AND pt.displayname='乡镇（街道）' then 1  else 0 end),0) streetNum,(nvl(SUM(case when pt.displayname='乡镇（街道）'then 1  else 0 end),0)-nvl(SUM(case when ((n.ORGNAME LIKE '%乡%' OR n.ORGNAME LIKE '%镇%' OR n.ORGNAME LIKE '%乡镇%') ) AND pt.displayname='乡镇（街道）' then 1  else 0 end),0)-nvl(SUM(case when (n.ORGNAME LIKE '%街道%' )  AND pt.displayname='乡镇（街道）' then 1  else 0 end),0)) otherNum,(nvl(SUM(case when pt.displayname='村（社区）' then 1  else 0 end),0)-nvl(SUM(case when ((n.ORGNAME LIKE '%村' OR n.ORGNAME LIKE '%村民委员会' OR n.ORGNAME LIKE '%农村工作中心站' OR n.ORGNAME LIKE '%村委会')) AND pt.displayname='村（社区）' then 1  else 0 end),0)) communityNum,nvl(SUM(case when ((n.ORGNAME LIKE '%村' OR n.ORGNAME LIKE '%村民委员会' OR n.ORGNAME LIKE '%农村工作中心站' OR n.ORGNAME LIKE '%村委会')) AND pt.displayname='村（社区）' then 1  else 0 end),0) villageNum,nvl(SUM(case when pt.displayname='片组片格'then 1  else 0 end),0) gridNumfrom dc_f_organization nleft join dc_d_property pt on n.orglevel = pt.idGROUP BY n.YEARMONTH,n.ZZDATE,n.orgid,n.orgcodeWITH CUBE;3、优化方案2：从业务逻辑上进行优化。（发现SQL逻辑中存在重复的计算）-- ############################	预处理：去除重复计算和减少CUBE的计算量	############################drop table if exists temp_dc_c_organization;create table temp_dc_c_organizationas selectn.yearmonth,n.zzdate,n.orgid,n.orgcode,n.id as id,case when pt.displayname='省' then 1  else 0 end as provincenum,case when pt.displayname='市' then 1  else 0 end as citynum,case when pt.displayname='县（区）' then 1  else 0 end as districtnum,case when pt.displayname='县（区）' and n.orgtagging= 31 then 1  else 0 end as old_districtnum,
【重复1】	case when ((n.orgname like '%乡%' or n.orgname like '%镇%' or n.orgname like '%乡镇%')) and pt.displayname='乡镇（街道）' then 1  else 0 end townnum,
【重复2】	case when (n.orgname like '%街道%') and pt.displayname='乡镇（街道）' then 1  else 0 end streetnum,case when pt.displayname='乡镇（街道）'then 1  else 0 end as total_streetnum_01,
【重复1】	case when ((n.orgname like '%乡%' or n.orgname like '%镇%' or n.orgname like '%乡镇%')) and pt.displayname='乡镇（街道）' then 1  else 0 end as total_streetnum_02,
【重复2】	case when (n.orgname like '%街道%') and pt.displayname='乡镇（街道）' then 1  else 0 end as total_streetnum_03,case when pt.displayname='村（社区）' then 1  else 0 end as communitynum_01,
【重复3】	case when ((n.orgname like '%村' or n.orgname like '%村民委员会' or n.orgname like '%农村工作中心站' or n.orgname like '%村委会')) and pt.displayname='村（社区）' then 1  else 0 end as communitynum_02,
【重复3】	case when ((n.orgname like '%村' or n.orgname like '%村民委员会' or n.orgname like '%农村工作中心站' or n.orgname like '%村委会')) and pt.displayname='村（社区）' then 1  else 0 end villagenum,case when pt.displayname='片组片格'then 1  else 0 end gridnumfromdc_f_organization nleft join dc_d_property pt on n.orglevel = pt.id;-- ############################	CUBE：节省的时间相当于预处理的时间。############################create table dc_c_organization_02as select  yearmonth,zzdate,orgid,count(id) total,sum(provincenum) as provincenum,sum(citynum) as citynum,sum(districtnum) as districtnum,sum(districtnum)-sum(old_districtnum) as newdistrictnum,sum(townnum) townnum,sum(streetnum) streetnum,sum(total_streetnum_01)-sum(townnum)-sum(streetnum) othernum,sum(communitynum_01)-sum(villagenum) communitynum,sum(villagenum) villagenum,sum(gridnum) gridnumfrom temp_dc_c_organization as ngroup by yearmonth, zzdate, orgid with cube;