Index

Have anice day :)
友链:

HIVE 数仓

目录

  1. HIVE 数仓
    1. HIVE
      1. 内部表和外部表的区别
      2. 分区和分桶的区别
        1. 分区
        2. 分桶
          1. 上传到分区目录, 令分区表和数据关联
      3. order/sort/distribute/cluster by 的区别
      4. HIVE 的数据倾斜
        1. 针对数据内容设置合理的 Map 数量
        2. 小文件合并
        3. 复杂文件增加 Map 数
        4. 合理设置 Reduce 数
      5. HIVE 的 UDF 怎么实现
      6. HIVE 的工作流
      7. HIVE 分区是否越多越好
      8. HIVE 调优
        1. hive-site.xml
        2. HIVE CLI 调整
      9. HIVE 压缩
        1. HIVE 数据压缩
          1. 压缩配置参数
        2. HIVE 文件压缩

HIVE

HIVE 是基于 Hadoop 的一个数据仓库工具, 可以将结构化的数据文件映射为一张数据库表, 并提供类 SQL 查询功能.

内部表和外部表的区别

通常情况下使用外部表保证数据安全, 中间表, 结果表则使用内部表 (管理表).

分区和分桶的区别

分区

指按照数据表的某列或某几列进行分区, 区域从形式上可以理解为文件目录, 若大量的数据保存在一个目录下, 查询的时候会很慢而且占用大量资源.

这个时候就可以按照数据中具有特征和共同性的字段作为分区字段, 不同特征存在不同分区, 这时查询只需要按照字段名就能在特定分区下查询.

简单理解分区就是 HDFS 上分目录, 分桶就是分成单独文件.

分桶

分桶则是对分区更细粒度的划分.

分桶将整个数据内容安装某列属性值的哈希值进行分区. 按照某一属性分为 N 个桶, 就是对该属性值的哈希值对 N 取模, 按照结果对数据分桶.

上传到分区目录, 令分区表和数据关联

直接将新的分区文件上传到 HDFS, HIVE 没有对应元数据所以无法查询到.

order/sort/distribute/cluster by 的区别

HIVE 的数据倾斜

通过 YARN 监控平台的 Task 的运行状态, 超时和失败的都可能是发生数据倾斜的地方.

数据倾斜的根源是 Key 分布不均匀, 不让数据分区, 直接在 map 端搞定, 或者在分区时清洗集中无效的 Key, 或打乱 Key 使其进入到不同的 Reduce 中.

针对数据内容设置合理的 Map 数量

主要的决定因素有: input 的文件总个数, input 的文件大小, 集群设置的文件块大小.

通常情况下, 作业会通过 input 的目录产生一个或者多个 map 任务.

map 数越多越好?

不. 如果一个任务有很多小文件 (远远小于块大小 128m), 则每个小文件也会被当做一个块, 用一个 map 任务来完成, 而一个 map 任务启动和初始化的时间远远大于逻辑处理的时间, 就会造成很大的资源浪费. 而且, 同时可执行的 map 数是受限的.

保证每个 map 都是 128 mb?

不. 比如有一个 127m 的文件, 正常会用一个 map 去完成, 但这个文件只有一个或者两个小字段, 却有大量记录, 如果 map 处理的逻辑比较复杂, 用一个 map 任务去做, 肯定也比较耗时.

小文件合并

在 Map 前合并小文件. 例如合并小于 iAmSize 的文件:

set mapred.max.split.size = iAmSize;
set mapred.min.split.size.per.node = iAmSize;
set mapred.min.split.size.per.rack = iAmSize;
set hive.input.format = org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

复杂文件增加 Map 数

当 Input 的文件都很大, 任务逻辑复杂, Map 执行非常慢的时候, 可以考虑增加 Map 数, 来使得每个 Map 处理的数据量减少, 从而提高任务的执行效率.

根据 computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize))) 公式调整 maxSize 最大值. 让 maxSize 最大值低于 blocksize 就可以增加 map 的个数

-- 默认值为 1 --
set mapreduce.input.fileinputformat.split.minsize = 1

-- 默认值 Long.MAXValue, 默认情况下, 切片大小 = blocksize --
set mapreduce.input.fileinputformat.split.maxsize = Long.MAXValue

-- 切片最大值, 参数如果比 blocksize 小, 则会让切片变小, 且等于配置的参数值. --
set maxsize:
-- 切片最小值, 参数调的比 blockSize 大, 则让切片变得比 blocksize 大. --
set minsize:
-- 设置 maxsize 大小为 N mb, 即单个 fileSplit 的大小为 N mb. --
set mapreduce.input.fileinputformat.split.maxsize = N*1024*1024;

合理设置 Reduce 数

设置每一个 job 中 reduce 个数 set mapreduce.job.reduces = N;.

HIVE 的 UDF 怎么实现

UDF 是 Hive 提供的自定义函数工具.

HIVE 的工作流

  1. 查询: HIVE 接口, 命令行或 Web UI 发送查询驱动程序.
  2. get Plan: 驱动程序查询编译器.
  3. 语法分析.
  4. 语义分析.
  5. 逻辑计划产生.
  6. 逻辑计划优化.
  7. 物理计划生成.
  8. 物理计划优化.
  9. 物理计划执行.
  10. 返回查询结果.

HIVE 分区是否越多越好

合理的分区不应该有过多的分区和文件目录, 并且每个目录下的文件应该足够大.

HIVE 调优

不根据业务内容的调优都是整蛊.

hive-site.xml

