Hive存储格式及优化_hive存储优化

Hive的数据存储格式

• 列式存储、行式存储

• Hive中表的数据存储格式，不是只支持text文本格式，还支持其他很多格式。

• 建表的时候通过STORED AS 语法指定。如果没有指定默认都是textfile（行存储）。

• Hive中主流的几种文件格式。

  • textfile 文件格式

  • ORC、Parquet 列式存储格式。

    都是列式存储格式，底层是以二进制形式存储。数据存储效率极高，查询方便。
    1

  • 栗子

    分别使用3种不同格式存储数据，去HDFS上查看底层文件存储空间的差异。

    --1、创建表，存储数据格式为TEXTFILE
    create table log_text (
    track_time string,
    url string,
    session_id string,
    referer string,
    ip string,
    end_user_id string,
    city_id string
    )
    ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
    STORED AS TEXTFILE;  --如果不写stored as textfile 默认就是textfile
    
    --加载数据
    load data local inpath '/root/hivedata/log.data' into table log_text;
    
    --2、创建表，存储数据格式为ORC
    create table log_orc(
    track_time string,
    url string,
    session_id string,
    referer string,
    ip string,
    end_user_id string,
    city_id string
    )
    ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
    STORED AS orc ;
    
    --load是纯复制移动操作 不会调整文件格式。
    insert into table log_orc select * from log_text;
    
    --3、创建表，存储数据格式为parquet
    create table log_parquet(
    track_time string,
    url string,
    session_id string,
    referer string,
    ip string,
    end_user_id string,
    city_id string
    )
    ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
    STORED AS PARQUET ;
    
    --向表中插入数据 
    insert into table log_parquet select * from log_text ;
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47

• 在实际开发中，可以根据需求选择不同的文件格式并且搭配不同的压缩算法。可以得到更好的存储效果。

  -- 不压缩
  create table log_orc_none(
  track_time string,
  url string,
  session_id string,
  referer string,
  ip string,
  end_user_id string,
  city_id string
  )
  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
  STORED AS orc tblproperties ("orc.compress"="NONE");
  -- 写入数据
  insert into table log_orc_none select * from log_text ;
  
  -- 压缩数据
  create table log_orc_snappy(
  track_time string,
  url string,
  session_id string,
  referer string,
  ip string,
  end_user_id string,
  city_id string
  )
  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
  STORED AS orc tblproperties ("orc.compress"="SNAPPY");
  -- 写入
  insert into table log_orc_snappy select * from log_text ;
  
  
  --不指定压缩格式 代表什么呢？
  --orc 存储文件默认采用ZLIB 压缩。比 snappy 压缩的小
  STORED AS orc;   --2.78M
  
  --以ORC格式存储 不压缩
  STORED AS orc tblproperties ("orc.compress"="NONE");  --7.69M
  
  --以ORC格式存储  使用snappy压缩
  STORED AS orc tblproperties ("orc.compress"="SNAPPY"); --3.78M
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40

• 结论建议：在Hive中推荐使用ORC+snappy压缩。

---

Hive通用调优

• 能不使用MR就不使用MR 查询数据时尽量不适用mr，直接读文件进行切割数据展示
• 使用mr计算时，能之直接调用本地资源，就不再取yarn服务中请求资源
• mr计算速度慢可以切换计算框架 spark

1、Fetch抓取机制

• 功能：在执行sql的时候，能不走MapReduce程序处理就尽量不走MapReduce程序处理。

• 尽量直接去操作数据文件。

• 设置： hive.fetch.task.conversion= more。

  --在下述3种情况下 sql不走mr程序
  
  --全局查找
  select * from student;
  --字段查找
  select num,name from student;
  --limit 查找
  select num,name from student limit 2;
  1
  2
  3
  4
  5
  6
  7
  8

2、mapreduce本地模式

• 功能：如果非要执行MapReduce程序，能够本地执行的，尽量不提交yarn上执行。

• 默认是关闭的。意味着只要走MapReduce就提交yarn执行。

  mapreduce.framework.name = local 本地模式
  mapreduce.framework.name = yarn 集群模式 
  1
  2

