Hive调优，

hive优化(数据倾斜)

问题描述：

1.倾斜：
	小文件 join 大文件
	内容倾斜
2.JOBS多
	join
	union
	sub_query
3.count(distinct)
4.Task过多

解决方案：整体考虑

1.建模
		分层=>轻量聚合
		分区=>避免交换
		压缩=>减少体量	
2.配置优化
		合理减少job,并行无依赖job,增加jvm重用
		合理控制mapper和reducer的数量
			核数
			内存
			单节点并行jvm数
			数据的规模
			hive内部配置
		开启分区倾斜自动处理机制
		合并小文件
	语句优化
		列剪裁			不用*进行全列扫描，只列出需要的列
							set hive.optimize.cp=true;
	分区剪裁		不指定分区进行全表扫描
						set hive.optimize.pruner=true;
	谓词下推PPD		条件优先放在子查询中执行,减少下游处理的数据量
	sort by 替 order by		全局排序导致一个Reducer
										distribute by + sort by
	group by 替 distinct	distinct类似于order by
									group by + set mapred.reduce.tasks=N; 并发
	build table 前置		map join
	probe table 后置		#小表前置，方便自动mapjoin
									set hive.cbo.enable=default true 
									set hive.auto.convert.join=true;
									set hive.map.aggr=true;
									set hive.mapjoin.smalltable.filesize=25m; 小于该值自动mapjoin
									set hive.mapjoin.cache.numrows=25000; 缓存小表多少行记录到内存
									set hive.optimize.bucketmapjoin=true; 分通表专用
									#set hive.compute.query.using.stats=default true;
									#set hive.stats.fetch.column.stats=default false;
									#set hive.stats.fetch.partition.stats=default undefined;
									
			join是key相同			合并job
			倾斜均衡配置			#join的键对应的记录条数超过这个值则会进行分拆，值根据具体数据量设置
									set hive.skewjoin.key=default 100000;
									#如果是join过程出现倾斜应该设置为true
									set hive.optimize.skewjoin=default false;
									#优化第二job的mapper数量
									set hive.skewjoin.mapjoin.map.tasks=10000;
									
	#配置
	#hadoop
	{
		#memory
			free/top
						total        used       free      shared  buff/cache   available
			Mem:        7990132      335388     7493208   11880   161536       7431068
			Swap:       8257532           0     8257532
			
			#虚拟机安装后将所有大数据环境安装，查看虚拟机的资源剩余5.2G
			#预留20%内存防止系统运行卡阻或OOM
			#剩余：5.2-1.6=3.6
			#container(内存从1G自动递增1G) 1~4
			yarn.nodemanager.resource.memory-mb=4					#default 1.5G
			
		#core
			yarn.nodemanager.resource.pcores-vcores-multiple=2		#default undefined
			yarn.nodemanager.resource.cpu-vcores=4					#default -1
			yarn.scheduler.minimum-allocation-vcores=1 				#default 1
			#官方和cloudrea建议不超过5个
			yarn.scheduler.maxiimum-allocation-vcores=4				#default undefined
	}
	#hive
	{
		<!--Container内存上限,由nodemanager读取并控制,实际使用超出时会被nodemanager kill Connection reset by peer-->
		<property>
	　　    <name>mapreduce.map.memory.mb</name>
	　　    <value>1024</value>
		</property>
		<!--Container启动jvm时,传递给jvm的参数,实际使用超出时会抛OutofMemory异常-->
		<property>
	　　    <name>mapreduce.map.java.opts</name>
	　　    <value>-Xmx1024M</value>
		</property>
		<property>
	　　    <name>mapreduce.map.java.opts</name>
	　　    <value>-Xms128M</value>
		</property>
		<property>
	　　    <name>mapreduce.reduce.memory.mb</name>
	　　    <value>2048</value>
		</property>
		<property>
	　　    <name>mapreduce.reduce.java.opts</name>
	　　    <value>-Xmx2048M</value>
		</property>
		<property>
	　　    <name>mapreduce.map.java.opts</name>
	　　    <value>-Xms256M</value>
		</property>
	}

	# 控制Mapper数量
	Mapper数量=input size / math.max(minsize,math.min(blocksize,maxsize))
	set mapred.map.tasks=default 2
	
	#Mapper合并：输入合并
	#每个Map最大输入大小，决定合并后的文件数
	set mapred.max.split.size=256000000;
	#一个节点上split的至少的大小 ，决定了多个data node上的文件是否需要合并
	set mapred.min.split.size.per.node=100000000;
	#一个交换机下split的至少的大小，决定了多个交换机上的文件是否需要合并
	set mapred.min.split.size.per.rack=100000000;
	#执行Map前进行小文件合并
	set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
	
	#Reducer合并：输出合并
	#在map-only job后合并文件，默认true
	set hive.merge.mapfiles 
	#在map-reduce job后合并文件，默认false
	set hive.merge.mapredfiles
	#合并后每个文件的大小，默认256000000
	set hive.merge.size.per.task 
	#平均文件大小，是决定是否执行合并操作的阈值，默认16000000
	set hive.merge.smallfiles.avgsize
	
	# 控制Reducer数量
	set mapred.reduce.tasks=8;
	#不设置reducer数量，hive自动计算
	set hive.exec.reducers.bytes.per.reducer=default 0.25G
	set hive.exec.reducers.max=default 1009		#如果input bytes size / bytes per reducer >max? max : insize/bytesPerReducer
	
	# 启动本地模式涉及到三个参数:
	# 打开hive自动判断是否启动本地模式的开关
	set hive.exec.mode.local.auto=default false;
	# map任务数最大值，不启用本地模式的task最大个数
	set hive.exec.mode.local.auto.input.files.max=default 4
	# map输入文件最大大小，不启动本地模式的最大输入文件大小
	set hive.exec.mode.local.auto.inputbytes.max=default 128M;
	
	# 开启任务并行执行
	# 同一个sql允许并行任务的最大线程数
	set hive.exec.parallel=true;				#default false
	set hive.exec.parallel.thread.number=8;		#default 8
	set hive.exec.reducers.max=8;				#default 1009
	set mapred.reduce.tasks=8;					#default -1
	
	<property>
		<name>hive.exec.parallel</name>
		<value>true</value>
	</property>
	<property>
		<name>hive.exec.parallel.thread.number</name>
		<value>16</value>
	</property>
	
	#虚拟机重用
	set mapred.job.reuse.jvm.num.tasks=10;多小文件或多任务，且数据不倾斜
	因为小任务端的mapper或reducer会一直占用task插槽
	
	#set hive.groupby.skewindata=true;	default false
	#set hive.groupby.mapaggr.checkinterval=100000; map端聚合的最大行数,超出则分解


	
	#拉链表 
	new_data_table & history_data_table
	增量覆盖导入 new_data_table
	insert overwrite table dim_date_100w
	-- 旧的改变了的数据
	select t2.date_key,t2.day,t2.month,t2.month_name,t2.year,t2.year_month,t2.week_of_year,t2.week,t2.week_name,1001 as quarter
	from dim_date_100w t1
	join dim_date_1w t2 on t1.date_key=t2.date_key
	-- 旧的不变的数据
	union all
	select t1.*
	from dim_date_100w t1
	left join dim_date_1w t2 on t1.date_key=t2.date_key
	where t2.date_key is null
	-- 新增的数据
	union all
	select t1.*
	from dim_date_1w t1
	left join dim_date_100w t2 on t1.date_key=t2.date_key
	where t2.date_key is null
	#hive 2之后新增批量更新功能 ：代替拉链表
	set hive.support.concurrency = true;
	set hive.enforce.bucketing = true;
	set hive.exec.dynamic.partition.mode = nonstrict;
	set hive.txn.manager = org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
	set hive.compactor.initiator.on = true;
	set hive.compactor.worker.threads = 1;
	set hive.auto.convert.join=false;
	set hive.merge.cardinality.check=false; #目标表中出现重复匹配时要设置该参数才行
	
	MERGE INTO dim_date_100w AS T USING dim_date_1w AS S
	ON t.date_key=s.date_key
	WHEN MATCHED THEN UPDATE SET quarter=1001
	WHEN NOT MATCHED THEN INSERT VALUES(S.date_key,S.day,S.month,S.month_name,S.year,S.year_month,S.week_of_year,S.week,S.week_name,S.quarter);