hive-UDF/UDTF/UDAF

1.Hive三种自定义函数

1.1 UDF

UDF，即用户定义函数(user-defined function)，作用于单行数据，并且产生一个数据行作为输出。Hive中大多数函数都属于这一类，比如数学函数和字符串函数。UDF函数的输入与输出值是1:1关系。

1.2 UDTF

UDTF，即用户定义表生成函数（user-defined table-generating function），作用于单行数据，并且产生多个数据行。UDTF函数的输入与输出值是1:n的关系。

1.3 UDAF

UDAF，用户定义聚集函数（user-defined aggregate function），作用于多行数据，并且产生一个输出数据行。Hive中像COUNT、MAX、MIN和SUM这样的函数就是聚集函数。UDAF函数的输入与输出值是n:1的关系。

2.UDF自定义函数实现

2.1 使用说明

UDF函数可以直接应用于select语句，对查询结构做格式化处理后，再输出内容。

2.2 注意事项

编写UDF函数的时候需要注意一下几点：

1）自定义UDF需要继承org.apache.hadoop.hive.ql.UDF。

2）需要实现evaluate函数，evaluate函数支持重载。

3) UDF必须要有返回类型，可以返回null，但是返回类型不能为void；

4）UDF中常用Text/LongWritable等类型，不推荐使用java类型；

2.3 添加hive依赖

<groupId>org.apache.hive</groupId>

</dependency>

2.4 自定义udf

功能：输入一条用户行为json数据，提取用户uid

package com.bigdata.hive;

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

import org.json.JSONObject;

