hive之udf/udaf/udtf
程序员文章站
2022-04-29 08:51:21
...
https://blog.csdn.net/erliang20088/article/details/78685409
目录
1、udf
2、udaf
3、udtf
4、练习题
详情
1、udf(user defined function)
-
背景
- 系统内置函数无法解决实际的业务问题，需要开发者自己编写函数实现自身的业务实现诉求。
- 应用场景非常多，面临的业务不同导致个性化实现很多，故udf很需要。
-
意义
- 函数扩展得到解决，极大丰富了可定制化的业务需求。
- IO要求-要解决的问题
- in:out=1:1,只能输入一条记录当中的数据，同时返回一条处理结果。
- 属于最常见的自定义函数，像cos,sin,substring,indexof等均是如此要求
-
实现步骤(Java创建自定义UDF类)
- 自定义一个java类
- 继承UDF类
- 重写evaluate方法
- 打包类所在项目成一个all-in-one的jar包并上传到hive所在机器
- 在hive中执行add jar操作，将jar加载到classpath中。
- 在hive中创建模板函数，使得后边可以使用该函数名称调用实际的udf函数
- hive sql中像调用系统函数一样使用udf函数
-
代码实现
- 功能要求：实现当输入字符串超过2个字符的时候，多余的字符以”…”来表示。
- 如“12”则返回“12”，如“123”返回“12…”
- 自定义类、继承UDF、重写evaluate方法已在代码中体现
- 功能要求：实现当输入字符串超过2个字符的时候，多余的字符以”…”来表示。
package com.tianliangedu.hive.udf;
import org.apache.hadoop.hive.ql.exec.UDF;
/*
* 功能：实现当输入字符串超过2个字符的时候，多余的字符以"..."来表示。
* 输入/输出：* 如“12”则返回“12”，如“123”返回“12..."
*/
public class ValueMaskUDF extends UDF{
public String evaluate(String input,int maxSaveStringLength,String replaceSign) {
if(input.length()<=maxSaveStringLength){
return input;
}
return input.substring(0,maxSaveStringLength)+replaceSign;
}
public static void main(String[] args) {
System.out.println(new ValueMaskUDF().evaluate("河北省",2,"..."));;
}
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
-
布署步骤
-
打包并上传
- 均采用maven管理打包的方式，maven pom配置项为：
-
<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<modelVersion>4.0.0</modelVersion>
<groupId>com.tianliangedu.course</groupId>
<artifactId>TlHadoopCore</artifactId>
<version>0.0.1-SNAPSHOT</version>
<!-- 首先配置仓库的服务器位置,首选阿里云，也可以配置镜像方式，效果雷同 -->
<repositories>
<repository>
<id>nexus-aliyun</id>
<name>Nexus aliyun</name>
<url>http://maven.aliyun.com/nexus/content/groups/public</url>
</repository>
</repositories>
<dependencies>
<!-- 引入hadoop-cli-2.7.4依赖 -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.4</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-cli</artifactId>
<version>1.2.1</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<finalName>TlHadoopCore</finalName>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>assembly</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.3.2</version>
<configuration>
<source>1.7</source>
<target>1.7</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
</plugins>
</build>
</project>- 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
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
-
maven打包操作
-
上传jar包至hive操作环境中
- 进入到自己的所操作的hive环境目录中。
rz命令上传至服务器上
-
加载jar包、声明函数、使用函数
-
加载jar包
- 进入到hive cli中（输入hive即可进入）
-
-
将jar包加入hive 交互中
add jar的shell
add jar /home/hive/tianliangedu_course/04_udf/TlHadoopCore-jar-with-dependencies.jar;声明函数
create temporary function mask as ‘com.tianliangedu.hive.udf.ValueMaskUDF’;使用函数
-
2、udaf（user defined aggregation function）
-
自定义udaf函数self_count，实现系统udaf count的功能
-
Input/Output要求-要解决的问题
- in:out=n:1,即接受输入N条记录当中的数据，同时返回一条处理结果。
- 属于最常见的自定义函数，像count,sum,avg,max等均是如此要求
实现步骤
自定义一个java类
- 继承UDAF类
- 内部定义一个静态类，实现UDAFEvaluator接口
实现方法init,iterate,terminatePartial,merge,terminate，共5个方法.
在hive中执行add jar操作，将jar加载到classpath中。
- 在hive中创建模板函数，使得后边可以使用该函数名称调用实际的udf函数
- hive sql中像调用系统函数一样使用udaf函数
代码实现
功能要求：实现与hive原生的count相似的计数功能。
如select count(1) from tablename 或者select key,count(1) from tablename group by key;
- 源码
- in:out=n:1,即接受输入N条记录当中的数据，同时返回一条处理结果。
-
package com.tianliangedu.hive.udaf;
import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.apache.log4j.Logger;
/**
* 自行实现sql的count操作
*/
//主类继承UDAF
public class DIYCountUDAF extends UDAF {
//日志对象初始化,使访类有输出日志的能力
public static Logger logger=Logger.getLogger(DIYCountUDAF.class);
//静态类实现UDAFEvaluator
public static class Evaluator implements UDAFEvaluator {
//设置成员变量，存储每个统计范围内的总记录数
private int totalRecords;
//初始化函数,map和reduce均会执行该函数,起到初始化所需要的变量的作用
public Evaluator() {
init();
}
//初始化，初始值为0,并日志记录下相应输出
public void init() {
totalRecords = 0;
logger.info("init totalRecords="+totalRecords);
}
//map阶段，返回值为boolean类型，当为true则程序继续执行，当为false则程序退出
public boolean iterate(String input) {
//当input输入不为空的时候，即为有值存在,即为存在1行，故做+1操作
if (input != null) {
totalRecords += 1;
}
//输出当前组处理到第多少条数据了
logger.info("iterate totalRecords="+totalRecords);
return true;
}
/**
* 类似于combiner,在map范围内做部分聚合，将结果传给merge函数中的形参mapOutput
* 如果需要聚合，则对iterator返回的结果处理，否则直接返回iterator的结果即可
*/
public int terminatePartial() {
logger.info("terminatePartial totalRecords="+totalRecords);
return totalRecords;
}
// reduce 阶段，用于逐个迭代处理map当中每个不同key对应的 terminatePartial的结果
public boolean merge(int mapOutput) {
totalRecords +=mapOutput;
logger.info("merge totalRecords="+totalRecords);
return true;
}
//处理merge计算完成后的结果，此时的count在merge完成时候，结果已经得出，无需再进一次对整体结果做处理，故直接返回即可
public int terminate() {
logger.info("terminate totalRecords="+totalRecords);
return totalRecords;
}
}
} - 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
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
-
布署步骤
-
跟udf完全一致
-
加载jar包、声明函数、使用函数
上述三步与udf基本一致。
注意在使用时候是按跟count或sum一样的聚合函数去使用。 加载与声明
使用-全局self count
使用-分组self count
自定义udaf函数，实现多条学生成绩的合并
-
业务需求
- 数据输入
- 数据输入
数据输出
代码
-
-
package com.tianliangedu.hive.udaf;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.apache.log4j.Logger;
/**
* 实现多条数据合并成一条数据
*/
// 主类继承UDAF
public class StudentScoreAggUDAF extends UDAF {
// 日志对象初始化
public static Logger logger = Logger.getLogger(StudentScoreAggUDAF.class);
// 静态类实现UDAFEvaluator
public static class Evaluator implements UDAFEvaluator {
// 设置成员变量，存储每个统计范围内的总记录数
private Map<String, String> courseScoreMap;
//初始化函数,map和reduce均会执行该函数,起到初始化所需要的变量的作用
public Evaluator() {
init();
}
// 初始化函数间传递的中间变量
public void init() {
courseScoreMap = new HashMap<String, String>();
}
//map阶段，返回值为boolean类型，当为true则程序继续执行，当为false则程序退出
public boolean iterate(String course, String score) {
if (course == null || score == null) {
return true;
}
courseScoreMap.put(course, score);
return true;
}
/**
* 类似于combiner,在map范围内做部分聚合，将结果传给merge函数中的形参mapOutput
* 如果需要聚合，则对iterator返回的结果处理，否则直接返回iterator的结果即可
*/
public Map<String, String> terminatePartial() {
return courseScoreMap;
}
// reduce 阶段，用于逐个迭代处理map当中每个不同key对应的 terminatePartial的结果
public boolean merge(Map<String, String> mapOutput) {
this.courseScoreMap.putAll(mapOutput);
return true;
}
// 处理merge计算完成后的结果，即对merge完成后的结果做最后的业务处理
public String terminate() {
return courseScoreMap.toString();
}
}
}- 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
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
* 布署过程与之前相同
* 测试脚本
- 1
- 2
- 3
select id,username,score_agg(course,score) from student_score group by id,username;
3、udtf(User-Defined Table-Generating Functions)
-
要解决一行输入多行输出的问题，问题的应用场景不少
- 用udtf解决一行输入多行输出的不多，往往被lateral view explode+udf等替代实现，比直接用udtf会更简单、直接一些
- 有兴趣同学可以学习标准的udtf如何写
-
本节由学员自学实现，如何用lateral view explode+udf替代udtf实现
-
lateral view explode+udf替代udtf应用案例
- 需求：将一个array类型按列存储的学生成绩表，转变成按行来显示，学生名字超过2个字符的，后边用”…”来代替。
- 数据准备
- 学生成绩表
- 学生成绩表
-
通过lateral view explode实现行转列
select id,name,score from test_array lateral view explode(score_array) score_table as score;
- 加入udf处理业务需求
select id,mask(name,2,'...'),score from test_array lateral view explode(score_array) score_table as score;- 1
4、练习题
1、实现UDF，实现给定任意明文密码，返回md5加密后32位的密文密码。
- 场景
- 有用户密码表userinfo(id string,username string,password string);当下为空表。
- 利用动态插入数据insert into方法，
将数据记录{id:”001”,username:”张三”,明文密码password:”123456”}插入到表userinfo中， - 实际效果select * from userinfo的结果如下图，使用户看不到真实的用户密码。
001 张三 XXXXXXXXXXXXXXXXX
2、自定义UDAF名字为self_max，求取给定一整数值列表中的最大值函数max功能。
-
需求说明
数据样例：
系统自带的max函数使用
select id,username,max(score) from student_score group by id,username;
3、自定义UDAF名字为self_min，求取给定一整数值列表中的最小值函数min功能。
-
需求说明
数据样例
系统自带的min函数使用
select id,username,min(score) from student_score group by id,username;
上一篇： php获取当前页地址