hive3之自定义函数压缩和存储存储和压缩结合

Posted 爱上口袋的天空

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了hive3之自定义函数压缩和存储存储和压缩结合相关的知识,希望对你有一定的参考价值。

一、自定义函数

1、简介

        Hive 自带了一些函数,比如:max/min 等,但是数量有限,自己可以通过自定义 UDF 来 方便的扩展。

        当 Hive 提供的内置函数无法满足你的业务处理需要时,此时就可以考虑使用用户自定义 函数(UDF:user-defined function)。

2、根据用户自定义函数类别分为以下三种:

        (1)UDF(User-Defined-Function)

                一进一出

        (2)UDAF(User-Defined Aggregation Function)

                聚集函数,多进一出 类似于:count/max/min

        (3)UDTF(User-Defined Table-Generating Functions)

                一进多出 如 lateral view explode()

3、编程步骤

      (1)继承 Hive 提供的类

                org.apache.hadoop.hive.ql.udf.generic.GenericUDF                 org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;

     (2)实现类中的抽象方法

     (3)在 hive 的命令行窗口创建函数         

          添加 jar:add jar linux_jar_path

          创建 function:create [temporary] function [dbname.]function_name AS class_name;

     (4)在 hive 的命令行窗口删除函数 

           drop [temporary] function [if exists] [dbname.]function_name;     

二、自定义 UDF 函数

1、需求

        自定义一个 UDF 实现计算给定字符串的长度,例如:

        hive(default)> select my_len("abcd");

        4

2、创建一个 Maven 工程 Hive

        

3、导入依赖

<dependencies>
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-exec</artifactId>
        <version>3.1.2</version>
    </dependency>
</dependencies>

4、创建一个类

package com.kgf.hive;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

/***
 * 自定义一个UDF函数,计算给定字符的长度
 */
public class MyStringLength extends GenericUDF 

    /***
     *
     * @param objectInspectors 输入参数类型的鉴别器对象
     * @return 返回值类型的鉴别器对象
     * @throws UDFArgumentException
     */
    public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException 
        // 判断输入参数的个数
        if(objectInspectors.length !=1)
            throw new UDFArgumentLengthException("Input Args LengthError!!!");
        
        // 判断输入参数的类型,一个枚举类Category,定义了5种类型:基本类型(Primitive),集合(List),键值对映射(Map),结构体(Struct),联合体(Union)。
        if(!objectInspectors[0].getCategory().equals(ObjectInspector.Category.PRIMITIVE))
            throw new UDFArgumentTypeException(0,"Input Args TypeError!!!");
        
        //函数本身返回值为 int,需要返回 int 类型的鉴别器对象
        return PrimitiveObjectInspectorFactory.javaIntObjectInspector;
    

    /***
     * 函数的逻辑处理
     * @param deferredObjects:输入的参数
     * @return
     * @throws HiveException
     */
    public Object evaluate(DeferredObject[] deferredObjects) throws HiveException 
        if(deferredObjects[0].get() == null)
            return 0;
        
        return deferredObjects[0].get().toString().length();
    

    public String getDisplayString(String[] strings) 
        return "";
    

5、打成 jar 包上传到服务器/opt/module/data/hive-demo-1.0-SNAPSHOT.jar

6、将 jar 包添加到 hive 的 classpath

hive (default)> add jar /opt/module/data/hive-demo-1.0-SNAPSHOT.jar
              > ;
Added [/opt/module/data/hive-demo-1.0-SNAPSHOT.jar] to class path
Added resources: [/opt/module/data/hive-demo-1.0-SNAPSHOT.jar]
hive (default)>

7、创建临时函数与开发好的 java class 关联

hive (default)> create temporary function my_len as "com.kgf.hive.MyStringLength";
OK
Time taken: 0.451 seconds
hive (default)>

8、即可在 hql 中使用自定义的函数

hive (default)> select * from business;
OK
business.name   business.orderdate      business.cost
jack    2017-01-01      10
tony    2017-01-02      15
jack    2017-02-03      23
tony    2017-01-04      29
jack    2017-01-05      46
jack    2017-04-06      42
tony    2017-01-07      50
jack    2017-01-08      55
mart    2017-04-08      62
mart    2017-04-09      68
neil    2017-05-10      12
mart    2017-04-11      75
neil    2017-06-12      80
mart    2017-04-13      94
Time taken: 1.759 seconds, Fetched: 14 row(s)
hive (default)> select my_len(name) from business;
OK
_c0
4
4
4
4
4
4
4
4
4
4
4
4
4
4
Time taken: 0.345 seconds, Fetched: 14 row(s)
hive (default)>

