43 压缩和存储、调优

2023-10-02 22:08:04

压缩参数配置

  • 要在Hadoop中启用压缩，可以在mapred-site.xml中配置如下参数
默认值 阶段 参数 建议
org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec, org.apache.hadoop.io.compress.Lz4Codec 输入压缩 io.compression.codecs （在core-site.xml中配置） Hadoop使用文件扩展名判断是否支持某种编解码器
false mapper输出 mapreduce.map.output.compress 这个参数设为true启用压缩
org.apache.hadoop.io.compress.DefaultCodec mapper输出 mapreduce.map.output.compress.codec 使用LZO、LZ4或snappy编解码器在此阶段压缩数据
false reducer输出 mapreduce.output.fileoutputformat.compress 这个参数设为true启用压缩
org.apache.hadoop.io.compress. DefaultCodec reducer输出 mapreduce.output.fileoutputformat.compress.codec 使用标准工具或者编解码器，如gzip和bzip2
RECORD reducer输出 mapreduce.output.fileoutputformat.compress.type SequenceFile输出使用的压缩类型：NONE和BLOCK

Map输出阶段压缩

  • 开启map输出阶段压缩可以减少job中map和Reduce Task间数据传输量

案例实操

# 开启hive中间传输数据压缩功能
set hive.exec.compress.intermediate=true;

# 开启mapreduce中map输出压缩功能
set mapreduce.map.output.compress=true;

# 设置mapreduce中map输出数据的压缩方式
set mapreduce.map.output.compress.codec=
 org.apache.hadoop.io.compress.SnappyCodec;

# 执行查询语句
select count(ename) name from emp;

Reduce输出阶段压缩

# 开启hive最终输出数据压缩功能
set hive.exec.compress.output=true;

# 开启mapreduce最终输出数据压缩
set mapreduce.output.fileoutputformat.compress=true;

# 设置mapreduce最终数据输出压缩方式
set mapreduce.output.fileoutputformat.compress.codec =
 org.apache.hadoop.io.compress.SnappyCodec;

# 设置mapreduce最终数据输出压缩为块压缩
set mapreduce.output.fileoutputformat.compress.type=BLOCK;

# 测试一下输出结果是否是压缩文件
insert overwrite local directory
 '/opt/module/hive/datas/distribute-result' select * from emp distribute by deptno sort by empno desc;

文件存储格式

列式存储和行式存储

43 压缩和存储、调优

特点

  • 行存储：查询速度更快
  • 列存储：可以针对性设计更好的设计压缩算法
  • textfile和sequencefile都是基于行存储的
  • orc和parquet是基于列式存储的

Orc格式

43 压缩和存储、调优

Parquet格式

  • 以二进制方式存储，不可以直接读取
  • 文件中包括该文件的数据和元数据，因此Parquet格式的文件是自解析的

存储和压缩结合

  • 在实际项目开发中，Hive表的数据存储格式一般选择orc或parquet
  • 压缩方式一般选择snappy，lzo

执行计划（Explain）

explain select * from emp;

Fetch抓取

  • Hive中对某些情况的查询可以不使用MapReduce计算
  • 例如select * from employees
  • 这种情况下Hive可以简单地读取employee对应的存储目录下的文件，然后输出查询结果到控制台

案例实操

# 把 hive.fetch.task.conversion 设置成 none，如下查询都会走mapreduce
# 设置成more,都不会走mapreduce
set hive.fetch.task.conversion=more;
select * from emp;
select ename from emp;
select ename from emp limit 3;

本地模式

# 开启本地mr
set hive.exec.mode.local.auto=true;  

# 设置local mr的最大输入数据量，当输入数据量小于这个值时采用local  mr的方式，默认为134217728，即128M
set hive.exec.mode.local.auto.inputbytes.max=50000000;

# 设置local mr的最大输入文件个数，当输入文件个数小于这个值时采用local mr的方式，默认为4
set hive.exec.mode.local.auto.input.files.max=10;

小表大表Join(MapJoin)

# 设置自动选择Mapjoin
set hive.auto.convert.join = true; 默认为true