• Hive提供了一个参数，自动切换MapReduce程序为本地模式，如果不满足条件，就执行yarn模式。

  set hive.exec.mode.local.auto = true;
   
  --3个条件必须都满足 自动切换本地模式
  The total input size of the job is lower than: hive.exec.mode.local.auto.inputbytes.max (128MB by default)  --数据量小于128M
  
  The total number of map-tasks is less than: hive.exec.mode.local.auto.tasks.max (4 by default)  --maptask个数少于4个
  
  The total number of reduce tasks required is 1 or 0.  --reducetask个数是0 或者 1
  1
  2
  3
  4
  5
  6
  7
  8

• 切换Hive的执行引擎

  WARNING: Hive-on-MR is deprecated in Hive 2 and may not be available in the future versions. Consider using a different execution engine (i.e. spark, tez) or using Hive 1.X releases.
  
  如果针对Hive的调优依然无法满足你的需求 还是效率低， 尝试使用spark计算引擎 或者Tez.
  1
  2
  3

3、join优化

尽量不进行两个表以上的关联拆卸

from t2 join t1

from t2 join t3

• 底层还是MapReduce的join优化。

• MapReduce中有两种join方式。指的是join的行为发生什么阶段。

  • map端join
  • reduce端join

• 优化1：Hive自动尝试选择map端join提高join的效率 省去shuffle的过程。

  开启 mapjoin 参数设置：
  （1）设置自动选择 mapjoin
  set hive.auto.convert.join = true;  --默认为 true
  （2）大表小表的阈值设置：
  set hive.mapjoin.smalltable.filesize= 25000000; 单位是 25m
  低于25 在map进行join输出数据
  1
  2
  3
  4
  5
  6

• 优化2：大表join大表

  --背景：
  大表join大表本身数据就十分具体，如果join字段存在null空值 如何处理它？
  
  --方式1：空key的过滤  此行数据不重要
  参与join之前 先把空key的数据过滤掉
  SELECT a.* FROM (SELECT * FROM nullidtable WHERE id IS NOT NULL ) a JOIN ori b ON a.id =b.id;
  
  --方式2：空Key转换
  CASE WHEN a.id IS NULL THEN 'xxx任意字符串' ELSE a.id END     hive001  hive11 hive231  
  CASE WHEN a.id IS NULL THEN concat('hive', rand()) ELSE a.id  --避免转换之后数据倾斜 随机分布打散
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

• 优化3：桶表join提高优化效率。bucket mapjoin

  1.1 条件
  	1） set hive.optimize.bucketmapjoin = true;
  	2） 一个表的bucket数是另一个表bucket数的整数倍
  	3） bucket列 == join列
  	4） 必须是应用在map join的场景中
  
  1.2 注意
  	1）如果表不是bucket的，只是做普通join。
  
  1
  2
  3
  4
  5
  6
  7
  8
  9

4、group by 数据倾斜优化

（1）是否在 Map 端进行聚合，默认为 True
set hive.map.aggr = true;
（2）在 Map 端进行聚合操作的条目数目
set hive.groupby.mapaggr.checkinterval = 100000;
（3）有数据倾斜的时候进行负载均衡（默认是 false）
set hive.groupby.skewindata = true;

--Q:在hive中数据倾斜开启负载均衡之后 底层执行机制是什么样？
男性 女性
--step1:启动一个MapReduce程序 将倾斜的数据随机发送到各个reduce中 进行打散 
        每个reduce进行聚合都是局部聚合
        
--step2:再启动第二个MapReduce程序 将上一步局部聚合的结果汇总起来进行最终的聚合       

1
2
3
4
5
6
7
8
9
10
11
12
13
14

5、hive中如何调整底层MapReduce中task的个数（并行度）

