Hive使用UDAF自定义聚合函数

[原创] Hive使用UDAF自定义聚合函数

仧尐 2018/02/28 36 hive/ 聚合/ UDAF/ 使用/ 定义/ 函数/ 自定义

在使用Hive进行数据处理时，经常会用到group by语法，但对分组的合并操作，hive没有MySQL支持得好：
group_concat([DISTINCT] 要连接的字段 Order BY ASC/DESC 排序字段 Separator '分隔符'

hive只有一个collect_set内置函数，返回去重后的元素数组，但我们可以通过编写UDAF，来实现想要的功能。
编写通用型UDAF需要两个类：解析器和计算器。

解析器负责UDAF的参数检查，操作符的重载以及对于给定的一组参数类型来查找正确的计算器，建议继承AbstractGenericUDAFResolver类，具体实现如下：

```java
	@Override
	public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException {
		if (parameters.length  container = Lists.newArrayList();
        }
        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
            ((MkListAggregationBuffer) agg).container.clear();
        }
        @Override
        public AggregationBuffer getNewAggregationBuffer() throws HiveException {
            MkListAggregationBuffer ret = new MkListAggregationBuffer();
            return ret;
        }
        @Override
        public void iterate(AggregationBuffer agg, Object[] parameters)
                throws HiveException {
            if(parameters == null || parameters.length != 1){
                return;
            }
            Object key = parameters[0];
            if (key != null) {
                MkListAggregationBuffer myagg = (MkListAggregationBuffer) agg;
                putIntoList(key, myagg.container);
            }
        }

private void putIntoList(Object key, List container) {
            Object pCopy = ObjectInspectorUtils.copyToStandardObject(key,  this.inputKeyOI);
            container.add(pCopy);
        }

@Override
        public Object terminatePartial(AggregationBuffer agg)
                throws HiveException {
            MkListAggregationBuffer myagg = (MkListAggregationBuffer) agg;
            List ret = Lists.newArrayList(myagg.container);
            return ret;
        }
        @Override
        public void merge(AggregationBuffer agg, Object partial)
                throws HiveException {
            if(partial == null){
                return;
            }
            MkListAggregationBuffer myagg = (MkListAggregationBuffer) agg;
            List partialResult = (List) internalMergeOI.getList(partial);
            for (Object ob: partialResult) {
                putIntoList(ob, myagg.container);
            }
            return;
        }

@Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            MkListAggregationBuffer myagg = (MkListAggregationBuffer) agg;
            Map map = Maps.newHashMap();
            for (int i = 0; i> listData = Lists.newArrayList(map.entrySet());
            Collections.sort(listData, new Comparator>() {
                public int compare(Map.Entry o1, Map.Entry o2) {
                    if (o1.getValue()  ret =  Lists.newArrayList();
            for(Map.Entry entry : listData){
                ret.add(entry.getKey());
                ret.add(new Text(entry.getValue().toString()));
            }
            return ret;
        }
    }
```

使用方法：

```mysql
add jar /export/data/hiveUDF.jar;
create temporary function collect_list  as 'com.test.hive.udf.CollectListUDAF';
select id, collect_list(value) from test group by id;
```

test表中数据为：

```shell
+------+-------+
| id   | value |
+------+-------+
|    1 | a     |
|    1 | a     |
|    1 | b     |
|    2 | c     |
|    2 | d     |
|    2 | d     |
+------+-------+
```

运行结果为：

```shell
1    ["a", "2", "b", "1"]
2    ["d", "2", "c", "1"]  
```

附：用Map实现的话会比较简单，但在数据量大的时候，统计的数据有些会不准确（在只有一个map/reduce时，统计无误），最后我也没有找到好的解决办法，所以这里用的ArrayList。

![Model各阶段对应Evaluator方法调用](http://img.blog.csdn.net/20151130115312010)

![Evaluator各个阶段下处理mapreduce流程](http://img.blog.csdn.net/20151130115324649)

智能推荐

注意！

本站用户原创文章，版权为本站用户，转摘请注明本文地址：http://www.silva-art.net/blog/2018/02/28/d6857b944675a27f5c3a137c726da0fe.html。

猜您在找

Hive用户自定义聚合函数(UDAF) Hive通用型自定义聚合函数（UDAF） Hive自定义UDF和聚合函数UDAF Hive自定义函数UDAF开发 hive 自定义函数UDF,UDAF

赞助商链接

[原创] Hive使用UDAF自定义聚合函数

注意！

赞助商广告