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之自定义函数压缩和存储存储和压缩结合的主要内容,如果未能解决你的问题,请参考以下文章