• maptask个数

  • 如果是在MapReduce中 maptask是通过逻辑切片机制决定的。

  • 但是在hive中，影响的因素很多。比如逻辑切片机制，文件是否压缩、压缩之后是否支持切割。

  • 因此在Hive中，调整MapTask的个数，直接去HDFS调整文件的大小和个数，效率较高。

    如果小文件多，就进行小文件的合并  合并的大小最好=block size
    130m 65m 65m   128m  64m
    如果大文件多，就调整blocl size
    
    hdfs-site  dfs.block.size
    
    1
    2
    3
    4
    5
    6

• reducetask个数

  • 如果在MapReduce中，通过代码可以直接指定 job.setNumReduceTasks(N)

  • 在Hive中，reducetask个数受以下几个条件控制的

    （1）每个 Reduce 处理的数据量默认是 256MB
    hive.exec.reducers.bytes.per.reducer=256000000
    （2）每个任务最大的 reduce 数，默认为 1009
    hive.exec.reducsers.max=1009
    （3）mapreduce.job.reduce=3
    该值默认为-1，由 hive 自己根据任务情况进行判断。
    
    
    --如果用户用户不设置 hive将会根据数据量或者sql需求自己评估reducetask个数。
    --用户可以自己通过参数设置reducetask的个数
      set mapreduce.job.reduces = N
    --用户设置的不一定生效，如果用户设置的和sql执行逻辑有冲突，比如order by，在sql编译期间，hive又会将reducetask设置为合理的个数。  
    
    Number of reduce tasks determined at compile time: 1
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15

6、其他几个通用调优

• 执行计划explain

  • 通过执行计划可以看出hive接下来是如何打算执行这条sql的。

  • 语法格式：explain + sql语句

  • 栗子

    explain select * from student;
    
    +----------------------------------------------------+
    |                      Explain                       |
    +----------------------------------------------------+
    | STAGE DEPENDENCIES:                                |
    |   Stage-0 is a root stage                          |
    |                                                    |
    | STAGE PLANS:                                       |
    |   Stage: Stage-0                                   |
    |     Fetch Operator                                 |
    |       limit: -1                                    |
    |       Processor Tree:                              |
    |         TableScan                                  |
    |           alias: student                           |
    |           Statistics: Num rows: 1 Data size: 5260 Basic stats: COMPLETE Column stats: NONE |
    |           Select Operator                          |
    |             expressions: num (type: int), name (type: string), sex (type: string), age (type: int), dept (type: string) |
    |             outputColumnNames: _col0, _col1, _col2, _col3, _col4 |
    |             Statistics: Num rows: 1 Data size: 5260 Basic stats: COMPLETE Column stats: NONE |
    |             ListSink                               |
    |                                                    |
    +----------------------------------------------------+
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24

• 并行执行机制

  • 如果hivesql的底层某些stage阶段可以并行执行，就可以提高执行效率。

  • 前提是stage之间没有依赖 并行的弊端是瞬时服务器压力变大。

  • 参数

    set hive.exec.parallel=true; --是否并行执行作业。适用于可以并行运行的 MapReduce 作业，例如在多次插入期间移动文件以插入目标
    set hive.exec.parallel.thread.number=16; --最多可以并行执行多少个作业。默认为8。
    
    1
    2
    3

• Hive的严格模式

  • 注意。不要和动态分区的严格模式搞混淆。

  • 这里的严格模式指的是开启之后 hive会禁止一些用户都影响不到的错误包括效率低下的操作，不允许运行一些有风险的查询。

  • 设置

    set hive.mapred.mode = strict --默认是非严格模式  nonstrict
    
    1
    2

  • 解释

    1、如果是分区表，没有where进行分区裁剪 禁止执行
    2、order by语句必须+limit限制
    
    1
    2
    3

• 推测执行机制

  • MapReduce中task的一个机制。
  • 功能：
    • 一个job底层可能有多个task执行，如果某些拖后腿的task执行慢，可能会导致最终job失败。
    • 所谓的推测执行机制就是通过算法找出拖后腿的task,为其启动备份的task。
    • 两个task同时处理一份数据，谁先处理完，谁的结果作为最终结果。
  • 推测执行机制默认是开启的，但是在企业生产环境中建议关闭。