# 大表小表的阈值设置（默认25M以下认为是小表）
set hive.mapjoin.smalltable.filesize = 25000000;

MapJoin工作机制

43 压缩和存储、调优

大表Join大表

空key过滤

  • 有时join超时是因为某些key对应的数据太多
  • 而相同的key都会发送到相同的reducer上，导致内存不够
  • 很多情况下，这些key对应的数据是异常的数据
  • 我们需要在SQL语句中进行过滤，例如key对应的字段为空

空key转换

  • 有时候虽然某个key为空对应的数据很多，但相应的数据并不是异常数据
  • 此时我们可以为key为空的字段赋一个随机的值，使数据随机均匀地分布到不同的reducer上

Group by

  • 默认情况下，Map阶段同一Key数据分发给一个reduce，当一个key数据过大时就倾斜了
  • 可以在Map端进行部分聚合，最后在Reduce端得出最终结果
# 是否在Map端进行聚合，默认为True
set hive.map.aggr = true

# 在Map端进行聚合操作的条目数目
set hive.groupby.mapaggr.checkinterval = 100000

# 有数据倾斜的时候进行负载均衡（默认是false）
set hive.groupby.skewindata = true

Count(Distinct)去重统计

  • 数据量大的情况下，由于count distinct操作需要用到一个reduce task来完成
  • 这个reduce需要处理的数据量太大，就会导致整个job很难完成
  • 一般count distinct使用先group by再count的方式替换

合理设置Map及Reduce数

是不是map数越多越好？

  • 答案是否定的
  • 如果一个任务有很多个小文件，则每个小文件也会被当作一个块，用一个map任务来完成
  • 而一个map任务启动和初始化的时间远大于逻辑处理时间，就会造成很大的资源浪费
  • 而且，同时可执行的map数是受限的

是不是保证每个map处理接近128M的文件快，就可以高枕无忧了？

  • 答案是不一定
  • 比如有一个127M的文件，正常会用一个map去完成
  • 但这个文件只有一个或两个小字段，却有几千万的记录
  • 如果map处理的逻辑比较复杂，用一个map任务去做，肯定比较耗时
  • 此时应该增加map数

小文件进行合并

# 在map执行前合并小文件，减少map数
set hive.input.format= org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

# 在map-only任务结束时合并小文件，默认true
SET hive.merge.mapfiles = true;

# 在map-reduce任务结束时合并小文件，默认false
SET hive.merge.mapredfiles = true;

# 合并文件的大小，默认256M
SET hive.merge.size.per.task = 268435456;

# 当输出文件的平均大小小于该值时，启动一个独立的map-reduce任务进行文件merge
SET hive.merge.smallfiles.avgsize = 16777216;

合理设置Reduce数

# 方式一
# 每个Reduce处理的数据量默认是256MB
hive.exec.reducers.bytes.per.reducer=256000000
# 每个任务最大的reduce数，默认为1009
hive.exec.reducers.max=1009
# 计算reducer数的公式
N=min(参数2，总输入数据量/参数1)

# 方式二
# mapred-default.xm
set mapreduce.job.reduces = 15;

reduce个数并不是越多越好

  • 过多的启动和初始化reduce会消耗时间和资源
  • 有多少个reduce，就会有多少个小文件，如果这些小文件作为下一个任务的输入，也会出现小文件过多的问题

并行执行

# 打开任务并行执行
set hive.exec.parallel=true; 

# 同一个sql允许最大并行度，默认为8
set hive.exec.parallel.thread.number=16;

严格模式

  • Hive可以通过设置防止一些危险操作

分区表不使用分区过滤

# 对于分区表，除非where语句中含有分区字段过滤条件来限制范围，否则不允许执行
hive.strict.checks.no.partition.filter=true;

order by没有 limit过滤

# 对于使用了order by语句的查询，要求必须使用limit语句
hive.strict.checks.orderby.no.limit=true;

笛卡尔积

# 会限制笛卡尔积的查询
hive.strict.checks.cartesian.product=true
上一篇：IS 和 ==


下一篇：【无标题】PAT (Basic Level) Practice （中文）1011 A+B 和 C (15 分)——解析