public class ParseJsonUDF extends UDF {

public String evaluate(String line, String key){

JSONObject baseJson = new JSONObject(line.trim());

String result = "";

if(baseJson.has(key)){

return baseJson.getString(key);

}

return result;

}

public static void main(String[] args) {

String line = "{\"userId\":9527,\"day\":\"2021-06-12\",\"begintime\":1595058161312,\"endtime\":1595058469010,\"data\":[{\"package\":\"com.browser\",\"activetime\":120000}]}";

String userId = new ParseJsonUDF().evaluate(line,"userId");

System.out.println(userId);

}

2.5 项目打包

通过maven对自定义udf项目打包（比如learninghive-1.0-SNAPSHOT.jar）

mvn clean package

2.6 项目包上传至HDFS

通过hdfs 命令，将项目打好的包上传至hdfs指定目录下。

bin/hdfs dfs -put home/hadoop/shell/lib/learninghive-1.0-SNAPSHOT.jar user/hive/jars

2.7 hive中添加项目包

通过bin/hive进入hive控制台，将项目包添加到hive中。

hive> add jar hdfs://mycluster/user/hive/jars/learninghive-1.0-SNAPSHOT.jar;

2.8 创建udf 函数模板

根据自定义udf类，创建一个自定义函数名称，供后面hive SQL 使用。

hive> create temporary function json_udf as 'com.bigdata.hive.ParseJsonUDF';

2.9 创建测试表

CREATE TABLE behavior(line STRING)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '\n'

STORED AS TEXTFILE;

2.10 准备测试数据集

准备测试数据文件behavior.log，每行数据代表用户行为数据。

{"userId":9527,"day":"2021-05-12","begintime":1620785058833,"endtime":1620785217511,"data":[{"package":"com.browser","activetime":120000},{"package":"com.qq","activetime":80000}]}

{"userId":3854,"day":"2021-05-12","begintime":1620785290612,"endtime":1595058469010,"data":[{"package":"com.browser","activetime":60000},{"package":"com.qq","activetime":150000}]}

2.11 数据加载至hive表

将本地文件加载到刚刚创建的behavior表中。

LOAD DATA LOCAL INPATH '/home/hadoop/shell/data/behavior.log' INTO TABLE behavior;

2.12 应用自定义udf函数

hive> select json_udf(line,'userId') from behavior;

可以通过自定义json_udf函数提取用户行为json数据中的用户uid。

3.UDTF自定义函数实现

3.1 使用说明

自定义UDTF需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF类,重新initialize, process, close三个方法：

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

2）初始化完成后，会调用process()方法,真正的处理过程在process函数中，在process中，每一次forward()调用产生一行；

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

3.2自定义udtf

功能：输入一条用户行为json数据，输出多条用户行为日志数据。

package com.bigdata.hive;

import groovy.json.JsonException;

import org.apache.commons.lang.StringUtils;

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 org.json.JSONArray;

import org.json.JSONException;

import java.util.ArrayList;

public class ParseJsonUDTF extends GenericUDTF {

* 该方法中，我们将指定输出参数的名称和参数类型

* @param argOIs

* @return

* @throws UDFArgumentException

@Override

public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {

ArrayList<String> fieldNames = new ArrayList<String>();

ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();

fieldNames.add("package");

fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

fieldNames.add("activetime");

fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);

}

* 输入 1 条记录，输出 n 条记录

* @param objects

* @throws HiveException

@Override

public void process(Object[] objects) throws HiveException {

获取传入的data

String data = objects[0].toString();

如果传进来的数据为空，直接返回过滤掉该数据

if(StringUtils.isBlank(data)){

return ;

}else{

try {

获取一共有几条操作记录

JSONArray dataArray = new JSONArray(data);

if (dataArray == null) return;

循环遍历每个事件

for (int i=0;i<dataArray.length();i++){

String[] output = new String[2];

try {

output[0]=dataArray.getJSONObject(i).getString("package");

output[1]=dataArray.getJSONObject(i).getString("activetime");

}catch (JSONException e){

continue;

}

将结果返回

forward(output);

}

}catch (JSONException e){

e.printStackTrace();

}

* 当没有记录处理的时候该方法会被调用，用来清理代码或者产生额外的输出

* @throws HiveException

@Override

public void close() throws HiveException {

}

3.3 项目打包

通过maven对自定义udf项目打包（比如learninghive-1.0-SNAPSHOT.jar）

mvn clean package

3.4 项目包上传至HDFS

通过hdfs 命令，将项目打好的包上传至hdfs指定目录下。

bin/hdfs dfs -put home/hadoop/shell/lib/learninghive-1.0-SNAPSHOT.jar user/hive/jars

3.5 hive中添加项目包

通过bin/hive进入hive控制台，将项目包添加到hive中。

hive> add jar hdfs://mycluster/user/hive/jars/learninghive-1.0-SNAPSHOT.jar;

3.6 创建udf 函数模板

根据自定义udf类，创建一个自定义函数名称，供后面hive SQL 使用。

hive> create temporary function json_udtf as 'com.bigdata.hive.ParseJsonUDTF';

hive> create temporary function json_udf as 'com.bigdata.hive.ParseJsonUDF';

3.7 创建测试表

CREATE TABLE behavior(line STRING)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '\n'

STORED AS TEXTFILE;

3.8 准备测试数据集

准备测试数据文件behavior.log，每行数据代表用户行为数据。

{"userId":9527,"day":"2021-05-12","begintime":1620785058833,"endtime":1620785217511,"data":[{"package":"com.browser","activetime":120000},{"package":"com.qq","activetime":80000}]}

{"userId":3854,"day":"2021-05-12","begintime":1620785290612,"endtime":1595058469010,"data":[{"package":"com.browser","activetime":60000},{"package":"com.qq","activetime":150000}]}

3.9 数据加载至hive表

将本地文件加载到刚刚创建的behavior表中。

LOAD DATA LOCAL INPATH '/home/hadoop/shell/data/behavior.log' INTO TABLE behavior;

3.10应用自定义udtf函数

hive> select

json_udf(line,'userId'),

json_udf(line,'day'),

json_udf(line,'begintime'),

json_udf(line,'endtime'),

package,

activetime

from behavior lateral view json_udtf(json_udf

(line,'data')) tmpdata as package,activetime;

备注：与lateral view一起使用，执行过程相当于单独执行了两次抽取，然后union到一个表里。

可以通过自定义 json_udtf函数解析用户行为json数据，然后输出多条用户行为具体操作日志。

4.UDAF自定义函数实现

4.1 使用说明

1）必须继承

org.apache.hadoop.hive.ql.exec.UDAF(函数类继承)

org.apache.hadoop.hive.ql.exec.UDAFEvaluator(内部类Evaluator实现UDAFEvaluator接口)

2）Evaluator需要实现以下几个函数

init():初始化

iterate():迭代处理每一行数据

terminatePartial():输出map/reduce的阶段结果

merge():combiner/reduce对数据进行聚合

terminate():返回最终的聚集函数结果

4.2 udaf运行过程详解

UDAF 就是一个多行导成一行的聚合函数，它的过程与MR过程紧密结合

4.3 自定义udaf

功能：统计表中某一列的最大值

package com.bigdata.hive;

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

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

import org.apache.hadoop.io.IntWritable;

public class MyMaxUDAF extends UDAF {

static public class MaxIntEvaluator implements UDAFEvaluator {

private int mMax;

private boolean mEmpty;

构造方法

public MaxIntEvaluator() {

super();

init();

}

类似于构造函数，用于UDAF的初始化

public void init() {

mMax = 0;

mEmpty = true;

}

迭代处理每一行数据

public boolean iterate(IntWritable o) {

if (o != null) {

if (mEmpty) {

mMax = o.get();

mEmpty = false;

} else {

mMax = Math.max(mMax, o.get());

}

return true;

}

//输出map/reduce的阶段结果

public IntWritable terminatePartial() {

return mEmpty ? null : new IntWritable(mMax);

}

//combiner/reduce对数据进行聚合

public boolean merge(IntWritable o) {

return iterate(o);

}

//返回最终的聚集函数结果

public IntWritable terminate() {

return mEmpty ? null : new IntWritable(mMax);

}

4.4 项目打包

通过maven对自定义udf项目打包（比如learninghive-1.0-SNAPSHOT.jar）

mvn clean package

4.5 项目包上传至HDFS

通过hdfs 命令，将项目打好的包上传至hdfs指定目录下。

bin/hdfs dfs -put /home/hadoop/shell/lib/learninghive-1.0-SNAPSHOT.jar /user/hive/jars

4.6 hive中添加项目包

通过bin/hive进入hive控制台，将项目包添加到hive中。

hive> add jar hdfs://mycluster/user/hive/jars/learninghive-1.0-SNAPSHOT.jar;

4.7 创建udf 函数模板

根据自定义udf类，创建一个自定义函数名称，供后面hive SQL 使用。

create temporary function max_udaf as 'com.bigdata.hive.MyMaxUDAF';

4.8 创建测试表

create table if not exists temperature

(id string comment '气象站id',year string comment '年',temperature int comment '气温')

comment '天气表'

ROW FORMAT DELIMITED FIELDS TERMINATED BY ','

STORED AS TEXTFILE ;

4.9 准备测试数据集

准备测试数据文件temperature，每行数据代表气象站天气气温数据。

03103,1980,41

03103,1981,98

03103,1982,70

03103,1983,74

03103,1984,77

03103,1985,78

03103,1986,82

03103,1987,75

03103,1988,81

03103,1989,89

4.10 数据加载至hive表

将本地文件加载到刚刚创建的temperature表中。

load data local inpath '/weather/temperature' overwrite into table temperature ;

4.11 应用自定义udaf函数

select max_udaf(temperature) from temperature;

可以通过自定义max_udaf函数统计出tempera-

ture表中气温最大值。

hive-UDF/UDTF/UDAF相关推荐

Hive UDF 开发手册
文档目的笔者在工作中有接触到 Hive UDF 的开发任务,大部分 UDF 开发并不困难,困难的往往是: 不清楚 UDF 代码的编写逻辑(UDF.UDTF.UDAF) 不清楚如何传入特定类型的参数 ...
Hive 10、Hive的UDF、UDAF、UDTF
Hive自定义函数包括三种UDF.UDAF.UDTF UDF(User-Defined-Function) 一进一出 UDAF(User- Defined Aggregation Funcation) ...
UDF、UDAF、UDTF函数编写
一.UDF函数编写 1.步骤 1.继承UDF类 2.重写evalute方法 1.继承GenericUDF 2.实现initialize.evaluate.getDisplayString方法 2.案例 ...
UDF、UDAF、UDTF之间的区别
1.UDF:用户定义(普通)函数,只对单行数值产生作用: 继承UDF类,添加方法 evaluate() /*** @function 自定义UDF统计最小值**/public class Min ex ...
自定义UDF、UDAF、UDTF函数
注意事项: 1.udf.udaf函数的使用都需要使用sqlContext来创建function,如果是scala里需要引用Java的方法或者函数的话,需要包装一下,再写个scala的方法,将Java的 ...
【Flink】Flink Table SQL 用户自定义函数: UDF、UDAF、UDTF
本文总结Flink Table & SQL中的用户自定义函数: UDF.UDAF.UDTF. UDF: 自定义标量函数(User Defined Scalar Function).一行输入一行 ...
hive中UDF跟UDAF使用说明
Hive进行UDF开发十分简单,此处所说UDF为Temporary的function,所以需要hive版本在0.4.0以上才可以. 一.背景: 1.Hive是基于Hadoop中的MapReduce,提 ...
hive UDF函数
虽然Hive提供了很多函数,但是有些还是难以满足我们的需求.因此Hive提供了自定义函数开发自定义函数包括三种UDF.UADF.UDTF UDF(User-Defined-Function) UDA ...
Hive UDF，就这
摘要:Hive UDF是什么?有什么用?怎么用?什么原理?本文从UDF使用入手,简要介绍相关源码,UDF从零开始. 本文分享自华为云社区<Hive UDF,就这>,作者:汤忒撒. Hive ...
CDH 创建Hive UDF函数
导入依赖包: hive-exec.jar hadoop-common.jar 注意:函数名必须为 evaluate ,否则hive无法识别! package com.example.hive.udf; ...

hive-UDF/UDTF/UDAF

hive-UDF/UDTF/UDAF相关推荐

最新文章

热门文章