一：UDF

1.自定义UDF

　　

二：UDAF

2.UDAF

　　

3.介绍AbstractGenericUDAFResolver

　　

4.介绍GenericUDAFEvaluator

　　

5.程序

 package org.apache.hadoop.hive_udf;

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

 import org.apache.hadoop.hive.ql.metadata.HiveException;

 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.io.DoubleWritable;

 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;

 import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;

 import org.apache.hadoop.hive.serde2.objectinspector.primitive.AbstractPrimitiveWritableObjectInspector;

 import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

 import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;

 import org.apache.hadoop.io.LongWritable;

 /**

  *

  * 需求：实现sum函数，支持int和double类型

  *

  */

 public class UdafProject extends AbstractGenericUDAFResolver{

     public GenericUDAFEvaluator getEvaluator(GenericUDAFParameterInfo info)

             throws SemanticException {

         //判断参数是否是全部列

         if(info.isAllColumns()){

             throw new SemanticException("不支持*的参数");

         }

         //判断是否只有一个参数

         ObjectInspector[] inspector = info.getParameterObjectInspectors();

         if(inspector.length != 1){

             throw new SemanticException("参数只能有一个");

         }

         //判断输入列的数据类型是否为基本类型

         if(inspector[0].getCategory() != ObjectInspector.Category.PRIMITIVE){

             throw new SemanticException("参数必须为基本数据类型");

         }

         AbstractPrimitiveWritableObjectInspector woi = (AbstractPrimitiveWritableObjectInspector) inspector[0];

         //判断是那种基本数据类型

         switch(woi.getPrimitiveCategory()){

         case INT:

         case LONG:

         case BYTE:

         case SHORT:

             return new udafLong();

         case FLOAT:

         case DOUBLE:

             return new udafDouble();

             default:

                 throw new SemanticException("参数必须是基本类型，且不能为string等类型");

         }

     }

     /**

      * 对整形数据进行求和

      */

     public static class udafLong extends  GenericUDAFEvaluator{

         //定义输入数据类型

         public  PrimitiveObjectInspector inputor;

         //实现自定义buffer

         static class sumlongagg implements AggregationBuffer{

             long sum;

             boolean empty;

         }

         //初始化方法

         @Override

         public ObjectInspector init(Mode m, ObjectInspector[] parameters)

                 throws HiveException {

             // TODO Auto-generated method stub

             super.init(m, parameters);

             if(parameters.length !=1 ){

                 throw new UDFArgumentException("参数异常");

             }

             if(inputor == null){

                 this.inputor = (PrimitiveObjectInspector) parameters[0];

             }

             //注意返回的类型要与最终sum的类型一致

             return PrimitiveObjectInspectorFactory.writableLongObjectInspector;

         }

         @Override

         public AggregationBuffer getNewAggregationBuffer() throws HiveException {

             // TODO Auto-generated method stub

             sumlongagg slg = new sumlongagg();

             this.reset(slg);

             return slg;

         }

         @Override

         public void reset(AggregationBuffer agg) throws HiveException {

             // TODO Auto-generated method stub

             sumlongagg slg = (sumlongagg) agg;

             slg.sum=0;

             slg.empty=true;

         }

         @Override

         public void iterate(AggregationBuffer agg, Object[] parameters)

                 throws HiveException {

             // TODO Auto-generated method stub

             if(parameters.length != 1){

                 throw new UDFArgumentException("参数错误");

             }

             this.merge(agg, parameters[0]);

         }

         @Override

         public Object terminatePartial(AggregationBuffer agg)

                 throws HiveException {

             // TODO Auto-generated method stub

             return this.terminate(agg);

         }

         @Override

         public void merge(AggregationBuffer agg, Object partial)

                 throws HiveException {

             // TODO Auto-generated method stub

             sumlongagg slg = (sumlongagg) agg;

             if(partial != null){

                 slg.sum += PrimitiveObjectInspectorUtils.getLong(partial, inputor);

                 slg.empty=false;

             }

         }

         @Override

         public Object terminate(AggregationBuffer agg) throws HiveException {

             // TODO Auto-generated method stub

             sumlongagg slg = (sumlongagg) agg;

             if(slg.empty){

                 return null;

             }

             return new LongWritable(slg.sum);

         }

     }

     /**

      * 实现浮点型的求和

      */

     public static class udafDouble extends GenericUDAFEvaluator{

         //定义输入数据类型

         public  PrimitiveObjectInspector input;

         //实现自定义buffer

         static class sumdoubleagg implements AggregationBuffer{

             double sum;

             boolean empty;

         }

         //初始化方法

         @Override

         public ObjectInspector init(Mode m, ObjectInspector[] parameters)

                 throws HiveException {

             // TODO Auto-generated method stub

             super.init(m, parameters);

             if(parameters.length !=1 ){

                 throw new UDFArgumentException("参数异常");

             }

             if(input == null){

                 this.input = (PrimitiveObjectInspector) parameters[0];

             }

             //注意返回的类型要与最终sum的类型一致

             return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector;

         }

         @Override

         public AggregationBuffer getNewAggregationBuffer() throws HiveException {

             // TODO Auto-generated method stub

             sumdoubleagg sdg = new sumdoubleagg();

             this.reset(sdg);

             return sdg;

         }

         @Override

         public void reset(AggregationBuffer agg) throws HiveException {

             // TODO Auto-generated method stub

             sumdoubleagg sdg = (sumdoubleagg) agg;

             sdg.sum=0;

             sdg.empty=true;

         }

         @Override

         public void iterate(AggregationBuffer agg, Object[] parameters)

                 throws HiveException {

             // TODO Auto-generated method stub

             if(parameters.length != 1){

                 throw new UDFArgumentException("参数错误");

             }

             this.merge(agg, parameters[0]);

         }

         @Override

         public Object terminatePartial(AggregationBuffer agg)

                 throws HiveException {

             // TODO Auto-generated method stub

             return this.terminate(agg);

         }

         @Override

         public void merge(AggregationBuffer agg, Object partial)

                 throws HiveException {

             // TODO Auto-generated method stub

             sumdoubleagg sdg =(sumdoubleagg) agg;

             if(partial != null){

                 sdg.sum += PrimitiveObjectInspectorUtils.getDouble(sdg, input);

                 sdg.empty=false;

             }

         }

         @Override

         public Object terminate(AggregationBuffer agg) throws HiveException {

             // TODO Auto-generated method stub

             sumdoubleagg sdg = (sumdoubleagg) agg;

             if (sdg.empty){

                 return null;

             }

             return new DoubleWritable(sdg.sum);

         }

     }

 }

