Hive作业调优

作业调优方案

代码优化

• 数据清洗

  1. 读取表时分区过滤，避免全表扫描。

  2. 数据过滤之后再JOIN。

  3. 重复使用数据时，避免重复计算，构建中间表，重复使用中间表。

• 多DISTINCT优化

  • 优化前代码

    多个DISTINCT时，数据会出现膨胀。

    SELECT k,
           COUNT(DISTINCT CASE WHEN a > 1 THEN user_id END) user1,
           COUNT(DISTINCT CASE WHEN a > 2 THEN user_id END) user2,
           COUNT(DISTINCT CASE WHEN a > 3 THEN user_id END) user3,
           COUNT(DISTINCT CASE WHEN a > 4 THEN user_id END) user4
    FROM t  
    GROUP BY k;

  • 优化后代码

    通过两次GROUP BY的方式代替DISTINCT操作，通过内层的GROUP BY去重并降低数据量，通过外层的GROUP BY取SUM，即可实现DISTINCT的效果。

    SELECT k,
           SUM(CASE WHEN user1 > 0 THEN 1 ELSE 0 END) AS user1,
           SUM(CASE WHEN user2 > 0 THEN 1 ELSE 0 END) AS user2,
           SUM(CASE WHEN user3 > 0 THEN 1 ELSE 0 END) AS user3,
           SUM(CASE WHEN user4 > 0 THEN 1 ELSE 0 END) AS user4
    FROM 
            (SELECT k,
                    user_id,
                    COUNT(CASE WHEN a > 1 THEN user_id END) user1,
                    COUNT(CASE WHEN a > 2 THEN user_id END) user2,
                    COUNT(CASE WHEN a > 3 THEN user_id END) user3,
                    COUNT(CASE WHEN a > 4 THEN user_id END) user4
            FROM t
            GROUP BY k, user_id  
            ) tmp 
    GROUP BY k;

• GROUP BY 操作中的数据倾斜

  热点Key处理方式如下：

  • 当执行GROUP BY操作时如果发现存在热点键值（即某些键值对应的记录数远超其他），可以通过以下步骤来优化：

    1. 启用Map端聚合：首先，通过设置Hive参数开启Map阶段的部分聚合功能，减少传输到Reduce阶段的数据量。

       set hive.map.aggr=true;
       set hive.groupby.mapaggr.checkinterval=100000;  -- 用于设定Map端进行聚合操作的条目数

    2. 可以对Key随机化打散，多次聚合，或者直接设置。

       set hive.groupby.skewindata=true;

       当选项设定为true时，生成的查询计划有两个MapReduce任务。在第一个MapReduce中，Map的输出结果集合会随机分布到Reduce中， 每个部分进行聚合操作，并输出结果。这样处理的结果是，相同的Group By Key有可能分发到不同的Reduce中，从而达到负载均衡的目的；第二个MapReduce任务再根据预处理的数据结果按照Group By Key分布到Reduce中（这个过程可以保证相同的Group By Key分布到同一个Reduce中），最后完成最终的聚合操作。

  • 如果两个大表进行JOIN操作时，出现热点，则使用热点Key随机化。

    例如，log表存在大量user_id为null的记录，但是表bmw_users中不会存在user_id为空，则可以把null随机化再关联，这样就避免null值都分发到一个Reduce Task上。代码示例如下。

    SELECT * FROM log a LEFT OUTER 
    JOIN bmw_users b ON 
    CASE WHEN a.user_id IS NULL THEN CONCAT('dp_hive',RAND()) ELSE a.user_id=b.user_id END;

  • 如果大表和小表进行JOIN操作时，出现热点，则使用MAP JOIN。

内存参数

CPU参数

Task数量优化

• Map Task数量优化

  在分布式计算系统中，Map数量的决定因素之一是原始数据块的数量，因为通常每个Task会读取一个数据块；然而，这不是绝对的，当文件数量多且文件较小时，可以减少初始Map Task的数量以节约资源，而当文件数量少但文件较大时，可以增加Map Task的数量以减轻单个Task的负担。

  通常，Map Task数量是由mapred.map.tasks、mapred.min.split.size和dfs.block.size决定的。

  • Hive的文件基本上都是存储在HDFS上，而HDFS上的文件，都是分块的，所以具体的Hive数据文件在HDFS上分多少块，可能对应的是默认Hive起始的Task的数量，使用default_mapper_num参数表示。使用数据总大小除以dfs默认的最大块大小来决定初始默认数据分区数。

    初始默认的Map Task数量，具体公式如下。

    default_mapper_num = total_size/dfs.block.size

  • 计算Split的size，具体公式如下。

    default_split_size = max(mapred.min.split.size, min(mapred.max.split.size, dfs.block.size))

    上述公式中的mapred.min.split.size和mapred.max.split.size，分别为Hive计算的时Split的最小值和最大值。

  • 将数据按照计算出来的size划分为数据块，具体公式如下。

    split_num = total_size/default_split_size；

  • 计算的Map Task数量，具体公式如下。

    map_task_num = min(split_num, max(mapred.map.tasks, default_mapper_num))

    从上面的过程来看，Task的数量受各个方面的限制，不至于Task的数量太多，也不至于Task的数量太少。如果需要提高Task数量，就要降低mapred.min.split.size的数值，在一定的范围内可以减小default_split_size的数值，从而增加split_num的数量，也可以增大mapred.map.tasks的数量。

    重要

    Hive on TEZ和Hive on MR使用是有差异的。例如，在Hive中执行一个Query时，可以发现Hive的执行引擎在使用Tez与MR时，两者生成的mapper数量差异较大。主要原因在于Tez中对inputSplit做了grouping操作，可以将多个inputSplit组合成更少的groups，然后为每个group生成一个mapper任务，而不是为每个inputSplit生成一个mapper任务。

• Reduce Task数量优化

  • 通过hive.exec.reducers.bytes.per.reducer参数控制单个Reduce处理的字节数。

    Reduce的计算方法如下。

    reducer_num = min(total_size/hive.exec.reducers.bytes.per.reducers, hive.exec.reducers.max)。

  • 通过mapred.reduce.tasks参数来设置Reduce Task的数量。

    说明

    在TEZ引擎模式下，通过命令set hive.tez.auto.reducer.parallelism = true;，TEZ将会根据vertice的输出大小动态预估调整Reduce Task的数量。

    同Map一样，启动和初始化Reduce也会消耗时间和资源。另外，有多少个Reduce，就会有多少个输出文件，如果生成了很多个小文件，并且这些小文件作为下一个任务的输入，则会出现小文件过多的问题。

并行运行

并行运行表示同步执行Hive的多个阶段。Hive在执行过程中，将一个查询转化成一个或者多个阶段。某个特定的Job可能包含多个阶段，而这些阶段可能并非完全相互依赖的，也就是可以并行运行的，这样可以使得整个Job的运行时间缩短。

Fetch task

开启向量化

合并小文件

大量小文件容易在文件存储端造成瓶颈，影响处理效率。对此，您可以通过合并Map和Reduce的结果文件来处理。

配置示例

SET hive.merge.mapfiles = true;
SET hive.merge.mapredfiles = true;
SET hive.merge.size.per.task = 536870912; -- 512 MB