一般做网站宽度是多少,重庆建工网,网站服务器速度查询,网站建设公司杭州一.自定义函数
1.Hive自带了一些函数#xff0c;比如#xff1a;max/min等#xff0c;但是数量有限#xff0c;自己可以通过自定义UDF来方便的扩展。
2.当Hive提供的内置函数无法满足你的业务处理需要时#xff0c;此时就可以考虑使用用户自定义函数。
3.根据用户自定义…一.自定义函数
1.Hive自带了一些函数比如max/min等但是数量有限自己可以通过自定义UDF来方便的扩展。
2.当Hive提供的内置函数无法满足你的业务处理需要时此时就可以考虑使用用户自定义函数。
3.根据用户自定义函数类别分为以下三种
1UDFUser-Defined-Function 一进一出。
2UDAFUser-Defined Aggregation Function 用户自定义聚合函数多进一出 。
3UDTFUser-Defined Table-Generating Functions 用户自定义表生成函数一进多出。
4.编程步骤
1继承Hive提供的类
org.apache.hadoop.hive.ql.udf.generic.GenericUDF
org.apache.hadoop.hive.ql.udf.generic.GenericUDTF
org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator2实现类中的抽象方法
3在hive的命令行窗口创建函数
4 创建临时函数
需要把jar包上传到服务器上面
添加jar。
add jar linux_jar_path创建function
create temporary function dbname.function_name AS class_name;删除函数
drop temporary function if exists dbname.function_name;5创建永久函数
需要把jar包上传到hdfs上面创建函数时jar包的位置使用hdfs的地址。
创建function
create function if exists my_udtf as com.zxl.hive.udf.ExplodeJSONArray using jar hdfs://flinkv1:8020/my_function/hive_udtf_funtion.jar;删除函数
drop function if exists my_udtf ;注意永久函数跟会话没有关系创建函数的会话断了以后其他会话也可以使用。 永久函数创建的时候在函数名之前需要自己加上库名如果不指定库名的话会默认把当前库的库名给加上。 永久函数使用的时候需要在指定的库里面操作或者在其他库里面使用的话加上库名.函数名。
二.UDF
官方案例https://cwiki.apache.org/confluence/display/Hive/HivePlugins#HivePlugins-CreatingCustomUDFs
计算给定基本数据类型的长度
package com.zxl.hive.udf;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;public class MyUDF extends GenericUDF {/** 判断传进来的参数的类型和长度* 约定返回的数据类型* */Overridepublic ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {//判断传进来的参数的长度if (arguments.length !1) {throw new UDFArgumentLengthException(please give me only one arg);}//判断传进来的参数的类型if (!arguments[0].getCategory().equals(ObjectInspector.Category.PRIMITIVE)){throw new UDFArgumentTypeException(1, i need primitive type arg);}// 约定返回的数据类型return PrimitiveObjectInspectorFactory.javaIntObjectInspector;}/** 具体解决逻辑* */Overridepublic Object evaluate(DeferredObject[] arguments) throws HiveException {Object o arguments[0].get();if(onull){return 0;}return o.toString().length();}/** 用于获取解释的字符串* */Overridepublic String getDisplayString(String[] strings) {return ;}
}1创建永久函数 注意因为add jar本身也是临时生效所以在创建永久函数的时候需要制定路径并且因为元数据的原因这个路径还得是HDFS上的路径。 三.UDTF
官网案例https://cwiki.apache.org/confluence/display/Hive/DeveloperGuideUDTF
执行步骤
要实现UDTF我们需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF,实现initialize, process, close三个方法。 UDTF首先会调用initialize方法此方法返回UDTF的返回行的信息返回个数类型; 初始化完成后会调用process方法真正的处理过程在process函数中在process中每一次forward()调用产生一行如果产生多列可以将多个列的值放在一个数组中然后将该数组传入到forward()函数 最后close()方法调用对需要清理的方法进行清理。
关于HIVE的UDTF自定义函数使用的更多详细内容请看 转载原文链接https://blog.csdn.net/lidongmeng0213/article/details/110877351
下面是json日志解析案例
package com.zxl.hive.udf;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.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 java.util.ArrayList;
import java.util.List;public class ExplodeJSONArray extends GenericUDTF {/*** 初始化方法,里面要做三件事* 1.约束函数传入参数的个数* 2.约束函数传入参数的类型* 3.约束函数返回值的类型*/Overridepublic StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {// TODO: 2023/12/6 返回结构体类型。udtf函数有可能炸开之后形成多列。所以用返回结构体来封装。属性名属性值。属性名就是列名属性值就是列的类型。//用结构体来约束函数传入参数的个数//List? extends StructField allStructFieldRefs argOIs.getAllStructFieldRefs(); --见名知意获取结构体所有属性的引用 可以看见返回值是个list类型.if(argOIs.getAllStructFieldRefs().size()!1){ //只要个数不等于1就抛出异常throw new UDFArgumentLengthException(explode_json_array()函数的参数个数只能为1);}//2.约束函数传入参数的类型// StructField structField argOIs.getAllStructFieldRefs().get(0);//只能有一个参数所以index给0 可以看见是获得结构体的属性//ObjectInspector fieldObjectInspector argOIs.getAllStructFieldRefs().get(0).getFieldObjectInspector();//获得属性的对象检测器 。通过检查器我们才能知道是什么类型.String typeName argOIs.getAllStructFieldRefs().get(0).getFieldObjectInspector().getTypeName();//我们要确保传入的类型是stringif(!string.equals(typeName)){throw new UDFArgumentTypeException(0,explode_json_array函数的第1个参数的类型只能为String.); //抛出异常}//3.约束函数返回值的类型ListString fieldNames new ArrayList(); //② 表示我建立了一个String类型的集合。表示存储的列名ListObjectInspector fieldOIs new ArrayList(); //②fieldNames.add(item); //炸裂之后有个列名如果不重新as那这个item就是列名fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector); //表示item这一列是什么类型.基本数据类型工厂类获取了个string类型的检查器//用一个工厂类获取StructObjectInspector类型。return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);//①获取标准结构体检查器。fieldNames,fieldOI是两个变量名}//这里是实现主逻辑的方法。首先分析下需求把json array字符串变成一个json字符串Overridepublic void process(Object[] args) throws HiveException {//1 获取函数传入的jsonarray字符串String jsonArrayStr args[0].toString(); //我要把jsonArrayStr字符串划分为一个一个的json通过字符串这种类型是不好划分的。不知道如何split切分//2 将jsonArray字符串转换成jsonArray数组。正常情况下我们要引入依赖比如fastjson啥的。JSONArray jsonArray new JSONArray(jsonArrayStr); //通过JSONArray这种类型我们就比较容易获得一条条的json字符串//3 得到jsonArray里面的一个个json并把他们写出。将actions里面的一个个action写出for (int i 0; i jsonArray.length(); i) { //普通for循环进行遍历String jsonStr jsonArray.getString(i);//前面定义了要返回String//forward是最后收集数据返回的方法forward(jsonStr);}}Overridepublic void close() throws HiveException {}
}注意UDTF函数不能和其他字段同时出现在select语句中负责SQL会执行失败
三.UDAF
官网案例https://cwiki.apache.org/confluence/display/Hive/GenericUDAFCaseStudy#GenericUDAFCaseStudy-Writingtheresolver
执行步骤
编写自定义函数需要创建三个类 1.继承 AbstractGenericUDAFResolver重写 getEvaluator方法对传入的值进行判断。 2.创建数据缓存区创建一些变量来进行调用赋值作为中间值类似于flink的checkpoints。 3.继承GenericUDAFEvaluator类重写方法即可实现具体逻辑的类。
参考文章
UDAF重要的类及原理分析(UDAF继承类的各个方法的用法) 原文链接https://blog.csdn.net/lidongmeng0213/article/details/110869457 Hive之ObjectInspector详解(UDAF中用到的类型详解) 原文链接https://blog.csdn.net/weixin_42167895/article/details/108314139
一个类似于SUM的自定义函数
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;// TODO: 2023/12/9 继承 AbstractGenericUDAFResolver重写 getEvaluator方法
public class FieldSum extends AbstractGenericUDAFResolver {Overridepublic GenericUDAFEvaluator getEvaluator(TypeInfo[] info) throws SemanticException {// TODO: 2023/12/9 判断传入的参数是否为一个if (info.length ! 1) {throw new UDFArgumentLengthException(只能传入一个参数, 但是现在有 info.length 个参数);}/*TypeInfoUtils是一个Java类它提供了一些用于处理Hive数据类型的实用方法。以下是TypeInfoUtils类中的一些方法及其功能getTypeInfoFromTypeString(String typeString) - 将类型字符串转换为Hive数据类型信息对象。getStandardJavaObjectInspectorFromTypeInfo(TypeInfo typeInfo) - 从Hive数据类型信息对象中获取标准Java对象检查器。isExactNumericType(PrimitiveTypeInfo typeInfo) - 检查给定的Hive原始数据类型是否为精确数值类型。getCategoryFromTypeString(String typeString) - 从类型字符串中获取Hive数据类型的类别。getPrimitiveTypeInfoFromPrimitiveWritable(Class? extends Writable writableClass) -从Hadoop Writable类中获取Hive原始数据类型信息对象。*/ObjectInspector objectInspector TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(info[0]);// TODO: 2023/12/9 判断是不是标准的java Object的primitive类型if (objectInspector.getCategory() ! ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentTypeException(0, Argument type must be PRIMARY. but objectInspector.getCategory().name() was passed!);}// 如果是标准的java Object的primitive类型说明可以进行类型转换PrimitiveObjectInspector inputOI (PrimitiveObjectInspector) objectInspector;// 如果是标准的java Object的primitive类型,判断是不是INT类型因为参数只接受INT类型if (inputOI.getPrimitiveCategory() ! PrimitiveObjectInspector.PrimitiveCategory.INT) {throw new UDFArgumentTypeException(0, Argument type must be INT, but inputOI.getPrimitiveCategory().name() was passed!);}return new FieldSumUDAFEvaluator();}Overridepublic GenericUDAFEvaluator getEvaluator(GenericUDAFParameterInfo info) throws SemanticException {return super.getEvaluator(info);}
}import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;// TODO: 2023/12/9 创建数据缓存区创建一些变量来进行调用赋值作为中间值类似于flink的checkpoints。
public class FieldSumBuffer extends GenericUDAFEvaluator.AbstractAggregationBuffer {Integer num 0;// TODO: 2023/12/9 实现变量的getset方法方便后面赋值取值public Integer getNum() {return num;}public void setNum(int num) {this.num num;}// TODO: 2023/12/9 创建累加的方法方便对变量进行累加public Integer addNum(int aum) {num aum;return num;}
}import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;// TODO: 2023/12/9 实现具体逻辑的地方 直接继承GenericUDAFEvaluator类重写方法即可
public class FieldSumUDAFEvaluator extends GenericUDAFEvaluator {// TODO: 2023/12/9 初始输入的变量 PrimitiveObjectInspector是Hadoop里面原始数据类别PrimitiveObjectInspector inputNum;PrimitiveObjectInspector middleNum;// TODO: 2023/12/9 最终输出的变量ObjectInspector outputNum;// TODO: 2023/12/9 最终统计值的变量int sumNum;// TODO: 2023/12/7 Model代表了UDAF在mapreduce的各个阶段。//* PARTIAL1: 这个是mapreduce的map阶段:从原始数据到部分数据聚合//* 将会调用iterate()和terminatePartial()//* PARTIAL2: 这个是mapreduce的map端的Combiner阶段负责在map端合并map的数据::从部分数据聚合到部分数据聚合://* 将会调用merge() 和 terminatePartial()//* FINAL: mapreduce的reduce阶段:从部分数据的聚合到完全聚合//* 将会调用merge()和terminate()//* COMPLETE: 如果出现了这个阶段表示mapreduce只有map没有reduce所以map端就直接出结果了:从原始数据直接到完全聚合//* 将会调用 iterate()和terminate()// TODO: 2023/12/7 确定各个阶段输入输出参数的数据格式ObjectInspectorsOverridepublic ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {super.init(m, parameters);// TODO: 2023/12/9 COMPLETE或者PARTIAL1输入的都是数据库的原始数据所以要确定输入的数据格式if (m Mode.PARTIAL1 || m Mode.COMPLETE) {inputNum (PrimitiveObjectInspector) parameters[0];} else {middleNum (PrimitiveObjectInspector) parameters[0];}// TODO: 2023/12/9 ObjectInspectorFactory是创建新的ObjectInspector实例的主要方法一般用于创建集合数据类型。输出的类型是Integer类型java类型outputNum ObjectInspectorFactory.getReflectionObjectInspector(Integer.class,ObjectInspectorFactory.ObjectInspectorOptions.JAVA);return outputNum;}// TODO: 2023/12/9 保存数据聚集结果的类Overridepublic AggregationBuffer getNewAggregationBuffer() throws HiveException {return new FieldSumBuffer();}// TODO: 2023/12/9 重置聚集结果Overridepublic void reset(AggregationBuffer agg) throws HiveException {//重新赋值为零((FieldSumBuffer) agg).setNum(0);}// TODO: 2023/12/9 map阶段迭代处理输入sql传过来的列数据不断被调用执行的方法最终数据都保存在agg中,parameters是新传入的数据Overridepublic void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {// TODO: 2023/12/9 判断如果传入的值是空的直接返回if (parameters null || parameters.length 1) {return;}Object javaObj inputNum.getPrimitiveJavaObject(parameters[0]);((FieldSumBuffer) agg).addNum(Integer.parseInt(javaObj.toString()));}// TODO: 2023/12/9 map与combiner结束返回结果得到部分数据聚集结果Overridepublic Object terminatePartial(AggregationBuffer agg) throws HiveException {return terminate(agg);}// TODO: 2023/12/9 combiner合并map返回的结果还有reducer合并mapper或combiner返回的结果。Overridepublic void merge(AggregationBuffer agg, Object partial) throws HiveException {((FieldSumBuffer) agg).addNum((Integer) middleNum.getPrimitiveJavaObject(partial));}// TODO: 2023/12/9 map阶段迭代处理输入sql传过来的列数据Overridepublic Object terminate(AggregationBuffer agg) throws HiveException {Integer num ((FieldSumBuffer) agg).getNum();return num;}
}打包上传注册函数进行测试
可以看到实现了对参数的判断和参数类型的判断 执行查询测试
