logo头像
Snippet 博客主题

Hive-简易教程

其它 2022/03/05

数据分析篇

Hive是一个HDFS上的sql执行引擎，它将sql语句转化为Hadoop上的map-reduce任务来执行。由于是写sql，所以使用Hive进行数据分析的好处是没有什么额外的学习成本，但是它是批量式处理的，可能会比较慢。本文将通过几个案例来简单介绍如何使用Hive。


样例数据

随机生成一批订单数据(order_id, price, tag, order_date)

1
2
3
4
5
6
7
8
9
10
11
12
from random import randint
from datetime import date
from datetime import timedelta

for i in range(1000):
    order_id = 'order_%s' % i
    seller_id = 'seller_%s' % randint(0, 300)
    price = randint(0, 100000) / 100.0
    tag = randint(0, 1)
    order_date = date.today() - timedelta(days=randint(0, 30))
    print order_id, seller_id, price, tag, order_date


存储数据到Hive

1
2
3
4
5
6
7
hive> create table test_order_sample
(order_id string, seller_id string, price double, tag int, order_date string) 
row format delimited fields terminated by ' ';

hive> load data 
local inpath '/data/order_sample' 
into table test_order_sample;

案例一

统计出近一周每天成功支付的订单总数，gmv，客单价

1
2
3
4
5
6
hive> select order_date,count(*),round(sum(price),2),round(avg(price),2) 
from test_order_sample 
where tag=1 
and order_date>=date_sub(from_unixtime(unix_timestamp(),'yyyy-MM-dd'),7) 
group by order_date 
order by order_date desc;

案例二

统计出近一周每天成功支付 及支付失败 各自的订单总数，gmv，客单价

1
2
3
4
5
6
7
8
select order_date,
sum(if(tag=0,1,0)),sum(if(tag=0,price,0)),avg(if(tag=0,price,0)),
sum(if(tag=1,1,0)),sum(if(tag=1,price,0)),avg(if(tag=1,price,0)) 
from test_order_sample 
where tag=1 
and order_date>=date_sub(from_unixtime(unix_timestamp(),'yyyy-MM-dd'),7) 
group by order_date 
order by order_date desc;

count函数和if条件组合，而不是两个sql join


案例三

挑选出近一周gmv>1000并且订单量>2单的卖家ID及其订单

1
2
3
4
5
6
7
hive> select seller_id,collect_set(order_id) 
from test_order_sample 
where tag=1 
and order_date>=date_sub(from_unixtime(unix_timestamp(),'yyyy-MM-dd'),7) 
group by seller_id 
having count(*)>2 
and sum(price)>1000;

常用函数

聚合相关函数

collect_set(c_1)

在使用group by之后只能select出group key以及相关的统计数字，但也可以以集合的形式select出任何其他的非group key，比如按卖家ID聚合之后又想查看在这个卖家下单的买家ID：sellect collect_set(buyer_id) from t group by seller_id。

collect_list(c_1)

与collect_set类似，元素可重复

explode(c_1)

explode函数可以把一个array类型的数据扁平化。比如，现在每行是一个seller_id集合，使用explode可以扁平化为每行一个seller_id。但explode不可以直接与group by一起使用，比如我想按某些条件筛选一些卖家然后在查看该店铺的买家的情况：select explode(b.buyer_ids) from (select collect_set(buyer_id) as buyer_ids from t group by seller_id) b;


时间函数

unix_timestamp()

当前时间

from_unixtime(timestamp, format)

将系统时间戳转化为人可读的数据格式 如：select from_unixtime(unix_timestamp(), ‘yyyy-MM-dd’);

date_sub(string startdate, int days)

求几天前的日期


其它

nvl(v1, v2)

nvl函数用于处理null值，当一个字段是null时，这个字段和其它字段进行算术运算时的结果依然为null。这时可以使用这个函数为值可能为null的字段赋予一个默认值，即v2.

instr(str1, ‘xxx’)

判断字符串’xxx’是否出现在str1中，如果str1是null或者不存在xxx返回值都是0

size(a1)

返回数组a1的大小

union_all()

合并两个查询结果，但结果的列数需要一致！！！


自定义UDF篇

简介

实现自定义的UDF需要编写Java程序，然后在Hive客户端中加载相关Jar并注册函数后就可以使用了。


示例

实现转化IP地址为二进制格式

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
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

public final class IPUnMasker extends UDF {
    public Text evaluate(final Text s) {
        if (s == null) { return null; }

        StringBuilder ret = new StringBuilder();

        String[] items = s.toString().split("\\.");
        if (items.length != 4)
            return null;

        for (String item : items) {
            StringBuilder sb = new StringBuilder();
            int a = Integer.parseInt(item);
            for (int i=0; i<8; i++) {
                sb.insert(0, a%2);
                a = a/2;
            }
            ret.append(sb);
        }

        return new Text(ret.toString());
    }

