map优化
优化并发个数
-
减少map数，合并小文件
- set mapred.max.split.size=100000000;单位b,代表一个map能处理多大的数据量
- set mapred.min.split.size.per.node=100000000;一个节点最少要处理多少数据
- set mapred.min.split.size.per.rack=100000000;
- set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
-
增加map数
- set mapred.map.tasks=10;
block大小会影响并发度
- set dfs.block.size = 128
相当于开启Combiner功能
- hive.map.aggr=true
reduce优化
优化并发个数
- hive.exec.reducers.bytes.per.reduce;reduce任务处理的数据量
- reduce个数计算公式
reduce个数=inputFileSize / bytes per reduce 就是文件大小除以一个reduce能处理的字节大小 - 调整reduce的个数
set mapred.reduce.tasks=10
MapReduce出现痛点
只有一个reduce情况
- 没有group by
- order by
- 发生在reduce阶段
- 全局排序，因此只有一个reduce，当输入数据规模较大时，计算时间消耗严重
- 建议用distribute by 和 sort by来替代
- sort by
- 发生在map阶段
- 不是全局排序，如果用sort by排序，并且设置多个reduce，每个reduce输出是有序的，但是不保证全局排序
- distribute by
- 控制map端的数据如何拆分给reduce，可控制分区文件的个数
- cluster by
- 相当于distribute by和sort by的结合
- 只默认升序排序
示例
笛卡尔积
- 使用join的时候，尽量有效的使用on条件
如何加快查询速度
精华总结
-
横向尽可能多并发
一个节点的并发越多越好，前提是性能的保证 -
纵向尽可能少依赖
一个MapReduce能完成的任务，就不要选择多个任务来完成，控制任务数
1.分区partition
- where中的分区条件，会提前生效，不必特意做子查询，直接join和groupby
2.Map Join
- 指定表是小表，内存处理，通常不超过1个G或者50W记录
/*+MAPJOIN(tablelist)*/
3.union all
- 先把两张表union all，然后再做join或者group by，可以减少MapReduce的数量
- union和union all的区别：union相当于记录合并，union不合并，后者性能更优
4.multi-insert & multi group by
- 从一份基础表中按照不同的维度，一次组合出不同的数据
5.Automatic merge
- 为了多个小文件合并，当文件大小比阈值小时，hive会启动一个MapReduce进行合并
- 将数据量大的节点的数据中途随机分发进行预处理数据压缩，最后再合并数据
- 配置
- hive.merge.mapfiles=true 是否合并map输出文件,默认true
- hive.merge.mapredfiles=false是否合并reduce输出文件,默认false
- hive.merge.size.per.task=25610001000 合并文件的大小
6.Multi-Count Distinct
- 与精华总结相违背
- 一个MapReduce拆成多个的目的是为了降低数据倾斜的压力，出现数据倾斜时，帮忙完成一定的负载均衡
- 必须设置的参数：set hive.groupby.skewindata=true
- 示例
select dt,count(distinct uniq_id),count(distinct ip) from logs where dt = xxxx group by dt
7.并行执行
- set hive.exec.parallel=true
- 同步执行hive的多个阶段，hive在执行过程，将一个查询转化成一个或者多个阶段，某个特定的job可能包含众多的阶段，而这些阶段可能并非完全相互依赖的，也就是说可以并行执行的，这样可能使得整个job的执行时间缩短
如何加快join操作
- 语句优化
- 多表连接：如果join中多个表的join key是同一个，则join会转化为单个mr任务
- 表的连接顺序：指定大小表，Hive默认把左表数据放到缓存中，右边的表的数据做流数据
/*+STREAMTABLE(a)*/指定该表是大表
- 如果避免join过程中出现大量结果，尽可能在on中完成所有条件判断（少用where）
数据倾斜问题
业务场景
- 操作
- join
- group by
- count distinct
- 原因
- key分布不均
- 人为的建表疏忽
- 业务数据特点
- 症状
- 任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或者几个）reduce子任务未完成
- 查看未完成的子任务，可以看到本地读写数据量积累非常大，通常超过10GB可以认定为发生数据倾斜
- 倾斜度
- 平均记录数超过50w且最大记录数是超过平均数的4倍
- 最长时长比平均时长超过4分钟，且最大时长超过平均时长的2倍
- 万能方法
- hive.groupby.skewindata=true
- 把一个任务拆分成多个任务
大小表关联
- 原因
Hive在进行join时，按照join的key进行分发，而在join左边的表的数据会首先读入内存，如果左边表的key相对分散，读入内存的数据会比较小，join任务执行会比较快；而如果左边的表key比较集中，而这张表的数据量很大，那么数据倾斜就会比较严重，而如果这张表是小表，则还是应该把这张表放在join左边 - 思路
- 将key相对分散并且数据量小的表放在join的左边，这样可以有效减少内存溢出错误发生的几率
- 使用map join让小的维度表先进内存
- 方法
- small table join big table
大大表关联
- 原因
日志中有一部分的userid是空或者是0的情况，导致在用userid进行hash分桶的时候，会将日志中userid为0或者空的数据分到一起，导致了过大的斜率 - 思路
把空值的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果 - 方法
on case when(x.uid = '-' or x.uid = '0' or x.uid is null)then concat('dp_hive_search',rand()) else x.uid end =f.userid
聚合时存在大量特殊值
- 原因
做count distinct时，该字段存在大量值为null或空的记录 - 思路
- count distinct时，将值为空的情况单独处理，如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1
- 如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union
-
方法
0019b4d3-89e8-4387-a166-2a36e0ad71fb-2500378.jpg
空间换时间
-
案例
bb0a6592-8819-4b4f-8175-5d92d4a88347-2500378.jpg - 问题
同一个reduce上进行distinct操作时压力很大 -
方法
a7fe3baa-6b50-498f-b989-5de260d9755c-2500378.jpg
