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 ) a
left 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 3G
stage 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 执行时间：11
stage 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_organization
select
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) gridNum
from dc_f_organization n
left join dc_d_property pt on n.orglevel = pt.id
GROUP BY n.YEARMONTH,n.ZZDATE,n.orgid,n.orgcode
WITH CUBE;
3、优化方案2：从业务逻辑上进行优化。（发现SQL逻辑中存在重复的计算）
-- ############################ 预处理：去除重复计算和减少CUBE的计算量 ############################
drop table if exists temp_dc_c_organization;
create table temp_dc_c_organization
as select
n.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 gridnum
from
dc_f_organization n
left join dc_d_property pt on n.orglevel = pt.id;
-- ############################ CUBE：节省的时间相当于预处理的时间。############################
create table dc_c_organization_02
as 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) gridnum
from temp_dc_c_organization as n
group by yearmonth, zzdate, orgid with cube;