三、自定义 UDTF 函数

1、需求

        自定义一个 UDTF 实现将一个任意分割符的字符串切割成独立的单词,例如:

hive(default)> select myudtf("hello,world,hadoop,hive", ",");
hello
world
hadoop
hive

2、代码实现

package com.kgf.hive;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

/***
 * 自定义一个UDTF函数
 */
public class MyUDTF extends GenericUDTF 

    private ArrayList<String> outList = new ArrayList<>();

    /***
     *
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException 
        //1.定义输出数据的列名和类型
        List<String> fieldNames = new ArrayList<>();
        List<ObjectInspector> fieldOIs = new ArrayList<>();

        //2.添加输出数据的列名和类型
        fieldNames.add("lineToWord");

        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);
    

    /***
     * 处理数据
     * @param objects
     * @throws HiveException
     */
    public void process(Object[] objects) throws HiveException 
        //1.获取原始数据
        String arg = objects[0].toString();
        //2.获取数据传入的第二个参数,此处为分隔符
        String splitKey = objects[1].toString();
        //3.将原始数据按照传入的分隔符进行切分
        String[] fields = arg.split(splitKey);
        //4.遍历切分后的结果,并写出
        for (String field : fields) 
            //集合为复用的,首先清空集合
            outList.clear();
            //将每一个单词添加至集合
            outList.add(field);
            //将集合内容写出
            forward(outList);
        
    

    public void close() throws HiveException 

    

3、打成 jar 包上传到服务器/opt/module/hive/data/myudtf.jar

4、将 jar 包添加到 hive 的 classpath 下

hive (default)> add jar /opt/module/data/myudtf.jar;
Added [/opt/module/data/myudtf.jar] to class path
Added resources: [/opt/module/data/myudtf.jar]
hive (default)>

5、创建临时函数与开发好的 java class 关联

hive (default)> create temporary function myudtf as "com.kgf.hive.MyUDTF";
OK
Time taken: 0.011 seconds
hive (default)>

6、使用自定义的函数

hive (default)> select myudtf("hello,world,hadoop,hive",",");
OK
linetoword
hello
world
hadoop
hive
Time taken: 0.956 seconds, Fetched: 4 row(s)
hive (default)>

四、压缩和存储

1、Hadoop 压缩配置

1.1、MR 支持的压缩编码

 

 1.2、压缩参数配置

要在 Hadoop 中启用压缩,可以配置如下参数(mapred-site.xml 文件中):


2、开启 Map 输出阶段压缩(MR 引擎) 

开启 map 输出阶段压缩可以减少 job 中 map 和 Reduce task 间数据传输量。具体配置如 下:

1)案例实操:

        (1)开启 hive 中间传输数据压缩功能

hive (default)>set hive.exec.compress.intermediate=true;

        (2)开启 mapreduce 中 map 输出压缩功能

hive (default)>set mapreduce.map.output.compress=true;

        (3)设置 mapreduce 中 map 输出数据的压缩方式

hive (default)>set mapreduce.map.output.compress.codec=
org.apache.hadoop.io.compress.SnappyCodec;

        (4)执行查询语句

hive (default)> select count(ename) name from emp;

3、开启 Reduce 输出阶段压缩

        当 Hive 将 输 出 写 入 到 表 中 时 , 输出内容同样可以进行压缩。属性 hive.exec.compress.output控制着这个功能。用户可能需要保持默认设置文件中的默认值false, 这样默认的输出就是非压缩的纯文本文件了。用户可以通过在查询语句或执行脚本中设置这 个值为 true,来开启输出结果压缩功能。

1)案例实操:

(1)开启 hive 最终输出数据压缩功能

hive (default)>set hive.exec.compress.output=true;

(2)开启 mapreduce 最终输出数据压缩

hive (default)>set mapreduce.output.fileoutputformat.compress=true;

(3)设置 mapreduce 最终数据输出压缩方式

hive (default)> set mapreduce.output.fileoutputformat.compress.codec =
org.apache.hadoop.io.compress.SnappyCodec;

(4)设置 mapreduce 最终数据输出压缩为块压缩

hive (default)> set 
mapreduce.output.fileoutputformat.compress.type=BLOCK;

(5)测试一下输出结果是否是压缩文件

hive (default)> insert overwrite local directory
'/opt/module/data/distribute-result' select * from emp distribute by 
deptno sort by empno desc;

4、文件存储格式

