UDAF（user defined aggregation function）

自定义udaf函数self_count，实现系统udaf count的功能

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;

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; 

        } 

    } 

}

布署步骤

跟udf完全一致

加载jar包、声明函数、使用函数

跟udf完全一致

测试运行

与count一样，使用前边定义的临时udaf函数。

案例2

自定义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();

        }

    }

}

 布署过程与之前相同

测试脚本

select

id,username,score_agg(course,score)

from

student_score

group by id,username;

 UDTF(User-Defined Table-Generating Functions

解决一行输入多行输出，即1:n，即行转列应用

往往被lateral view explode+udf等替代实现，比直接用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;