Hive窗口函数

2022-06-08 22:05:52 阅读：175 来源： 互联网

标签：窗口函数 over hive score Hive subject select name

1.1 Hive窗口函数

普通的聚合函数每组(Group by)只返回一个值，而开窗函数则可为窗口中的每行都返回一个值。简单理解，就是对查询的结果多出一列，这一列可以是聚合值，也可以是排序值。开窗函数一般就是说的是over（）函数，其窗口是由一个 OVER 子句定义的多行记录开窗函数一般分为两类,聚合开窗函数和排序开窗函数。

-- 聚合格式
select sum(字段名) over([partition by 字段名] [ order by 字段名]) as 别名,
    max(字段名) over() as 别名 
from 表名;

-- 排序窗口格式
select rank() over([partition by 字段名] [ order by 字段名]) as 别名 from 表名;

注意点：

over()函数中的分区、排序、指定窗口范围可组合使用也可以不指定，根据不同的业务需求结合使用
over()函数中如果不指定分区，窗口大小是针对查询产生的所有数据，如果指定了分区，窗口大小是针对每个分区的数据

测试数据

-- 创建表
create table t_fraction(
name string,
subject string, 
score int) 
row format delimited fields terminated by ","
lines terminated by '\n';

-- 测试数据 fraction.txt
孙悟空,语文,10
孙悟空,数学,73
孙悟空,英语,15
猪八戒,语文,10
猪八戒,数学,73
猪八戒,英语,11
沙悟净,语文,22
沙悟净,数学,70
沙悟净,英语,31
唐玄奘,语文,21
唐玄奘,数学,81
唐玄奘,英语,23

-- 上传数据
load data local inpath '/usr/local/soft/bigdata17/xiaohu/data/fraction.txt' into table t_fraction;

1.1.1 聚合开窗函数

sum(求和)

min(最小)

max(最大)

avg(平均值)

count(计数)