Hive 支持的存储数据的格式主要有:TEXTFILE 、SEQUENCEFILE、ORC、PARQUET。

4.1、列式存储和行式存储

 如图所示左边为逻辑表,右边第一个为行式存储,第二个为列式存储

1)行存储的特点

        查询满足条件的一整行数据的时候,列存储则需要去每个聚集的字段找到对应的每个列 的值,行存储只需要找到其中一个值,其余的值都在相邻地方,所以此时行存储查询的速度 更快。

2)列存储的特点

        因为每个字段的数据聚集存储,在查询只需要少数几个字段的时候,能大大减少读取的 数据量;每个字段的数据类型一定是相同的,列式存储可以针对性的设计更好的设计压缩算 法。

TEXTFILE 和 SEQUENCEFILE 的存储格式都是基于行存储的;

ORC 和 PARQUET 是基于列式存储的。

5、TextFile 格式

        默认格式,数据不做压缩,磁盘开销大,数据解析开销大。可结合 Gzip、Bzip2 使用, 但使用 Gzip 这种方式,hive 不会对数据进行切分,从而无法对数据进行并行操作。

6、Orc 格式

        Orc (Optimized Row Columnar)是 Hive 0.11 版里引入的新的存储格式。

        如下图所示可以看到每个 Orc 文件由 1 个或多个 stripe 组成,每个 stripe 一般为 HDFS 的块大小,每一个 stripe 包含多条记录,这些记录按照列进行独立存储,对应到 Parquet 中的 row group 的概念。每个 Stripe 里有三部分组成,分别是 Index Data,Row Data,Stripe Footer:

1)Index Data:一个轻量级的 index,默认是每隔 1W 行做一个索引。这里做的索引应该 只是记录某行的各字段在 Row Data 中的 offset。

 2)Row Data:存的是具体的数据,先取部分行,然后对这些行按列进行存储。对每个 列进行了编码,分成多个 Stream 来存储。

3)Stripe Footer:存的是各个 Stream 的类型,长度等信息。

         每个文件有一个 File Footer,这里面存的是每个 Stripe 的行数,每个 Column 的数据类 型信息等;每个文件的尾部是一个 PostScript,这里面记录了整个文件的压缩类型以及 FileFooter 的长度信息等。在读取文件时,会 seek 到文件尾部读 PostScript,从里面解析到 File Footer 长度,再读 FileFooter,从里面解析到各个 Stripe 信息,再读各个 Stripe,即从后 往前读


7、Parquet 格式

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

(1)行组(Row Group):每一个行组包含一定的行数,在一个 HDFS 文件中至少存储一 个行组,类似于 orc 的 stripe 的概念。

(2)列块(Column Chunk):在一个行组中每一列保存在一个列块中,行组中的所有列连 续的存储在这个行组文件中。一个列块中的值都是相同类型的,不同的列块可能使用不同的 算法进行压缩。

(3)页(Page):每一个列块划分为多个页,一个页是最小的编码的单位,在同一个列块 的不同页可能使用不同的编码方式。

通常情况下,在存储 Parquet 数据的时候会按照 Block 大小设置行组的大小,由于一般 情况下每一个 Mapper 任务处理数据的最小单位是一个 Block,这样可以把每一个行组由一 个 Mapper 任务处理,增大任务执行并行度。Parquet 文件的格式。

        上图展示了一个 Parquet 文件的内容,一个文件中可以存储多个行组,文件的首位都是 该文件的 Magic Code,用于校验它是否是一个 Parquet 文件,Footer length 记录了文件元数据的大小,通过该值和文件长度可以计算出元数据的偏移量,文件的元数据中包括每一个行 组的元数据信息和该文件存储数据的 Schema 信息。除了文件中每一个行组的元数据,每一 页的开始都会存储该页的元数据,在 Parquet 中,有三种类型的页:数据页、字典页和索引 页。数据页用于存储当前行组中该列的值,字典页存储该列值的编码字典,每一个列块中最 多包含一个字典页,索引页用来存储当前行组下该列的索引,目前 Parquet 中还不支持索引 页。 


7、主流文件存储格式对比实验

从存储文件的压缩比和查询速度两个角度对比。 存储文件的压缩比测试

1)测试数据,log.data文件中数据样例,大概18M左右

