我们知道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类型的数组了。 附两张图理解:   <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提供了聚合连接函数concat_ws,该函数无法连接ARRAY类型。所有,在实际的 你的当前访问异常,请进行认证后继续阅读剩余内容。 提交