lag(获取当前行上一行的数据）

-- 
select name,subject,score,sum(score) over() as sumover from t_fraction;
+-------+----------+--------+----------+
| name  | subject  | score  | sumover  |
+-------+----------+--------+----------+
| 唐玄奘   | 英语       | 23     | 321      |
| 唐玄奘   | 数学       | 81     | 321      |
| 唐玄奘   | 语文       | 21     | 321      |
| 沙悟净   | 英语       | 31     | 321      |
| 沙悟净   | 数学       | 12     | 321      |
| 沙悟净   | 语文       | 22     | 321      |
| 猪八戒   | 英语       | 11     | 321      |
| 猪八戒   | 数学       | 73     | 321      |
| 猪八戒   | 语文       | 10     | 321      |
| 孙悟空   | 英语       | 15     | 321      |
| 孙悟空   | 数学       | 12     | 321      |
| 孙悟空   | 语文       | 10     | 321      |
+-------+----------+--------+----------+

select name,subject,score,
sum(score) over() as sum1,
sum(score) over(partition by subject) as sum2,
sum(score) over(partition by subject order by score) as sum3, 
-- 由起点到当前行的窗口聚合，和sum3一样
sum(score) over(partition by subject order by score rows between unbounded preceding and current row) as sum4, 
-- 当前行和前面一行的窗口聚合
sum(score) over(partition by subject order by score rows between 1 preceding and current row) as sum5,
-- 当前行的前面一行到后面一行的窗口聚合  前一行+当前行+后一行
sum(score) over(partition by subject order by score rows between 1 preceding and 1 following) as sum6,
-- 当前和后面所有的行
sum(score) over(partition by subject order by score rows between current row and unbounded following) as sum7
from t_fraction;

+-------+----------+--------+-------+-------+-------+-------+-------+-------+-------+
| name  | subject  | score  | sum1  | sum2  | sum3  | sum4  | sum5  | sum6  | sum7  |
+-------+----------+--------+-------+-------+-------+-------+-------+-------+-------+
| 孙悟空   | 数学       | 12     | 359   | 185   | 12    | 12    | 12    | 31    | 185   |
| 沙悟净   | 数学       | 19     | 359   | 185   | 31    | 31    | 31    | 104   | 173   |
| 猪八戒   | 数学       | 73     | 359   | 185   | 104   | 104   | 92    | 173   | 154   |
| 唐玄奘   | 数学       | 81     | 359   | 185   | 185   | 185   | 154   | 154   | 81    |
| 猪八戒   | 英语       | 11     | 359   | 80    | 11    | 11    | 11    | 26    | 80    |
| 孙悟空   | 英语       | 15     | 359   | 80    | 26    | 26    | 26    | 49    | 69    |
| 唐玄奘   | 英语       | 23     | 359   | 80    | 49    | 49    | 38    | 69    | 54    |
| 沙悟净   | 英语       | 31     | 359   | 80    | 80    | 80    | 54    | 54    | 31    |
| 孙悟空   | 语文       | 10     | 359   | 94    | 10    | 10    | 10    | 31    | 94    |
| 唐玄奘   | 语文       | 21     | 359   | 94    | 31    | 31    | 31    | 53    | 84    |
| 沙悟净   | 语文       | 22     | 359   | 94    | 53    | 53    | 43    | 84    | 63    |
| 猪八戒   | 语文       | 41     | 359   | 94    | 94    | 94    | 63    | 63    | 41    |
+-------+----------+--------+-------+-------+-------+-------+-------+-------+-------+

rows必须跟在Order by 子句之后，对排序的结果进行限制，使用固定的行数来限制分区中的数据行数量。

OVER()：指定分析函数工作的数据窗口大小，这个数据窗口大小可能会随着行的变而变化。

CURRENT ROW：当前行

n PRECEDING：往前n行数据

n FOLLOWING：往后n行数据

UNBOUNDED：起点，UNBOUNDED PRECEDING 表示从前面的起点， UNBOUNDED FOLLOWING表示到后面的终点

LAG(col,n,default_val)：往前第n行数据，col是列名，n是往上的行数，当第n行为null的时候取default_val

LEAD(col,n, default_val)：往后第n行数据，col是列名，n是往下的行数，当第n行为null的时候取default_val

NTILE(n)：把有序分区中的行分发到指定数据的组中，各个组有编号，编号从1开始，对于每一行，NTILE返回此行所属的组的编号。

cume_dist（），计算某个窗口或分区中某个值的累积分布。假定升序排序，则使用以下公式确定累积分布：

小于等于当前值x的行数 / 窗口或partition分区内的总行数。其中，x 等于 order by 子句中指定的列的当前行中的值。

聚合开窗函数实战：

实战1：Hive用户购买明细数据分析

创建表和加载数据

name，orderdate，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


建表加载数据
vim business.txt

create table business
(
name string, 
orderdate string,
cost int
)ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

load data local inpath "/shujia/bigdata17/xiaohu/data/business.txt" into table business;

实战1需求：

需求1：查询在2017年4月份购买过的顾客及总人数

# 分析：按照日期过滤、分组count求总人数

select name,orderdate,cost,count(*) over() total_people from business where date_format(orderdate,'yyyy-MM')='2017-04';

需求2：查询顾客的购买明细及月购买总额

# 分析：按照顾客分组、sum购买金额

select name,orderdate,cost,sum(cost) over(partition by name) total_amount from business;

需求3：上述的场景,要将cost按照日期进行累加

# 分析：按照顾客分组、日期升序排序、组内每条数据将之前的金额累加

select name,orderdate,cost,sum(cost) over(partition by name order by orderdate rows between unbounded preceding and current row) cumulative_amountfrom business;

需求4：查询顾客上次的购买时间

# 分析：查询出明细数据同时获取上一条数据的购买时间(肯定需要按照顾客分组、时间升序排序)

select name,orderdate,cost,lag(orderdate,1) over(partition by name order by orderdate) last_date from business;

需求5：查询前20%时间的订单信息

分析：按照日期升序排序、取前20%的数据

select * from (select name,orderdate,cost,ntile(5) over(order by orderdate) sortgroup_num from business) t where t.sortgroup_num=1;

1.1.2 排序开窗函数

RANK() 排序相同时会重复，总数不会变
DENSE_RANK() 排序相同时会重复，总数会减少
ROW_NUMBER() 会根据顺序计算
PERCENT_RANK()计算给定行的百分比排名。可以用来计算超过了百分之多少的人(当前行的rank值-1)/(分组内的总行数-1)

select name,subject,score,
rank() over(partition by subject order by score desc) rp,
dense_rank() over(partition by subject order by score desc) drp,
row_number() over(partition by subject order by score desc) rnp
from t_fraction;

select name,subject,score,
rank() over(order by score) as row_number,
percent_rank() over(partition by subject order by score) as percent_rank
from t_fraction;

实战2：Hive分析学生成绩信息

创建表语加载数据

name    subject score
李毅  语文  87
李毅  数学  95
李毅  英语  68
黄仙  语文  94
黄仙  数学  56
黄仙  英语  84
小虎  语文  64
小虎  数学  86
小虎  英语  84
许文客 语文  65
许文客 数学  85
许文客 英语  78

建表加载数据
vim score.txt

create table score
(
name string,
subject string, 
score int
) row format delimited fields terminated by "\t";

load data local inpath '/shujia/bigdata17/xiaohu/data/score.txt' into table score;

需求1：每门学科学生成绩排名(是否并列排名、空位排名三种实现)

分析：学科分组、成绩降序排序、按照成绩排名

select name,subject,score,
rank() over(partition by subject order by score desc) rp,
dense_rank() over(partition by subject order by score desc) drp,
row_number() over(partition by subject order by score desc) rmp
from 
score;

需求2：每门学科成绩排名top n的学生

select * from ( select name,subject,score,row_number() over(partition by subject order by score desc) rmp from score) t
where t.rmp<=3;

Hive 行转列

lateral view explode

create table testArray2(
    name string,
    weight array<string>
)row format delimited 
fields terminated by '\t'
COLLECTION ITEMS terminated by ',';

小虎  "150","170","180"
火火  "150","180","190"



select name,col1  from testarray2 lateral view explode(weight) t1 as col1;

小虎  150
小虎  170
小虎  180
火火  150
火火  180
火火  190

select key from (select explode(map('key1',1,'key2',2,'key3',3)) as (key,value)) t;

key1
key2
key3

select name,col1,col2  from testarray2 lateral view explode(map('key1',1,'key2',2,'key3',3)) t1 as col1,col2;
小虎  key1    1
小虎  key2    2
小虎  key3    3
火火  key1    1
火火  key2    2
火火  key3    3


select name,pos,col1  from testarray2 lateral view posexplode(weight) t1 as pos,col1;

小虎  0   150
小虎  1   170
小虎  2   180
火火  0   150
火火  1   180
火火  2   190

Hive 列转行

// testLieToLine
name col1
小虎  150
小虎  170
小虎  180
火火  150
火火  180
火火  190

create table testLieToLine(
    name string,
    col1 int
)row format delimited 
fields terminated by '\t';


select name,collect_list(col1) from testLieToLine group by name;

// 结果
小虎  ["150","180","190"]
火火  ["150","170","180"]

select  t1.name
        ,collect_list(t1.col1) 
from (
    select  name
            ,col1 
    from testarray2 
    lateral view explode(weight) t1 as col1
) t1 group by t1.name;

Hive自定义函数UserDefineFunction

UDF：一进一出

定义UDF函数要注意下面几点:

继承org.apache.hadoop.hive.ql.exec.UDF

重写evaluate()，这个方法不是由接口定义的,因为它可接受的参数的个数,数据类型都是不确定的。Hive会检查UDF,看能否找到和函数调用相匹配的evaluate()方法

创建maven项目，并加入依赖

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

编写代码，继承org.apache.hadoop.hive.ql.exec.UDF，实现evaluate方法，在evaluate方法中实现自己的逻辑

import org.apache.hadoop.hive.ql.exec.UDF;

public class HiveUDF extends UDF {
    // hadoop => #hadoop$
    public String evaluate(String col1) {
    // 给传进来的数据 左边加上 # 号 右边加上 $
        String result = "#" + col1 + "$";
        return result;
    }
}

打成jar包并上传至Linux虚拟机
在hive shell中，使用 add jar 路径将jar包作为资源添加到hive环境中

add jar /usr/local/soft/jars/HiveUDF2-1.0.jar;

使用jar包资源注册一个临时函数，fxxx1是你的函数名，'MyUDF'是主类名

create temporary function fxxx1 as 'MyUDF';

使用函数名处理数据

select fxx1(name) as fxx_name from students limit 10;

#施笑槐$
#吕金鹏$
#单乐蕊$
#葛德曜$
#宣谷芹$
#边昂雄$
#尚孤风$
#符半双$
#沈德昌$
#羿彦昌$

案例2：转大写

public class FirstUDF extends UDF {
    public String evaluate(String str){
        String upper = null;
        //1、检查输入参数
        if (StringUtils.isEmpty(str)){
        } else {
            upper = str.toUpperCase();
        }
        return upper;
    }
    //调试自定义函数
    public static void main(String[] args){
        System.out.println(new firstUDF().evaluate("jiajingwen"));
    }

函数加载方式

命令加载

这种加载只对本session有效

# 1、将项目打包上传服务器：将打好的jar包传到linux系统中。(不要打依赖)
# 进入到hive客户端,执行下面命令
hive> add jar /usr/local/soft/bigdata17/data/xiaohu/hadoop-mapreduce-1.0-SNAPSHOT.jar
# 2、创建一个临时函数名,要跟上面hive在同一个session里面：
hive> create temporary function toUP as 'com.shujia.testHiveFun.udf.FirstUDF';

3、检查函数是否创建成功
show functions;

4. 测试功能
select toUp('abcdef');

5. 删除函数 
drop temporary function if exists toUp;

创建永久函数

将jar上传HDFS：

hadoop fs -put hadoop-mapreduce-1.0-SNAPSHOT.jar /jar/

在hive命令行中创建永久函数：

create function myUp as 'com.shujia.testHiveFun.udf.FirstUDF' using jar 'hdfs:/jar/hadoop-mapreduce-1.0-SNAPSHOT.jar';

退出hive，再进入，执行测试：

删除永久函数，并检查：

UDTF：一进多出

UDTF是一对多的输入输出,实现UDTF需要完成下面步骤

继承org.apache.hadoop.hive.ql.udf.generic.GenericUDF，重写initlizer（）、getdisplay（）、evaluate()。执行流程如下:

UDTF首先会调用initialize方法，此方法返回UDTF的返回行的信息（返回个数，类型）。

初始化完成后，会调用process方法,真正的处理过程在process函数中，在process中，每一次forward()调用产生一行；如果产生多列可以将多个列的值放在一个数组中，然后将该数组传入到forward()函数。

最后close()方法调用，对需要清理的方法进行清理。

"key1:value1,key2:value2,key3:value3"

key1 value1

key2 value2

key3 value3

方法一：使用 explode+split

select split(t.col1,":")[0],split(t.col1,":")[1] 
from (select explode(split("key1:value1,key2:value2,key3:value3",",")) as col1) t;

方法二：自定UDTF

代码

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;

public class HiveUDTF extends GenericUDTF {
    // 指定输出的列名 及 类型
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        ArrayList<String> filedNames = new ArrayList<String>();
        ArrayList<ObjectInspector> filedObj = new ArrayList<ObjectInspector>();
        filedNames.add("col1");
        filedObj.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        filedNames.add("col2");
        filedObj.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(filedNames, filedObj);
    }

    // 处理逻辑 my_udtf(col1,col2,col3)
    // "key1:value1,key2:value2,key3:value3"
    // my_udtf("key1:value1,key2:value2,key3:value3")
    public void process(Object[] objects) throws HiveException {
        // objects 表示传入的N列
        String col = objects[0].toString();
        // key1:value1  key2:value2  key3:value3
        String[] splits = col.split(",");
        for (String str : splits) {
            String[] cols = str.split(":");
            // 将数据输出
            forward(cols);
        }

    }

    // 在UDTF结束时调用
    public void close() throws HiveException {

    }
}

select my_udtf("key1:value1,key2:value2,key3:value3");

字段：id,col1,col2,col3,col4,col5,col6,col7,col8,col9,col10,col11,col12 共13列

数据：

a,1,2,3,4,5,6,7,8,9,10,11,12

b,11,12,13,14,15,16,17,18,19,20,21,22

c,21,22,23,24,25,26,27,28,29,30,31,32

转成3列：id,hours,value

例如：

a,1,2,3,4,5,6,7,8,9,10,11,12

a,0时,1

a,2时,2

a,4时,3

a,6时,4

......

create table udtfData(
    id string
    ,col1 string
    ,col2 string
    ,col3 string
    ,col4 string
    ,col5 string
    ,col6 string
    ,col7 string
    ,col8 string
    ,col9 string
    ,col10 string
    ,col11 string
    ,col12 string
)row format delimited fields terminated by ',';

代码：

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;

public class HiveUDTF2 extends GenericUDTF {
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        ArrayList<String> filedNames = new ArrayList<String>();
        ArrayList<ObjectInspector> fieldObj = new ArrayList<ObjectInspector>();
        filedNames.add("col1");
        fieldObj.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        filedNames.add("col2");
        fieldObj.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(filedNames, fieldObj);
    }

    public void process(Object[] objects) throws HiveException {
        int hours = 0;
        for (Object obj : objects) {
            hours = hours + 1;
            String col = obj.toString();
            ArrayList<String> cols = new ArrayList<String>();
            cols.add(hours + "时");
            cols.add(col);
            forward(cols);
        }
    }

    public void close() throws HiveException {

    }
}

添加jar资源:

add jar /usr/local/soft/HiveUDF2-1.0.jar;

注册udtf函数：

create temporary function my_udtf as 'MyUDTF';

SQL:

select id,hours,value from udtfData lateral view my_udtf(col1,col2,col3,col4,col5,col6,col7,col8,col9,col10,col11,col12) t as hours,value ;

UDAF：多进一出

Hive Shell

第一种：

hive -e "select * from test1.students limit 10"

第二种：

hive -f hql文件路径

将HQL写在一个文件里，再使用 -f 参数指定该文件

标签：窗口,函数,over,hive,score,Hive,subject,select,name
来源： https://www.cnblogs.com/Longtianm/p/16357359.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9