6.打成jar包

　　并放入路径：/etc/opt/datas/

7.添加jar到path

　　格式：

　　　　add jar linux_path;

　　即：

　　　　add jar /etc/opt/datas/af.jar

8.创建方法

　　create temporary function af as 'org.apache.hadoop.hive_udf.UdafProject';

9.在hive中运行

　　select sum(id),af(id) from stu_info;

三：UDTF

1.UDTF

　　

2.程序　

 package org.apache.hadoop.hive.udf;

 import java.util.ArrayList;

 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;

 public class UDTFtest extends GenericUDTF {

     @Override

     public StructObjectInspector initialize(StructObjectInspector argOIs)

             throws UDFArgumentException {

         // TODO Auto-generated method stub

         if(argOIs.getAllStructFieldRefs().size() != 1){

             throw new UDFArgumentException("参数只能有一个");

         }

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

         fieldname.add("name");

         fieldname.add("email");

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

         fieldio.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

         fieldio.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

         return ObjectInspectorFactory.getStandardStructObjectInspector(fieldname, fieldio);

     }

     @Override

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

         // TODO Auto-generated method stub

         if(args.length == 1){

             String name = args[0].toString();

             String email = name + "@ibeifneg.com";

             super.forward(new String[] {name,email});

         }

     }

     @Override

     public void close() throws HiveException {

         // TODO Auto-generated method stub

         super.forward(new String[] {"complete","finish"});

     }

 }

3.同样的步骤

4.在hive中运行

　　select tf(ename) as (name,email) from emp;

最新文章

  1. 给numpy矩阵添加一列
  2. 浅析session&amp;cookie
  3. Java 同步代码块 - Synchronized Blocks
  4. Samba快速配置
  5. PHP Switch case 条件并用实例
  6. maven依赖本地非repository中的jar包
  7. jsp学习--JavaBean定义和在Jsp中使用JavaBean
  8. [STL][C++]MAP
  9. 【转】如何在ubuntu12.04设置adb驱动
  10. AngularJs学习(1)
  11. Oracle EBS-SQL (AR-2):检查应收收款核销额
  12. js去除首尾空格
  13. 百度CSND博客在搜索栏中显示图片
  14. List list = new ArrayList()
  15. CSS3文本
  16. 每天学习点jquery
  17. Linux系统xinetd服务启动不了
  18. VS2015在win10上编译的程序不能在Win7上运行的原因
  19. WIN32控件使用系统样式
  20. 进阶之路（基础篇） - 004 I/O的模拟量输出

热门文章

  1. ACM-ICPC 2018 徐州赛区网络预赛 G题
  2. UML 类图 说明
  3. 让webstrom更好用的设置
  4. 关于django1.7.7使用ajax后出现“CSRF token missing or incorrect”问题的解决办法
  5. Maven私服
  6. yum和rpm的区别
  7. ROS 错误之 [rospack] Error: package &#39;beginner_tutorials&#39; not found
  8. Linux用户组相关指令
  9. MVC自定义视图引擎地址
  10. zabbix3.0.4关于java服务端程序内存溢出的处理