2023.11.28-电商平台建设03 - 大数据调优手段

1.优化手段

        1.1分桶表

HIVE的分桶本质上就是MR的分区操作

建表语句:

create table 表名(
    字段 类型,
    ....
)
clustered by(分桶字段) [sorted by (字段 [asc | desc])] into N buckets   --- 定义分桶表核心语句
row format......

分桶的作用

1) 进行数据采样工作
    1.1) 当表的数据量比较庞大的时候, 在编写SQL语句后, 需要首先测试 SQL是否可以正常的执行,  需要在表中执行查询操作, 由于表数据量比较庞大, 在测试一条SQL的时候整个运行的时间比较久, 为了提升测试效率, 可以整个表抽样出一部分的数据, 进行测试
    1.2) 校验数据的可行性(质量校验)
    1.3) 进行统计分析的时候, 并不需要统计出具体的指标, 可能统计的都是一些相对性指标, 比如说一些比率(合格率)问题, 此时可以通过采样处理
  

2) 提升查询的效率(更主要是提升JOIN的效率)
    可以减少JOIN次数, 从而提升效率

1.2 数据采样

采样函数:

tablesample(bucket x out of y  [on column] )

使用位置:跟在表名的后面,如果表有别名,必须放置在别名的前面.

说明:

        x :从第几个桶开始进行采样

        y:抽样比例

        column:分桶的字段,可以省略   (注:x不能大于y,y必须是表的分桶数量的倍数或者因子)

 

1.3 Jion优化操作

reduce端Join操作, 存在那些弊端呢? 
1- 可能会存在数据倾斜的问题 (某几个reduce接收数据量远远大于其他的reduce接收数据量)
2- 所有的数据处理的操作, 全部都压在reduce中进行处理, 而reduce数量相比Map来说少的多,导致整个reduce压力比较大

解决思路:不让reduce做聚合处理,将这项工作交给mapTask

1 - Map Join

       每一个maptask在读取数据的时候,每读取一条数据,就会和内存中表的数据进行匹配,如果能匹配的上,将匹配上数据合并在一起,输出即可

        好处:原有的reduce join 问题可以解决

        弊端:比较消耗内存,要求整个join中,必须都有一个小表,否则无法放到内存中.

具体使用:

-- map join
set hive.auto.convert.join; -- 开启 map join的支持  默认值为True
set hive.auto.convert.join.noconditionaltask.size; -- 设置 小表数据量的最大阈值: 默认值为20971520

如果不满足条件, HIVE会自动使用 reduce join 操作

2 - Bucket Map Join

适用场景:中型表和大表join:

        方案一:如果中型表能对数据进行提前过滤,过滤后,有可能满足了MapJoin条件

        方案二:Bucket Map join

-- bucket map join
set hive.optimize.bucketmapjoin --默认false
/*
    1. Join两个表必须是分桶表
    2.开启Bucket Map Join支持 : set hive.optimize.bucketmapjoin  = True;
    3.一个表的分桶数量是另一个表的分桶数量的整数倍;
    4.分桶列必须是Join的on条件的列
    5.必须建立在Map Join场景中(中型表是小表的3倍,此时分至少3个桶)
*/

3 - SMB  Join

• 适用场景: 大表 和 大表 join

• 解决方案: SMB Join ( sort merge bucket map join)

-- 使用条件:
--     1- 两个表必须都是分桶表
--     2- 开启 SMB Join 支持:
        set hive.auto.convert.sortmerge.join; -- 默认false
        set hive.optimize.bucketmapjoin.sortedmerge ;-- 默认false
        set hive.auto.convert.sortmerge.join.noconditionaltask;--  Hive 0.13.0默认开启
--    3- 两个表的分桶的数量是一致的
--    4- 分桶列 必须是 join的 on条件的列, 同时必须保证按照分桶列进行排序操作
       -- 开启强制排序
       set hive.enforce.sorting; -- hive2.x移除 默认true
       -- 在建分桶表使用: 必须使用sorted by()
 
--    5-  应用在Bucket Map Join 场景中
       -- 开启 bucket map join
       set hive.optimize.bucketmapjoin ; --默认false
 
--    6- 必须开启HIVE自动尝试使用SMB 方案:
       set hive.optimize.bucketmapjoin.sortedmerge; -- 默认false
       
  
建表:
  create table test_smb_2(mid string,age_id string) CLUSTERED BY(mid) SORTED BY(mid) INTO 500 BUCKETS;

1.4 HIVE的索引

 hive的原始索引存在弊端: 
        hive原始索引不会自动更新,每次表中数据发生变化后, 都是需要手动重建索引操作, 比较耗费时间和资源, 整体提升性能一般

1.4.1 Row Group Index索引

Row Group Index索引

row group index: 行组索引

条件: 
    1) 要求表的存储类型为ORC存储格式
    2) 在创建表的时候, 必须开启 row group index 索引支持
        'orc.create.index'='true'
    3) 在插入数据的时候, 必须保证需求进行索引列, 按序插入数据
 
适用于: 数值类型的, 并且对数值类型进行 >  <  = 操作
 
思路: 
    插入数据到ORC表后, 会自动进行划分为多个script片段, 每个片段内部, 会保存着每个字段的最小, 最大值, 这样, 当执行查询 > < = 的条件筛选操作的时候, 根据最小最大值锁定相关的script片段, 从而减少数据扫描量, 提升效率
    