2017-08-10 13:00:00	http://www.taobao.com/17/?tracker_u=1624169&type=1	B58W48U4WKZCJ5D1T3Z9ZY88RU7QA7B1	http://hao.360.cn/	1.196.34.243	NULL	-1
2017-08-10 13:00:00	http://www.taobao.com/item/962967_14?ref=1_1_52_search.ctg_1	T82C9WBFB1N8EW14YF2E2GY8AC9K5M5P	http://www.yihaodian.com/ctg/s2/c24566-%E5%B1%B1%E6%A5%82%E5%88%B6%E5%93%81?ref=pms_15_78_258	222.78.246.228	134939954	156
2017-08-10 13:00:00	http://www.taobao.com/1/?tracker_u=1013304189&uid=2687512&type=3	W17C89RU8DZ6NMN7JD2ZCBDMX1CQVZ1W	http://www.yihaodian.com/1/?tracker_u=1013304189&uid=2687512&type=3	118.205.0.18	NULL	-20

2)TextFile

 (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;

(2)向表中加载数据

hive (default)> load data local inpath '/opt/module/data/log.data' into table log_text;
Loading data to table default.log_text
OK
Time taken: 1.523 seconds
hive (default)>

(3)查看表中数据大小

hive (default)> dfs -du -h /user/hive/warehouse/log_text;
18.1 M  54.4 M  /user/hive/warehouse/log_text/log.data
hive (default)>

3)ORC

(1)创建表,存储数据格式为 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
tblproperties("orc.compress"="NONE"); -- 设置 orc 存储不使用压缩

(2)向表中加载数据

hive (default)> insert into table log_orc select * from log_text;

(3)查看表中数据大小

hive (default)>  dfs -du -h /user/hive/warehouse/log_orc/ ;
7.7 M  23.1 M  /user/hive/warehouse/log_orc/000000_3
hive (default)>

4)Parquet

   (1)创建表,存储数据格式为 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;

(2)向表中加载数据

 insert into table log_parquet select * from log_text;

(3)查看表中数据大小

hive (default)>  dfs -du -h /user/hive/warehouse/log_parquet/;
13.1 M  39.3 M  /user/hive/warehouse/log_parquet/000000_0
hive (default)>

存储文件的对比总结:

ORC > Parquet > textFile

存储文件的查询速度测试:

(1)TextFile

insert overwrite local directory 
'/opt/module/data/log_text' select substring(url,1,4) from log_text;

(2)ORC

hive (default)> insert overwrite local directory 
'/opt/module/data/log_orc' select substring(url,1,4) from log_orc;

(3)Parquet

hive (default)> insert overwrite local directory 
'/opt/module/data/log_parquet' select substring(url,1,4) from 
log_parquet;

存储文件的查询速度总结:查询速度相近

五、存储和压缩结合

1、测试存储和压缩

1)创建一个 ZLIB 压缩的 ORC 存储方式

(1)建表语句 

create table log_orc_zlib(
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"="ZLIB");

(2)插入数据

hive (default)> insert into log_orc_zlib select * from log_text;

(3)查看插入后数据

hive (default)> dfs -du -h /user/hive/warehouse/log_orc_zlib/ ;
2.8 M  8.3 M  /user/hive/warehouse/log_orc_zlib/000000_1
hive (default)>

2)创建一个 SNAPPY 压缩的 ORC 存储方式

(1)建表语句

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");

(2)插入数据

insert into log_orc_snappy select * from log_text;

(3)查看插入后数据

hive (default)> dfs -du -h /user/hive/warehouse/log_orc_snappy/;
3.7 M  11.2 M  /user/hive/warehouse/log_orc_snappy/000000_0
hive (default)>

ZLIB 比 Snappy 压缩的还小。原因是 ZLIB 采用的是 deflate 压缩算法。比 snappy 压缩的 压缩率高。

3)创建一个 SNAPPY 压缩的 parquet 存储方式

(1)建表语句

create table log_parquet_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 parquet
tblproperties("parquet.compression"="SNAPPY");

(2)插入数据

insert into log_parquet_snappy select * from log_text;

(3)查看插入后数据

hive (default)>  dfs -du -h /user/hive/warehouse/log_parquet_snappy/;
6.4 M  19.2 M  /user/hive/warehouse/log_parquet_snappy/000000_0
hive (default)>

4)存储方式和压缩总结

        在实际的项目开发当中,hive 表的数据存储格式一般选择:orc 或 parquet。压缩方式一 般选择 snappy,lzo。

以上是关于hive3之自定义函数压缩和存储存储和压缩结合的主要内容,如果未能解决你的问题,请参考以下文章

大数据技术之Hive函数压缩和存储

大数据技术之Hive函数压缩和存储

Hive生产上,压缩和存储结合使用案例

压缩和存储

Hive:第 9 章 压缩和存储

SQL入门经典 之自定义函数