在使用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。   <ins class="adsbygoogle" style="display:block; text-align:center;" data-ad-layout="in-article" data-ad-format="fluid" data-ad-client="ca-pub-4353345653789615" data-ad-slot="8840342077"></ins> <script> (adsbygoogle = window.adsbygoogle || []).push({});</script>在使用Hive进行数据处理时,经常会用到group by语法,但对分组的合并操作,hive没有My 你的当前访问异常,请进行认证后继续阅读剩余内容。 提交