操作: 
    CREATE TABLE lxw1234_orc2 (字段列表 ....) stored AS ORC 
    TBLPROPERTIES ( 
      'orc.compress'='SNAPPY', 
      -- 开启行组索引 
      'orc.create.index'='true' 
    )
    
    插入数据的时候, 需要保证数据有序的
    insert overwrite table lxw1234_orc2
    SELECT  id, pcid FROM lxw1234_text 
    -- 插入的数据保持排序(可以使用全局排序, 也可以使用局部排序, 只需要保证一定有序即可, 建议使用局部排序 插入数据效率高一些, 因为全局排序只有一个reduce) 
    DISTRIBUTE BY id sort BY id;
    
使用: 
    set hive.optimize.index.filter; -- 默认true
    SELECT COUNT(1) FROM lxw1234_orc1 WHERE id >= 1382 AND id <= 1399;

1.4.2 Bloom Fliter Index 索引

布隆过滤器

条件: 
    1) 要求表的存储类型为 ORC存储方案
    2) 在建表的时候, 必须设置为那些列构建布隆索引
    3) 仅能适合于等值过滤查询操作
 
思路: 
    在开启布隆过滤索引后, 可以针对某个列, 或者某几列来建立索引, 构建索引后, 会将这一列的数据的值存储在对应script片段的索引信息中, 这样当进行 等值查询的时候, 首先会到每一个script片段的索引中, 判断是否有这个值, 如果没有, 直接跳过script, 从而减少数据扫描量, 提升效率
    

操作:  
    CREATE TABLE lxw1234_orc2 (字段列表....)
    stored AS ORC 
    TBLPROPERTIES ( 
      'orc.compress'='SNAPPY', 
      -- 开启 行组索引 (可选的, 支持全部都打开, 也可以仅开启一个)
      'orc.create.index'='true', 
      -- pcid字段开启BloomFilter索引 
      'orc.bloom.filter.columns'='pcid,字段2,字段3...'
    )
    
    插入数据: 没有要求, 当然如果开启行组索引, 可以将需要使用行组索引的字段, 进行有序插入即可
    
使用:
    set hive.optimize.index.filter; -- 默认true
    SELECT COUNT(1) FROM lxw1234_orc1 
    WHERE id >= 0 AND id <= 1000 -- 底层用了行组索引
    AND pcid IN ('001','002');  -- 底层用了布隆过滤索引

1- 对于行组索引: 我们建议只要数据存储格式为ORC, 建议将这种索引全部打开, 至于导入数据的时候, 如果能保证有序, 那最好, 如果保证不了, 也无所谓, 大不了这个索引的效率不是特别好

2- 对于布隆过滤索引: 建议将后续会大量的用于等值连接的操作字段, 建立成布隆索引, 比如说: JOIN的字段  经常在where后面出现的等值连接字段

1.5如何解决数据倾斜问题

reduce端Join存在数据倾斜的问题

解决方案一:

可以通过  Map Join  Bucket Map Join   以及  SMB Join 解决
    
注意:  
   通过 Map Join,Bucket Map Join,SMB Join 来解决数据倾斜, 但是 这种操作是存在使用条件的, 如果无法满足这些条件,  无法使用 这种处理方案

解决方案二:

思路:  将那些产生倾斜的key和对应v2的数据, 从当前这个MR中移出去, 单独找一个MR来处理即可, 处理后, 和之前的MR进行汇总结果即可

关键问题:  如何找到那些存在倾斜的key呢?  特点: 这个key数据有很多

运行期处理方案:
    思路: 在执行MR的时候, 会动态统计每一个 k2的值出现重复的次数, 当这个重复的次数达到一定的阈值后, 认为当前这个k2的数据存在数据倾斜, 自动将其剔除, 交由给一个单独的MR来处理即可,两个MR处理完成后, 将结果基于union all 合并在一起即可
    
    实操:  
        set hive.optimize.skewjoin=true;  -- 开启运行期处理倾斜参数默认false
        set hive.skewjoin.key=100000;   -- 阈值,  此参数在实际生产环境中, 需要调整在一个合理的值(否则极易导致大量的key都是倾斜的),默认100000
            判断依据: 查看 join的 字段 对应重复的数量有多少个, 然后选择一个合理值
              比如判断:  id为 1  大概有 100w  id为 2 88w  id 为 3 大概有 500w   设置阈值为 大于500w次数据
               或者: 总数量大量1000w, 然后共有 1000个班级, 平均下来每个班级数量大概在 1w条, 设置阈值:  大于 3w条 ~5w条范围 (超过3~5倍才认为倾斜)
        
    
    适用于: 并不清楚那个key容易产生倾斜, 此时交由系统来动态检测

编译期处理方案: 
    思路:  在创建这个表的时候, 我们就可以预知到后续插入到这个表中数据, 那些key的值会产生倾斜, 在建表的时候, 将其提前配置设置好即可, 在后续运行的时候, 程序会自动将设置的key的数据单独找一个MR来进行处理即可, 处理完成后, 再和原有结果进行union all 合并操作
    
    实操:  
        set hive.optimize.skewjoin.compiletime=true;  -- 开启编译期处理倾斜参数
        
        CREATE TABLE list_bucket_single (key STRING, value STRING) 
        -- 倾斜的字段和需要拆分的key值 
        SKEWED BY (key) ON (1,5,6) 
        -- 为倾斜值创建子目录单独存放 
        [STORED AS DIRECTORIES];

    适用于:  提前知道那些key存在倾斜

group by 数据倾斜

 通过规约来解决数据倾斜, 处理完成后, 发现 两个reduce中从原来相差 3倍, 变更为相差 2倍, 减轻了数据倾斜问题
 
 
 如何配置呢? 
     只需要在HIVE中开启combiner提前聚合配置参数即可:  
         set hive.map.aggr=true;