    public static void main(String[] args) {
        String ip = "112.117.138.216";
        IPUnMasker unmasker = new IPUnMasker();
        System.out.println(unmasker.evaluate(new Text(ip)));
    }
}

打包

mvn package


加载

从本地加载
hive> add jar xxx.jar;

从hdfs加载并注册
create [temporary] function unmask_ip as ‘example.IPUnMasker’ using jar ‘hdfs:///user/zzz/jars/hive_udf-1.0.0.jar’;

删除函数
drop funciton unmask_ip;


使用udf

select unmask_ip(‘112.117.138.216’);


附录

pom依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-core</artifactId>
            <version>1.2.1</version>
        </dependency>

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

相关链接
https://cwiki.apache.org/confluence/display/Hive/HivePlugins


数据存储篇

hive是hdfs上的数据仓库，能够将一个个大文件有效地管理起来，并对其进行统计分析。数据仓库看待数据的方式与常见的数据库是完全不同的，它的最小粒度是文件而不是单条数据。hive一般在hdfs上的路径为/user/hive/warehouse，而hive中的数据库(如demo)的路径就是/user/hive/warehouse/demo.db，在往下就是表的路径如：/user/hive/warehouse/demo.db/test。管理hive上的数据分为两部分：定义数据以及操纵数据。


基本命令

切换数据库：use 数据库名;
显示当前数据库中的所有变：show tables;
查看表结构：desc 表名;
查看建表语句：show create table 表名;
数据定义语句(DDL)


创建表

完整语句

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name    -- (Note: TEMPORARY available in Hive 0.14.0 and later)
[(col_name data_type [COMMENT col_comment], ... [constraint_specification])]
[COMMENT table_comment]
[PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)]
[CLUSTERED BY (col_name, col_name, ...) [SORTED BY (col_name [ASC|DESC], ...)] INTO num_buckets BUCKETS]
[SKEWED BY (col_name, col_name, ...)                  -- (Note: Available in Hive 0.10.0 and later)]
ON ((col_value, col_value, ...), (col_value, col_value, ...), ...)
[STORED AS DIRECTORIES]
[
[ROW FORMAT row_format]
[STORED AS file_format]
| STORED BY 'storage.handler.class.name' [WITH SERDEPROPERTIES (...)]  -- (Note: Available in Hive 0.6.0 and later)
]
[LOCATION hdfs_path]
[TBLPROPERTIES (property_name=property_value, ...)]   -- (Note: Available in Hive 0.6.0 and later)
[AS select_statement];   -- (Note: Available in Hive 0.5.0 and later; not supported for external tables)
  1. hive可以创建临时表[temporary]，临时表在当前会话结束时(如关闭hive客户端)会被自动删除。
  2. hive可以外部的映射表[external]，即hive并不将这些数据存储在自己的hdfs路径中，而只保存表的元数据。这样hive便可访问更多的数据源，如hbase。
  3. hive可以直接将select语句的结果保存为一个表：create table t2 as select * from t1;
  4. hive的基本单元是文件，那么在创建表的时候可以指定文件中每行数据的分隔符，即按该分割符分割后即为表的各列：create table t(c1 string, c2 string) row format delimited fields terminated by ‘\t’。
  5. hive中的表就是hdfs上的一个路径，在创建表时可以在其下继续创建文件目录，这样的好处就是在查询时可以只访问一些指定的目录来提高性能。create table t1(id string) partitioned by (year string, month string)，这样t1路径下就是year的文件夹，而year的每个值路径下还有month文件夹，如图所示：


修改表

表重命名：alter table 原表名 rename to 新表名
删除分区：alter table xx drop partition (xx=’’)
数据操纵语言(DML)


加载数据

1）从本地加载： load data local inpath ‘data/t1’ into table t1;
2）从HDFS转移：load data inpath ‘/user/hive/project/data1’ into table xxx;


杂项篇

设置打印列名

set hive.cli.print.header=true;

设置动态分区

set hive.exec.dynamic.partition.mode=nonstrict;
hive cli默认的动态分区数量可能小于实际的分区数量，这时需要更改参数：
set hive.exec.dynamic.partitions=xxx;
set hive.exec.max.dynamic.partitions.pernode=xxx;

查看hive日志

hive -hiveconf hive.root.logger=INFO,console

以脚本方式执行并传递参数

test.sql:
select distinct seller_id from order where order_date=’${hiveconf:ORDER_DATE}’

hive -hiveconf ORDER_DATE=”xxx” -f test.sql

客户端中设置参数

set para=xxx
select a from b where c>${hiveconf:xxx}

客户端中执行hadoop命令

hive> dfs -ls /

评论系统未开启，无法评论！

Copyright © 2022 | Powered by Hexo | Theme by Snippet 京ICP备2022004959号