文章目录
一、Hadoop 框架计算特性
二、优化常用手段
三、排序选择
四、怎样做笛卡尔积
五、怎样写 in/exists 语句
六、设置合理的 maptask 数量
七、小文件合并
八、设置合理的 reduceTask 的数量
九、合并 MapReduce 操作
十、合理利用分桶：Bucketing 和 Sampling
十一、合理利用分区：Partition
十二、Join 优化
十三、Group By 优化
1、Map 端部分聚合
2、使用 Group By 有数据倾斜的时候进行负载均衡
十四、合理利用文件存储格式
十五、本地模式执行 MapReduce
十六、并行化处理
十七、设置压缩存储
1、压缩的原因
2、常用压缩方法对比
3、压缩方式的选择
4、压缩使用
一、Hadoop 框架计算特性
数据量大不是问题，数据倾斜是个问题
jobs 数比较多的作业运行效率相对比较低，比如即使有几百行的表，如果多次关联多次汇总，产生十几个 jobs，耗时很长。原因是 map reduce 作业初始化的时间是比较长的
sum、count、max、min 等 UDAF，不怕数据倾斜问题，hadoop 在 map 端的汇总合并优化，这使得数据倾斜不成问题
count(distinct userid)，在数据量大的情况下，效率较低，如果是多 count(distinct userid,month)效率更低，因为 count(distinct)是按 group by 字段分组，按 distinct 字段排序， 一般这种分布方式是很容易引起倾斜的，比如 PV 数据，淘宝一天 30 亿的 pv，如果按性别分组，分配 2 个 reduce，每个 reduce 期望处理 15 亿数据，但现实必定是男少女多
二、优化常用手段
好的模型设计事半功倍
解决数据倾斜问题
减少 job 数
设置合理的 MapReduce 的 Task 数，能有效提升性能。(比如，10w+级别的计算，用 160个 reduce，那是相当的浪费，1 个足够)
了解数据分布，自己动手解决数据倾斜问题是个不错的选择。这是通用的算法优化，但 算法优化有时不能适应特定业务背景，开发人员了解业务，了解数据，可以通过业务逻辑精 确有效的解决数据倾斜问题
数据量较大的情况下，慎用 count(distinct)，group by 容易产生倾斜问题
对小文件进行合并，是行之有效的提高调度效率的方法，假如所有的作业设置合理的文 件数，对云梯的整体调度效率也会产生积极的正向影响
优化时把握整体，单个作业最优不如整体最优
三、排序选择
cluster by：对同一字段分桶并排序，不能和 sort by 连用
distribute by + sort by：分桶，保证同一字段值只存在一个结果文件当中，结合 sort by 保证 每个 reduceTask 结果有序
sort by：单机排序，单个 reduce 结果有序
order by：全局排序，缺陷是只能使用一个 reduce
一定要区分这四种排序的使用方式和适用场景!
四、怎样做笛卡尔积
当 Hive 设定为严格模式（hive.mapred.mode=strict）时，不允许在 HQL 语句中出现笛卡尔积， 这实际说明了 Hive 对笛卡尔积支持较弱。因为找不到 Join key，Hive 只能使用 1 个 reducer 来完成笛卡尔积。
当然也可以使用 limit 的办法来减少某个表参与 join 的数据量，但对于需要笛卡尔积语义的 需求来说，经常是一个大表和一个小表的 Join 操作，结果仍然很大（以至于无法用单机处 理），这时 MapJoin才是最好的解决办法。MapJoin，顾名思义，会在 Map 端完成 Join 操作。 这需要将 Join 操作的一个或多个表完全读入内存。
PS：MapJoin 在子查询中可能出现未知 BUG。在大表和小表做笛卡尔积时，规避笛卡尔积的 方法是，给 Join 添加一个 Join key。
原理很简单：将小表扩充一列 join key，并将小表的条 目复制数倍，join key 各不相同；将大表扩充一列 join key 为随机数。
精髓就在于复制几倍，最后就有几个 reduce 来做，而且大表的数据是前面小表扩张 key 值 范围里面随机出来的，所以复制了几倍 n，就相当于这个随机范围就有多大 n，那么相应的， 大表的数据就被随机的分为了 n 份。并且最后处理所用的 reduce 数量也是 n，而且也不会 出现数据倾斜。
五、怎样写 in/exists 语句
虽然经过测验，hive1.2.1 也支持 in/exists 操作，但还是推荐使用 hive 的一个高效替代方案：left semi join
比如说：
select a.id, a.name from a where a.id in (select b.id from b); select a.id, a.name from a where exists (select id from b where a.id = b.id);
应该转换成：
select a.id, a.name from a left semi join b on a.id = b.id;
六、设置合理的 maptask 数量
Map 数过大
- Map 阶段输出文件太小，产生大量小文件
- 初始化和创建 Map 的开销很大
Map 数太小
- 文件处理或查询并发度小，Job 执行时间过长
- 大量作业时，容易堵塞集群
在 MapReduce 的编程案例中，我们得知，一个MR Job的 MapTask 数量是由输入分片 InputSplit 决定的。而输入分片是由 FileInputFormat.getSplit()决定的。一个输入分片对应一个 MapTask， 而输入分片是由三个参数决定的：
输入分片大小的计算是这么计算出来的：
long splitSize = Math.max(minSize, Math.min(maxSize, blockSize))
默认情况下，输入分片大小和 HDFS 集群默认数据块大小一致，也就是默认一个数据块，启用一个 MapTask 进行处理，这样做的好处是避免了服务器节点之间的数据传输，提高 job 处 理效率
两种经典的控制 MapTask 的个数方案：减少 MapTask 数或者增加 MapTask 数
1、 减少 MapTask 数是通过合并小文件来实现，这一点主要是针对数据源
2、 增加 MapTask 数可以通过控制上一个 job 的 reduceTask 个数
因为 Hive 语句最终要转换为一系列的 MapReduce Job 的，而每一个 MapReduce Job 是由一 系列的 MapTask 和 ReduceTask 组成的，默认情况下， MapReduce 中一个 MapTask 或者一个 ReduceTask 就会启动一个 JVM 进程，一个 Task 执行完毕后， JVM 进程就退出。
这样如果任务花费时间很短，又要多次启动 JVM 的情况下，JVM 的启动时间会变成一个比较大的消耗， 这个时候，就可以通过重用 JVM 来解决：
set mapred.job.reuse.jvm.num.tasks=5
七、小文件合并
文件数目过多，会给 HDFS 带来压力，并且会影响处理效率，可以通过合并 Map 和 Reduce 的 结果文件来消除这样的影响、：
set hive.merge.mapfiles = true ##在 map only 的任务结束时合并小文件 set hive.merge.mapredfiles = false ## true 时在 MapReduce 的任务结束时合并小文件 set hive.merge.size.per.task = 256*1000*1000 ##合并文件的大小 set mapred.max.split.size=256000000; ##每个 Map 最大分割大小 set mapred.min.split.size.per.node=1; ##一个节点上 split 的最少值 set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; ##执行 Map 前进行小文件合并
八、设置合理的 reduceTask 的数量
Hadoop MapReduce 程序中，reducer 个数的设定极大影响执行效率，这使得 Hive 怎样决定 reducer 个数成为一个关键问题。遗憾的是 Hive 的估计机制很弱，不指定 reducer 个数的情况下，Hive 会猜测确定一个 reducer 个数，基于以下两个设定：
1、hive.exec.reducers.bytes.per.reducer（默认为 256000000）
2、hive.exec.reducers.max（默认为 1009）
3、mapreduce.job.reduces=-1（设置一个常量 reducetask 数量）
计算 reducer 数的公式很简单： N=min(参数 2，总输入数据量/参数 1) 通常情况下，有必要手动指定 reducer 个数。考虑到 map 阶段的输出数据量通常会比输入有 大幅减少，因此即使不设定 reducer 个数，重设参数 2 还是必要的。
依据 Hadoop 的经验，可以将参数 2 设定为 0.95*(集群中 datanode 个数)。
九、合并 MapReduce 操作
Multi-group by 是 Hive 的一个非常好的特性，它使得 Hive 中利用中间结果变得非常方便。 例如：
FROM (SELECT a.status, b.school, b.gender FROM status_updates a JOIN profiles b ON (a.userid = b.userid and a.ds='2009-03-20' ) ) subq1 INSERT OVERWRITE TABLE gender_summary PARTITION(ds='2009-03-20') SELECT subq1.gender, COUNT(1) GROUP BY subq1.gender INSERT OVERWRITE TABLE school_summary PARTITION(ds='2009-03-20') SELECT subq1.school, COUNT(1) GROUP BY subq1.school
上述查询语句使用了 multi-group by 特性连续 group by 了 2 次数据，使用不同的 group by key。 这一特性可以减少一次 MapReduce 操作。
