使用Hive中的UDAF实现GROUP_CONCAT函数合并ARRAY数组类型

[原创] 使用Hive中的UDAF实现GROUP_CONCAT函数合并ARRAY数组类型

仧尐 2018/04/04 34 UDAF/ 合并/ array/ group_concat/ 数组/ 实现/ 类型

我们知道Hive提供了聚合连接函数concat_ws，该函数无法连接ARRAY类型。所有，在实际的开发过程中有可能需要聚合连接ARRAY类型。比如说，同一个用户的标签进行group by后，进行连接，并去重。
  所以，借助于Hive的UDAF函数，我们实现类似的临时函数功能。
  废话不多说，直接上代码，相关解释也写在代码里：
  ```java
package org.apache.hadoop.hive.udf;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
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.GenericUDAFEvaluator.AggregationBuffer;
import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.log4j.Logger;

@SuppressWarnings({"deprecation","unchecked"})
public class GroupConcatInSetUdf extends AbstractGenericUDAFResolver{
	public static Logger log=Logger.getLogger(GroupConcatInSetUdf.class);
	@Override  
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)  
            throws SemanticException {  
        if (parameters.length!=1) {  
            throw new UDFArgumentTypeException(parameters.length - 1,  
                    "Exactly one argument is expected.");  
        }  
        return new GroupConcatInSetEvaluator();  
    }  
    
	//公共内部函数，用于记录每个分组的group by结果set
	public static class GroupConcatBuffer implements AggregationBuffer{
		Set container=new HashSet();
	}
	@SuppressWarnings("rawtypes")
    public static class GroupConcatInSetEvaluator extends GenericUDAFEvaluator {
    	ListObjectInspector inputOI;  
        ObjectInspector outputOI;  
        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
        	assert (parameters.length == 1);  
            super.init(m, parameters);  
            inputOI =(ListObjectInspector)parameters[0];
			//第一个阶段为map阶段，直接把数据转为标准类型输出就行了。
            if(m==Mode.PARTIAL1){
	        	outputOI = ObjectInspectorFactory.getStandardListObjectInspector(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            }else{//第二个阶段把上个阶段的输出转为List类型
            	outputOI=(StandardListObjectInspector) ObjectInspectorUtils.getStandardObjectInspector(inputOI);
            }
        	return outputOI;
        };
		@Override
		public AggregationBuffer getNewAggregationBuffer() throws HiveException {
			return new GroupConcatBuffer();
		}

//第一个init执行完后，马上执行iterate，把每个元素的值放入给定的List中
		@Override
		public void iterate(AggregationBuffer arg0, Object[] arg1)
				throws HiveException {
			if(arg1 == null || arg1.length != 1){  
	              return;  
	          }  
	          Object v = arg1[0];  
	          if (v != null) {  
	        	  GroupConcatBuffer myagg = (GroupConcatBuffer)arg0;  
	        	  List pCopy = (List)ObjectInspectorUtils.copyToStandardObject(v,this.inputOI);
	              myagg.container.addAll(pCopy);  
	          }
		}

//第二，三个阶段执行，第二个阶段为，本机节点的group by，第三个阶段为洗牌后的group by。
		@Override
		public void merge(AggregationBuffer arg0, Object arg1)
				throws HiveException {
			if(arg1==null)
				return;
			GroupConcatBuffer myagg = (GroupConcatBuffer) arg0;  
			List pCopy = (List)ObjectInspectorUtils.copyToStandardObject(arg1,this.inputOI);
			myagg.container.addAll(pCopy);
			return; 
		}

@Override
		public void reset(AggregationBuffer arg0) throws HiveException {
			((GroupConcatBuffer)arg0).container.clear();;
		}

@Override
		public Object terminate(AggregationBuffer agg) throws HiveException {
			GroupConcatBuffer myAgg = (GroupConcatBuffer) agg;  
            return myAgg.container;  
		}

@Override
		public Object terminatePartial(AggregationBuffer arg0)
				throws HiveException {
            return terminate(arg0);  
		}  
    }
	
}

```
将代码打包成jar，然后接下来就是加载jar包，创建自定义函数。
```sql
add jar /data0/custom/lib/hiveUDF.jar
create temporary function group_concat as 'org.apache.hadoop.hive.udf.GroupConcatInSetUdf'
```
至此就可以使用group_concat函数来聚合两个array类型的数组了。

附两张图理解：
![Model各阶段对应Evaluator方法调用](http://img.blog.csdn.net/20151130115312010)
![Evaluator各个阶段下处理mapreduce流程](http://img.blog.csdn.net/20151130115324649)

注意！

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

猜您在找

postgreSQL中的array_to_string函数和如何实现group_concat MySql中group_concat()函数的使用 MySQL中group_concat()函数的使用 mysql中函数DISTINCT，group by，CONCAT及GROUP_CONCAT的使用 MySQL中的数组切片/ group_concat限制 - Array slicing / group_concat limiting in MySQL