怎么做英文的网站,wordpress 右侧悬浮,中国知名企业排行榜,公司起名字大全免费好听必过的背景介绍
UDF来源于Hive#xff0c;Hive可以允许用户编写自己定义的函数UDF#xff0c;然后在查询中进行使用。星环Inceptor中的UDF开发规范与Hive相同#xff0c;目前有3种UDF#xff1a;
A. UDF--以单个数据行为参数#xff0c;输出单个数据行#xff1b;
UDF#…背景介绍
UDF来源于HiveHive可以允许用户编写自己定义的函数UDF然后在查询中进行使用。星环Inceptor中的UDF开发规范与Hive相同目前有3种UDF
A. UDF--以单个数据行为参数输出单个数据行
UDFUser Defined Function即用户自定义函数能结合SQL语句一起使用更好地表达复杂的业务逻辑一般以单个数据行为参数输出单个数据行比如数学函数、字符串函数、时间函数、拼接函数
B. UDTF 以一个数据行为参数输出多个数据行为一个表作为输出
UDTFUser Defined Table Function即用户自定义表函数它与UDF类似。区别在于UDF只能实现一对一而它用来实现多行/列对多行/列数据的处理逻辑。一般以一个数据行为参数输出多个数据行为一个表作为输出如lateral、view、explore
C. UDAF 以多个数据行为参数输出一个数据行
UDAFUser Defined Aggregate Function用户自定义聚合函数是由用户自主定义的用法同如MAX、MIN和SUM已定义的聚合函数一样的处理函数。而且不同于只能处理标量数据的系统定义的聚合函数UDAF的可以接受并处理更广泛的数据类型如用对象类型、隐式类型或者LOB存储的多媒体数据。由于UDAF也属于聚合函数中的一种同样也需要与GROUPBY结合使用。
一般UDAF以多个数据行为参数接收多个数据行并输出一个数据行比如COUNT、MAX UDF、UDTF、UDAF的开发要点及使用DEMO
星环Quark计算引擎中内置了很多函数同时支持用户自行扩展按规则添加后即可在sql执行过程中使用目前支持UDF、UDTF、UDAF三种类型一般UDF应用场景较多后面将着重介绍UDF的开发与使用。UDAF及UDTF将主要介绍开发要点以及Demo示例。 Quark的UDF接口兼容开源Hive的UDF接口用户可以参考开源Hive的UDF手册或者直接把开源Hive的UDF迁移到Quark上。 UDF
Quark数据类型 Quark类型 Java原始类型 Java包装类 hadoop.hive.ioWritable tinyintbyteByteByteWritablesmallintshortShortShortWritableintintIntegerIntWritablebigintlongLongLongWritablestring-StringTextcharcharCharacterHiveCharWritablebooleanbooleanBooleanBooleanWritablefloatfloatFloatFloatWritabledouble doubleDoubleDoubleWritabledecimal-BigDecimalHiveDecimalWritabledate-DateDateWritablearray-ListArrayListWritableMapK,V-MapK.VHashMapWritable
UDF函数
Quark 提供了两个实现 UDF 的方式
第一种继承 UDF 类
优点实现简单支持Quark的基本类型、数组和Map支持函数重载。缺点逻辑较为简单只适合用于实现简单的函数
第二种继承 GenericUDF 类
优点支持任意长度、任意类型的参数可以根据参数个数和类型实现不同的逻辑资源消耗更低可以实现初始化和关闭资源的逻辑(initialize、close)。缺点实现比继承UDF要复杂一些
一般在以下几种场景下考虑使用GenericUDF
传参情况复杂比如某UDF要传参数有多种数量或多种类型的情况在UDF中支持这种场景我们需要实现N个不同的evaluate()方法分别对应N种场景的传参在GenericUDF我们只需在一个方法内加上判断逻辑对不同的输入路由到不同的处理逻辑上即可。还有比如某UDF参数既要支持String list参数也要支持Integer list参数。你可能认为我们只要继续多重载方法就好了但是Java不支持同一个方法重载参数只有泛型类型不一样所以该场景只能用GenericUDF。需要传非Writable的或复杂数据类型作为参数。比如嵌套数据结构传入Map的key-value中的value为list数据类型或者比如数据域数量不确定的Struct结构都更适合使用GenericUDF在运行时捕获数据的内部构造。该UDF被大量、高频地使用所以从收益上考虑会尽可能地优化一切可以优化的地方则GenericUDF相比UDF在operator中避免了多次反射转化的资源消耗后面会细讲更适合被考虑。该UDF函数功能未来预期的重构、扩展场景较多需要做得足够可扩展则GenericUDF在这方面更优秀。
pom文件的依赖导入
UDF开发依赖dependencygroupIdorg.apache.hive/groupIdartifactIdinceptor-exec/artifactIdversionxxx/version
/dependency
继承示例
1.继承 UDF 类
该方式实现简单只需新建一个类继承org.apache.hadoop.hive.ql.exec.UDF
继承UDF类必须实现evaluate方法且返回值类型不能为 void支持定义多个evaluate方法不同参数列表用于处理不同类型数据
可通过完善Description展示UDF用法 UDF样例。
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.hive.ql.exec.Description;Description(namemy_plus,valuemy_plus() - if string, do concat; if integer, do plus,extended Example : \n select my_plus(a, b);\n ab\n select my_plus(3, 5);\n 8
)
/*** 实现UDF函数若字符串执行拼接int类型执行加法运算。*/
public class AddUDF extends UDF {/*** 编写一个函数要求如下* 1. 函数名必须为 evaluate* 2. 参数和返回值类型可以为Java基本类型、Java包装类、org.apache.hadoop.io.Writable等类型、List、Map* 3. 函数一定要有返回值不能为 void*/public String evaluate(String... parameters) {if (parameters null || parameters.length 0) {return null;}StringBuilder sb new StringBuilder();for (String param : parameters) {sb.append(param);}return sb.toString();}/*** 支持函数重载*/public int evaluate(IntWritable... parameters) {if (parameters null || parameters.length 0) {return 0;}long sum 0;for (IntWritable currentNum : parameters) {sum Math.addExact(sum, currentNum.get());}return (int) sum;}
}
2.继承 GenericUDF 类
GenericUDF相比与UDF功能更丰富支持所有参数类型实现起来也更加复杂。org.apache.hadoop.hive.ql.udf.generic.GenericUDF API提供了一个通用的接口将任何数据类型的对象当作泛型Object去调用和输出参数类型由ObjectInspector封装参数Writable类由DeferredObject封装使用时简单类型可直接从Writable获取复杂类型可由ObjectInspector解析。
Java的ObjectInspector类用于帮助Quark了解复杂对象的内部架构通过创建特定的ObjectInspector对象替代创建具体类对象在内存中储存某类对象的信息。在UDF中ObjectInspector用于帮助Hive引擎将HQL转成MR Job时确定输入和输出的数据类型。Hive语句会生成MapReduce Job执行所以使用的是Hadoop数据格式不是编写UDF的Java的数据类型比如Java的int在Hadoop为IntWritableString在Hadoop为Text格式所以我们需要将UDF内的Java数据类型转成正确的Hadoop数据类型以支持Hive将HQL生成MapReduce Job。
继承 GenericUDF 后必须实现以下三个方法
public class MyCountUDF extends GenericUDF {private PrimitiveObjectInspector.PrimitiveCategory[] inputType;private transient ObjectInspectorConverters.Converter intConverter;private transient ObjectInspectorConverters.Converter longConverter;// 初始化Overridepublic ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {}// DeferredObject封装实际参数的对应Writable类Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {}// 函数信息Overridepublic String getDisplayString(String[] strings) {}
}
initialize()方法只在 GenericUDF 初始化时被调用一次执行一些初始化操作包括参数个数检查参数类型检查与转换确定返回值类型。
a. 参数个数检查
可通过 arguments 数组的长度来判断函数参数的个数
// 检查该记录是否传过来正确的参数数量arguments的长度不为2时则抛出异常if (arguments.length ! 2) {throw new UDFArgumentLengthException(arrayContainsExample only takes 2 arguments: ListT, T);}
b. 参数类型检查与转换
针对该UDF的每个参数initialize()方法都会收到一个对应的ObjectInspector参数通过遍历ObjectInspector数组检查每个参数类型根据参数类型构造ObjectInspectorConverters.Converter用于将Hive传递的参数类型转换为对应的Writable封装对象ObjectInspector供后续统一处理。
ObjectInspector内部有一个枚举类 Category代表了当前 ObjectInspector 的类型。
public interface ObjectInspector extends Cloneable {public static enum Category {PRIMITIVE, // Hive原始类型LIST, // Hive数组MAP, // Hive MapSTRUCT, // 结构体UNION // 联合体};
}
Quark原始类型又细分了多种子类型PrimitiveObjectInspector 实现了 ObjectInspector可以更加具体的表示对应的Hive原始类型。
public interface PrimitiveObjectInspector extends ObjectInspector {/*** The primitive types supported by Quark.*/public static enum PrimitiveCategory {VOID, BOOLEAN, BYTE, SHORT, INT, LONG, FLOAT, DOUBLE, STRING,DATE, TIMESTAMP, BINARY, DECIMAL, VARCHAR, CHAR, INTERVAL_YEAR_MONTH, INTERVAL_DAY_TIME,UNKNOWN};
}
参数类型检查与转换示例
for (int i 0; i length; i) { // 遍历每个参数ObjectInspector currentOI arguments[i];ObjectInspector.Category type currentOI.getCategory(); // 获取参数类型if (type ! ObjectInspector.Category.PRIMITIVE) { // 检查参数类型throw new UDFArgumentException(The function my_count need PRIMITIVE Category, but get type);}PrimitiveObjectInspector.PrimitiveCategory primitiveType ((PrimitiveObjectInspector) currentOI).getPrimitiveCategory();inputType[i] primitiveType;switch (primitiveType) { // 参数类型转换case INT:if (intConverter null) {ObjectInspector intOI PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(primitiveType);intConverter ObjectInspectorConverters.getConverter(currentOI, intOI);}break;case LONG:if (longConverter null) {ObjectInspector longOI PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(primitiveType);longConverter ObjectInspectorConverters.getConverter(currentOI, longOI);}break;default:throw new UDFArgumentException(The function my_count need INT OR BIGINT, but get primitiveType);}
}
c. 确定函数返回值类型
initialize() 需要 return 一个 ObjectInspector 实例用于表示自定义UDF返回值类型。initialize() 的返回值决定了 evaluate() 的返回值类型。创建ObjectInspector时不要用new的方式创建应该用工厂模式去创建以保证相同类型的ObjectInspector只有一个实例且同一个ObjectInspector可以在代码中多处被使用。
// 自定义UDF返回值类型为Long
return PrimitiveObjectInspectorFactory.writableLongObjectInspector;
完整的 initialize() 函数
public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {int length arguments.length;inputType new PrimitiveObjectInspector.PrimitiveCategory[length];for (int i 0; i length; i) {ObjectInspector currentOI arguments[i];ObjectInspector.Category type currentOI.getCategory();if (type ! ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException(The function my_count need PRIMITIVE Category, but get type);}PrimitiveObjectInspector.PrimitiveCategory primitiveType ((PrimitiveObjectInspector) currentOI).getPrimitiveCategory();inputType[i] primitiveType;switch (primitiveType) {case INT:if (intConverter null) {ObjectInspector intOI PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(primitiveType);intConverter ObjectInspectorConverters.getConverter(currentOI, intOI);}break;case LONG:if (longConverter null) {ObjectInspector longOI PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(primitiveType);longConverter ObjectInspectorConverters.getConverter(currentOI, longOI);}break;default:throw new UDFArgumentException(The function my_count need INT OR BIGINT, but get primitiveType);}}return PrimitiveObjectInspectorFactory.writableLongObjectInspector;}
evaluate()方法是GenericUDF的核心方法自定义UDF的实现逻辑。代码实现步骤可以分为三部分参数接收自定义UDF核心逻辑返回处理结果。
第一步参数接收
evaluate() 的参数就是 自定义UDF 的参数。
/*** Evaluate the GenericUDF with the arguments.** param arguments* The arguments as DeferedObject, use DeferedObject.get() to get the* actual argument Object. The Objects can be inspected by the* ObjectInspectors passed in the initialize call.* return The*/
public abstract Object evaluate(DeferredObject[] arguments)throws HiveException;
通过源码注释可知DeferedObject.get() 可以获取参数的值。
/*** A Defered Object allows us to do lazy-evaluation and short-circuiting.* GenericUDF use DeferedObject to pass arguments.*/
public static interface DeferredObject {void prepare(int version) throws HiveException;Object get() throws HiveException;
};
再看看 DeferredObject 的源码DeferedObject.get() 返回的是 Object传入的参数不同会是不同的Java类型。
第二步自定义UDF核心逻辑
这一部分根据实际项目需求自行编写。
第三步返回处理结果
这一步和 initialize() 的返回值一一对应基本类型返回值有两种Writable类型 和 Java包装类型
在 initialize 指定的返回值类型为 Writable类型 时在 evaluate() 中 return 的就应该是对应的 Writable实例。在 initialize 指定的返回值类型为 Java包装类型 时在 evaluate() 中 return 的就应该是对应的 Java包装类实例。
evalute()示例
Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {LongWritable out new LongWritable();for (int i 0; i deferredObjects.length; i) {PrimitiveObjectInspector.PrimitiveCategory type this.inputType[i];Object param deferredObjects[i].get();switch (type) {case INT:Object intObject intConverter.convert(param);out.set(Math.addExact(out.get(), ((IntWritable) intObject).get()));break;case LONG:Object longObject longConverter.convert(param);out.set(Math.addExact(out.get(), ((LongWritable) longObject).get()));break;default:throw new IllegalStateException(Unexpected type in MyCountUDF evaluate : type);}}return out;}
getDisplayString() 返回的是 explain 时展示的信息。这里不能return null否则可能在运行时抛出空指针异常。
Override
public String getDisplayString(String[] strings) {return my_count( Joiner.on(, ).join(strings) );
}
自定义GenericUDF完整示例
Description(namemy_count,valuemy_count(...) - count int or long type numbers,extended Example :\n select my_count(3, 5);\n 8\n select my_count(3, 5, 25);\n 33
)
public class MyCountUDF extends GenericUDF {private PrimitiveObjectInspector.PrimitiveCategory[] inputType;private transient ObjectInspectorConverters.Converter intConverter;private transient ObjectInspectorConverters.Converter longConverter;Overridepublic ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {int length objectInspectors.length;inputType new PrimitiveObjectInspector.PrimitiveCategory[length];for (int i 0; i length; i) {ObjectInspector currentOI objectInspectors[i];ObjectInspector.Category type currentOI.getCategory();if (type ! ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException(The function my_count need PRIMITIVE Category, but get type);}PrimitiveObjectInspector.PrimitiveCategory primitiveType ((PrimitiveObjectInspector) currentOI).getPrimitiveCategory();inputType[i] primitiveType;switch (primitiveType) {case INT:if (intConverter null) {ObjectInspector intOI PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(primitiveType);intConverter ObjectInspectorConverters.getConverter(currentOI, intOI);}break;case LONG:if (longConverter null) {ObjectInspector longOI PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(primitiveType);longConverter ObjectInspectorConverters.getConverter(currentOI, longOI);}break;default:throw new UDFArgumentException(The function my_count need INT OR BIGINT, but get primitiveType);}}return PrimitiveObjectInspectorFactory.writableLongObjectInspector;}Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {LongWritable out new LongWritable();for (int i 0; i deferredObjects.length; i) {PrimitiveObjectInspector.PrimitiveCategory type this.inputType[i];Object param deferredObjects[i].get();switch (type) {case INT:Object intObject intConverter.convert(param);out.set(Math.addExact(out.get(), ((IntWritable) intObject).get()));break;case LONG:Object longObject longConverter.convert(param);out.set(Math.addExact(out.get(), ((LongWritable) longObject).get()));break;default:throw new IllegalStateException(Unexpected type in MyCountUDF evaluate : type);}}return out;}Overridepublic String getDisplayString(String[] strings) {return my_count( Joiner.on(, ).join(strings) );}
}
UDTF
UDTF函数作用都是输入一行数据将该行数据拆分、并返回多行数据。不同的UDTF函数只是拆分的原理不同、作用的数据格式不同而已。
适用场景
流应用中对数据处理如字符串解析hyperbase数据删除时间段去重时间段统计数仓数集应用中需要将单行转换为多行inceptor内置多种UDTF如explodeinlinejson_tuple等 注意返回UDTF结果的同时查询其他对象须引用关键字 LATERAL VIEW UDTF开发要点
1. 实现UDTF函数需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF
2. 然后重写/实现initialize, process, close三个方法
A. initialize初始化验证返回字段名和字段类型
initialize初始化UDTF首先会调用initialize方法此方法返回UDTF的返回行的信息返回个数类型名称。initialize针对任务调一次 作用是定义输出字段的列名、和输出字段的数据类型。
initialize方法示例
Override/*** 返回数据类型StructObjectInspector* 定义输出数据的列名、和数据类型。*/public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {ListString fieldNames new ArrayListString(); //fieldNames为输出的字段名fieldNames.add(world);ListObjectInspector fieldOIs new ArrayListObjectInspector(); //类型,列输出类型fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);}
B. 初始化完成后调用process方法对传入的参数进行处理通过forword()方法把结果返回
process初始化完成后会调用process方法对传入的参数进行处理可以通过forword()方法把结果写出。process传入一行数据写出去多次传入一行数据输出多行数据如mapreduce单词计数。process针对每行数据调用一次该方法。在initialize初始化的时候定义输出字段的数据类型是集合调用forward()将数据写入到一个缓冲区写入缓冲区的数据也要是集合。
process方法示例
//数据的集合private ListString dataList new ArrayListString();/*** process(Object[] objects) 参数是一个数组但是hive中的explode函数接受的是一个一进多出* param args* throws HiveException*/public void process(Object[] args) throws HiveException {//我们现在的需求是传入一个数据在传入一个分割符//1.获取数据String data args[0].toString();//2.获取分割符String splitKey args[1].toString();//3.切分数据,得到一个数组String[] words data.split(splitKey);//4.想把words里面的数据全部写出去。类似在map方法中通过context.write方法// 定义是集合、写出去是一个string类型不匹配写出也要写出一个集合for (String word : words) {//5.将数据放置集合EG传入hello,world,hdfs----写出需要写n次hello\worlddataList.clear();//清空数据集合dataList.add(word);//5.写出数据的操作forward(dataList);}}
C. 最后调用close()方法进行清理工作
最后close()方法调用对需要清理的方法进行清理close()方法针对整个任务调一次
UDTF DEMO
下面UDTF 实现的是字符串的分拆多行输出
package io.transwarp.udtf;
import java.util.ArrayList;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
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;
public class SplitUDF extends GenericUDTF{Overridepublic void close() throws HiveException {// TODO Auto-generated method stub}Overridepublic StructObjectInspector initialize(ObjectInspector[] arg0) throws UDFArgumentException {// TODO Auto-generated method stubif(arg0.length ! 1){throw new UDFArgumentLengthException(SplitString only takes one argument);}if(arg0[0].getCategory() ! ObjectInspector.Category.PRIMITIVE){throw new UDFArgumentException(SplitString only takes string as a parameter);}ArrayListString fieldNames new ArrayList();ArrayListObjectInspector fieldOIs new ArrayList();fieldNames.add(col1);fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);fieldNames.add(col2);fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);}Overridepublic void process(Object[] arg0) throws HiveException {// TODO Auto-generated method stubString input arg0[0].toString();String[] inputSplits input.split(#);for (int i 0; i inputSplits.length; i) {try {String[] result inputSplits[i].split(:);forward(result);} catch (Exception e) {continue;}}}
}
执行效果如下 如何使用UDTF
将UDTF打包后放在inceptor server 所在节点之上建议不要放在/user/lib/hive/lib/下之后在连接inceptor执行以下命令生成临时函数server有效重启inceptor失效
add jar /tmp/timestampUDF.jar
drop temporary function timestamp_ms;
create temporary function timestamp_ms as io.transwarp.udf.ToTimestamp;select date, timestamp_ms(date) from table1; UDAF
正如前面所说UDAF是由用户自主定义的虽然UDAF的使用可以方便对数据的运算处理但是使用它的数量建议不要过多因为UDAF的数量增长和性能下降成线性关系。另外如果存在大量的嵌套UDAF系统的性能也会降低建议用户在可能的情况下写一个没有嵌套或者嵌套较少的UDAF实现相同功能来提高性能。
UDAF开发要点
1. 用户的UDAF必须继承了org.apache.hadoop.hive.ql.exec.UDAF
2. 用户的UDAF必须包含至少一个实现了org.apache.hadoop.hive.ql.exec的静态类诸如常见的实现了 UDAFEvaluator。
3. 一个计算函数必须实现的5个方法的具体含义如下
- init()主要是负责初始化计算函数并且重设其内部状态一般就是重设其内部字段。一般在静态类中定义一个内部字段来存放最终的结果。- iterate()每一次对一个新值进行聚集计算时候都会调用该方法计算函数会根据聚集计算结果更新内部状态。当输入值合法或者正确计算了则就返回true。- terminatePartial()Hive需要部分聚集结果的时候会调用该方法必须要返回一个封装了聚集计算当前状态的对象。- merge()Hive进行合并一个部分聚集和另一个部分聚集的时候会调用该方法。- terminate()Hive最终聚集结果的时候就会调用该方法。计算函数需要把状态作为一个值返回给用户。
UDAF DEMO
下面的UDAF DEMO目标是实现找到最大值功能以表中某一字段为参数返回最大值。
package udaf.transwarp.io;
import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.apache.hadoop.io.IntWritable;//UDAF是输入多个数据行产生一个数据行
//用户自定义的UDAF必须是继承了UDAF且内部包含多个实现了exec的静态类
public class MaxiNumber extends UDAF{public static class MaxiNumberIntUDAFEvaluator implements UDAFEvaluator{//最终结果private IntWritable result;//负责初始化计算函数并设置它的内部状态result是存放最终结果的Overridepublic void init() {resultnull;}//每次对一个新值进行聚集计算都会调用iterate方法public boolean iterate(IntWritable value){if(valuenull)return false;if(resultnull)resultnew IntWritable(value.get());elseresult.set(Math.max(result.get(), value.get()));return true;}//Hive需要部分聚集结果的时候会调用该方法//会返回一个封装了聚集计算当前状态的对象public IntWritable terminatePartial(){return result;}//合并两个部分聚集值会调用这个方法public boolean merge(IntWritable other){return iterate(other);}//Hive需要最终聚集结果时候会调用该方法public IntWritable terminate(){return result;}}
}
UDF 的打包与使用
操作前提
将开发好自定义UDF函数的项目打包成jar包注意jar 包中的自定义UDF 类名不能和现有UDF 类在包名类名上完全相同。
部署方式
常见的UDF部署方式有以下三种
把UDF固化到image里重新打image推荐其次是通过创建临时UDFadd jar temporary function的方式创建永久UDFhdfs jarpermanent function的方式可行但不是很推荐
方式一 固化UDF
视频示例仅作示范详情查看下方文字
此方式的核心逻辑是把UDF jar包放到image的/usr/lib/inceptor/下面重新制作image。具体步骤如下
以更换inceptor中的inceptor_2.10-1.1.0-transwarp-6.1.0.jar为例
1. 进入inceptor image
docker run -it inceptor_image_id bash 2. 打开另一个terminal
3. 替换container中的jar包
docker cp jar包名称 container_id:/usr/lib/inceptor/ jar包名称 4. commit修改记录
docker commit container_id REPOSITORY:TAG 5. 打开manager管理页面重新启动inceptor服务 6.重启完成后即可查看quark server的pod下/usr/lib/inceptor/是否有新增的jar包 方式二 创建临时UDF
视频示例仅作示范详情查看下方文字
1. 查看已存在jar包
LIST JAR; 2. 添加jar包
ADD JAR[S] local_path;
// Local_path是jar包所在Inceptor server节点的路径。 3. 创建临时UDF
CREATE TEMPORARY FUNCTION [db_name.]function_name AS class_name;
临时UDF在Inceptor重启后失效。如果需要更新临时UDF需要重启Inceptor重新创建该临时UDF。
示例 4. 验证临时UDF
SELECT [db_name.]function_name() FROM SYSTEM.DUAL; 5. 删除临时UDF
DROP TEMPORARY FUNCTION if exists function_name;
方式三 创建永久UDF
建议优先选取前两种方式此方式虽然可行但不推荐故仅介绍基础命令暂无视频提供。
1. 查看已存在jar包
LIST JAR;
2. 添加jar包
ADD JAR[S] local_or_hdfs_path;
//Local_path是Inceptor server节点的路径。保证hive用户对jar所在的目录有读权限。
3. 创建永久UDF
CREATE PERMANENT FUNCTION [db_name.]function_name AS class_name;
如果Inceptor不在local mode那么资源的地址也必须是非本地URI比如HDFS地址。
4. 验证永久UDF
SELECT [db_name.]function_name() FROM SYSTEM.DUAL; 5. 删除永久UDF
DROP PERMANENT FUNCTION if exists function_name;