<configuration>
    <!-- 合并 block 减少 Task 数量 -->
    <property>
        <name>ngmr.partition.automerge</name>
        <value>true</value>
    </property>
    <!-- 将 n 个 block 排给单个线程处理 -->
    <property>
        <name>ngmr.partition.mergesize.mb</name>
        <value>n</value>
    </property>
    <!-- 小文件合并 -->
    <property>
        <name>hive.merge.sparkfiles</name>
        <value>true</value>
    </property>
    <property>
        <name>hive.map.agg</name>
        <value>true</value>
    </property>
    <!-- 使用向量化查询 -->
    <property>
        <name>hive.vectorized.execution.enabled</name>
        <value>true</value>
    </property>
    <!-- cbo 优化 HIVE 查询 -->
    <property>
        <name>hive.cbo.enable</name>
        <value>true</value>
    </property>
    <property>
        <name>hive.stats.fetch.column.stats</name>
        <value>true</value>
    </property>
    <property>
        <name>hive.stats.fetch.partition.stats</name>
        <value>true</value>
    </property>
    <property>
        <name>hive.compute.query.using.stats</name>
        <value>true</value>
    </property>
    <!-- 数据压缩 -->
    <property>
        <name>hive.exec.compress.intermediate</name>
        <value>true</value>
    </property>
    <property>
        <name>hive.exec.compress.output</name>
        <value>true</value>
    </property>
    <!-- 简单 SQL 不使用 Spark -->
    <property>
        <name>hive.fetch.Task.conversion</name>
        <value>more</value>
    </property>
    <!--
    有数据倾斜的时候进行负载均衡;
    group by 操作是否允许数据倾斜, 默认是false, 当设置为true时, 执行计划会生成两个 Map / Reduce 作业, 第一个 MapReduce 会将 Map 的结果随机分发到 Reduce 中, 达到负载均衡的目的来解决数据倾斜.
    -->
    <property>
        <name>hive.groupby.skewindata</name>
        <value>true</value>
    </property>
    <!-- 列裁剪, 默认为 true, 在做查询时只读取用到的列. -->
    <property>
        <name>hive.optimize.cp</name>
        <value>true</value>
    </property>
    <!-- JVM 重用 -->
    <property>
        <name>mapreduce.job.jvm.numTasks</name>
        <value>n</value>
        <description>${n} Tasks to run per JVM. -1 for unlimited.</description>
    </property>
</configuration>

HIVE CLI 调整

// 合并 Block 减少 Task 数量
set ngmr.partition.automerge = true;
// JVM 重用
set mapreduce.job.jvm.numTasks=n;
// 将 n 个 Block 排给单个线程处理
set ngmr.partition.mergesize.mb = n;
// 小文件合并
set hive.merge.sparkfiles = true;
set hive.map.agg = true;
// 使用向量化查询
set hive.vectorized.execution.enabled = true;
// cbo 优化 HIVE 查询
set hive.cbo.enable = true;
set hive.stats.fetch.column.stats = true;
set hive.stats.fetch.partition.stats = true;
set hive.compute.query.using.stats = true;
// 数据压缩
set hive.exec.compress.intermediate = true;
set hive.exec.compress.output = true;
// 有数据倾斜的时候进行负载均衡; group by 操作是否允许数据倾斜, 默认是false, 当设置为true时, 执行计划会生成两个 Map / Reduce 作业, 第一个 MapReduce 会将 Map 的结果随机分发到 Reduce 中, 达到负载均衡的目的来解决数据倾斜.
set hive.groupby.skewindata =  true;
// 列裁剪, 默认为 true, 在做查询时只读取用到的列.
set hive.optimize.cp = true;

HIVE 压缩

HIVE 数据压缩

压缩配置参数

要在 Hadoop 中使用压缩, 在 mapred-site.xml 中配置如下:

<configuration>
    <!-- something else... -->
    <property>
        <name>mapreduce.map.output.compress</name>
        <value>true</value>
        <description>Compress when Mapper output, true for enable.</description>
    </property>
    <property>
        <name>mapreduce.map.output.compress.codec</name>
        <value>org.apache.hadoop.io.compress.${iAmCodecName}</value>
        <description>Codec with LZO, LZ4, snappy.</description>
    </property>
    <property>
        <name>mapreduce.output.fileoutputformat.compress</name>
        <value>true</value>
        <description>Compress when Reducer output, true for enable.</description>
    </property>
    <property>
        <name>mapreduce.output.fileoutputformat.compress.codec</name>
        <value>org.apache.hadoop.io.compress.${iAmCodecName}</value>
        <description>Codec with LZO, LZ4, snappy.</description>
    </property>
    <property>
        <name>mapreduce.output.fileoutputformat.compress.type</name>
        <value>RECORD</value>
        <description>Compress logger</description>
    </property>
</configuration>

在 core-site.xml 中配置如下:

<configuration>
    <property>
        <name>io.compression.codecs</name>
        <!-- <value>org.apache.hadoop.io.compress.DefaultCodec</value> -->
        <!-- <value>org.apache.hadoop.io.compress.GzipCodec</value> -->
        <!-- <value>org.apache.hadoop.io.compress.BZip2Codec</value> -->
        <value>org.apache.hadoop.io.compress.Lz4Codec</value>
        <description>输入压缩阶段; Hadoop 使用文件扩展名判断是否支持某种编解码器</description>
    </property>
</configuration>

HIVE 文件压缩

HIVE 支持的存储数的格式主要有 TEXTFILE (行式存储), SEQUENCEFILE (行式存储), ORC (列式存储), PARQUET (列式存储).

存储文件的查询速度总结: ORC > TextFile > Parquet