Chinaunix首页 | 论坛 | 博客

宝马追猪

首页　| 　博文目录　| 　关于我

jelon521

  • 博客访问： 1062543
  • 博文数量： 165
  • 博客积分： 0
  • 博客等级： 民兵
  • 技术积分： 1346
  • 用 户 组： 普通用户
  • 注册时间： 2016-03-11 14:13
个人简介

狂甩酷拽吊炸天

文章分类

全部博文（165）

文章存档

2024年（1）

2023年（1）

2022年（3）

2021年（4）

2020年（17）

2019年（37）

2018年（17）

2017年（35）

2016年（50）

我的朋友
最近访客
推荐博文
相关博文
Hive UDAF 函数的编写

分类： LINUX

2019-11-06 11:55:21

hive的比较难的部分应该就是GenericUDAF，看了两天终于看明白了，有些点是我自己遇到卡住的点，记录下来希望对大家有所帮助。


一开始看的是《Hive 编程指南》中关于GenericUDAF的章节，例子有点难了。讲的是group_concat的实现。查了资料后觉得网上写的博客非常好，例子比较简单，更能够明白到底在说什么。


一定要结合MapReduce的过程来看，才会想明白。


不要偷懒，要将文章中的代码下载下来，具体查看他所继承的类，才能更快速理解内容。

关于ObjectInspector
想要理解GenericUDAF一定要理解ObjectInspector，这是贯彻在其中的一个概念。这里也说了一篇博文，一定要参考。Hive中ObjectInspector的作用。
ObjectInspector在我看来是什么？其实就是类型检查和通过这个类能得到值，并且帮助它序列化和反序列化，达到数据流转无障碍的目的。


UDAF是Hive中用户自定义的聚集函数，Hive内置UDAF函数包括有sum()与count（），UDAF实现有简单与通用两种方式，简单UDAF因为使用Java反射导致性能损失，而且有些特性不能使用，已经被弃用了；在这篇博文中我们将关注Hive中自定义聚类函数-GenericUDAF，UDAF开发主要涉及到以下两个抽象类：

点击(此处)折叠或打开

  1. org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver
  2. org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator


如果你想浏览代码：fork it on Github：

示例数据准备
首先先创建一张包含示例数据的表：people，该表只有name一列，该列中包含了一个或多个名字，该表数据保存在people.txt文件中。

点击(此处)折叠或打开

  1. ~$ cat ./people.txt
  2. John Smith
  3. John and Ann White
  4. Ted Green
  5. Dorothy


把该文件上载到HDFS目录/user/matthew/people中：

点击(此处)折叠或打开

  1. hadoop fs -mkdir people
  2. hadoop fs -put ./people.txt people


下面要创建Hive外部表，在Hive shell中执行

点击(此处)折叠或打开

  1. CREATE EXTERNAL TABLE people (name string)
  2. ROW FORMAT DELIMITED FIELDS
  3. TERMINATED BY '\t'
  4. ESCAPED BY ''
  5. LINES TERMINATED BY '\n'
  6. STORED AS TEXTFILE
  7. LOCATION '/user/matthew/people';



相关抽象类介绍
创建一个GenericUDAF必须先了解以下两个抽象类：

点击(此处)折叠或打开

  1. org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver
  2. org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator


为了更好理解上述抽象类的API，要记住hive只是mapreduce函数，只不过hive已经帮助我们写好并隐藏mapreduce，向上提供简洁的sql函数，所以我们要结合Mapper、Combiner与Reducer来帮助我们理解这个函数。要记住在hadoop集群中有若干台机器，在不同的机器上Mapper与Reducer任务独立运行。

所以大体上来说，这个UDAF函数读取数据（mapper），聚集一堆mapper输出到部分聚集结果（combiner），并且最终创建一个最终的聚集结果（reducer）。因为我们跨域多个combiner进行聚集，所以我们需要保存部分聚集结果。

AbstractGenericUDAFResolver
Resolver很简单，要覆盖实现下面方法，该方法会根据sql传人的参数数据格式指定调用哪个Evaluator进行处理。

点击(此处)折叠或打开

  1. public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException;
GenericUDAFEvaluator


UDAF逻辑处理主要发生在Evaluator中，要实现该抽象类的几个方法。
在理解Evaluator之前，必须先理解objectInspector接口与GenericUDAFEvaluator中的内部类Model。
ObjectInspector
作用主要是解耦数据使用与数据格式，使得数据流在输入输出端切换不同的输入输出格式，不同的Operator上使用不同的格式。可以参考这两篇文章：first post on Hive UDFs、Hive中ObjectInspector的作用，里面有关于objectinspector的介绍。
Model
Model代表了UDAF在mapreduce的各个阶段。

点击(此处)折叠或打开

  1. public static enum Mode {
  2. /**
  3.  * PARTIAL1: 这个是mapreduce的map阶段:从原始数据到部分数据聚合
  4.  * 将会调用iterate()和terminatePartial()
  5.  */
  6. PARTIAL1,
  7. /**
  8.  * PARTIAL2: 这个是mapreduce的map端的Combiner阶段，负责在map端合并map的数据::从部分数据聚合到部分数据聚合:
  9.  * 将会调用merge() 和 terminatePartial()
  10.  */
  11. PARTIAL2,
  12. /**
  13.  * FINAL: mapreduce的reduce阶段:从部分数据的聚合到完全聚合
  14.  * 将会调用merge()和terminate()
  15.  */
  16. FINAL,
  17. /**
  18.  * COMPLETE: 如果出现了这个阶段，表示mapreduce只有map，没有reduce，所以map端就直接出结果了:从原始数据直接到完全聚合
  19.  * 将会调用 iterate()和terminate()
  20.  */
  21. COMPLETE
  22. }; 

		
    
	
  1. 
		
    
		
    
			
			
    
				
				
    
					
					
    
						 一般情况下，完整的UDAF逻辑是一个mapreduce过程，如果有mapper和reducer，就会经历PARTIAL1(mapper)，FINAL(reducer)，如果还有combiner，那就会经历PARTIAL1(mapper)，PARTIAL2(combiner)，FINAL(reducer)。 
					
    

				
    

			
    

		
    

			
    

				
    2.
而有一些情况下的mapreduce，只有mapper，而没有reducer，所以就会只有COMPLETE阶段，这个阶段直接输入原始数据，出结果。

点击(此处)折叠或打开

  1. GenericUDAFEvaluator的方法
  2. // 确定各个阶段输入输出参数的数据格式ObjectInspectors
  3. public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException;
  4. // 保存数据聚集结果的类
  5. abstract AggregationBuffer getNewAggregationBuffer() throws HiveException;
  6. // 重置聚集结果
  7. public void reset(AggregationBuffer agg) throws HiveException;
  8. // map阶段，迭代处理输入sql传过来的列数据
  9. public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException;
  10. // map与combiner结束返回结果，得到部分数据聚集结果
  11. public Object terminatePartial(AggregationBuffer agg) throws HiveException;
  12. // combiner合并map返回的结果，还有reducer合并mapper或combiner返回的结果。
  13. public void merge(AggregationBuffer agg, Object partial) throws HiveException;
  14. // reducer阶段，输出最终结果
  15. public Object terminate(AggregationBuffer agg) throws HiveException;


图解Model与Evaluator关系

实例
下面将讲述一个聚集函数UDAF的实例，我们将计算people这张表中的name列字母的个数。
下面的函数代码是计算指定列中字符的总数（包括空格）

pom文件如下：

点击(此处)折叠或打开

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns=""
  3.          xmlns:xsi=""
  4.          xsi:schemaLocation=" ">
  5.     <modelVersion>4.0.0</modelVersion>

  7.     <groupId>TotalNumOfLetters</groupId>
  8.     <artifactId>com.xxxx.udaf</artifactId>
  9.     <version>1.0-SNAPSHOT</version>

  11.     <dependencies>
  12.         <dependency>
  13.             <groupId>org.apache.hive</groupId>
  14.             <artifactId>hive-exec</artifactId>
  15.             <version>2.6.0</version>
  16.         </dependency>

  18.         <dependency>
  19.             <groupId>org.apache.hadoop</groupId>
  20.             <artifactId>hadoop-client</artifactId>
  21.             <version>2.6.0</version>
  22.         </dependency>
  23.     </dependencies>

  25.     <build>
  26.         <plugins>
  27.             <plugin>
  28.                 <groupId>org.apache.maven.plugins</groupId>
  29.                 <artifactId>maven-jar-plugin</artifactId>
  30.                 <configuration>
  31.                     <archive>
  32.                         <manifest>
  33.                             <mainClass>com.xxxx.udaf.xxxx</mainClass>
  34.                         </manifest>
  35.                     </archive>
  36.                 </configuration>
  37.             </plugin>
  38.             <plugin>
  39.                 <groupId>com.jolira</groupId>
  40.                 <artifactId>onejar-maven-plugin</artifactId>
  41.                 <version>1.4.4</version>
  42.                 <executions>
  43.                     <execution>
  44.                         <configuration>
  45.                             <attachToBuild>true</attachToBuild>
  46.                             <classifier>onejar</classifier>
  47.                         </configuration>
  48.                         <goals>
  49.                             <goal>one-jar</goal>
  50.                         </goals>
  51.                     </execution>
  52.                 </executions>
  53.             </plugin>
  54.             <plugin>
  55.                 <groupId>org.apache.maven.plugins</groupId>
  56.                 <artifactId>maven-compiler-plugin</artifactId>
  57.                 <configuration>
  58.                     <source>7</source>
  59.                     <target>7</target>
  60.                 </configuration>
  61.             </plugin>
  62.         </plugins>
  63.     </build>

  65. </project>



代码

点击(此处)折叠或打开

  1. package com.xxxx.udaf;

  3. import org.apache.hadoop.hive.ql.exec.Description;
  4. import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
  5. import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
  6. import org.apache.hadoop.hive.ql.metadata.HiveException;
  7. import org.apache.hadoop.hive.ql.parse.SemanticException;
  8. import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
  9. import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
  10. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
  11. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
  12. import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
  13. import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
  14. import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;


  17. @Description(name = "letters", value = "__FUNC__(expr) - return the total count chars of the column(返回该列中所有字符串的字符总数)")
  18. public class TotalNumOfLettersGenericUDAF extends AbstractGenericUDAFResolver {

  20.     @Override
  21.     public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException {
  22.         if (parameters.length != 1) { // 判断参数长度
  23.             throw new UDFArgumentLengthException("Exactly one argument is expected, but " +
  24.                     parameters.length + " was passed!");
  25.         }

  27.         ObjectInspector objectInspector = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(parameters[0]);

  29.         if (objectInspector.getCategory() != ObjectInspector.Category.PRIMITIVE) { // 是不是标准的java Object的primitive类型
  30.             throw new UDFArgumentTypeException(0, "Argument type must be PRIMARY. but " +
  31.                     objectInspector.getCategory().name() + " was passed!");
  32.         }

  34.         // 如果是标准的java Object的primitive类型，说明可以进行类型转换
  35.         PrimitiveObjectInspector in putOI = (PrimitiveObjectInspector) objectInspector;

  37.         // 如果是标准的java Object的primitive类型,判断是不是string类型，因为参数只接受string类型
  38.         if (in putOI.getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING) {
  39.             throw new UDFArgumentTypeException(0, "Argument type must be Strig, but " +
  40.                     in putOI.getPrimitiveCategory().name() + " was passed!");
  41.         }

  43.         return new TotalNumOfLettersEvaluator();
  44.     }

  46.     public static class TotalNumOfLettersEvaluator extends GenericUDAFEvaluator {

  48.         PrimitiveObjectInspector in putIO;
  49.         ObjectInspector outputIO;
  50.         PrimitiveObjectInspector IntegerIO;

  52.         int total = 0;

  54.         @Override
  55.         public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
  56.             assert (parameters.length == 1);
  57.             super.init(m, parameters);

  59.             /**
  60.              * PARTIAL1: 这个是mapreduce的map阶段:从原始数据到部分数据聚合
  61.              * 将会调用iterate()和terminatePartial()

  63.              * PARTIAL2: 这个是mapreduce的map端的Combiner阶段，负责在map端合并map的数据::从部分数据聚合到部分数据聚合:
  64.              * 将会调用merge() 和 terminatePartial()

  66.              * FINAL: mapreduce的reduce阶段:从部分数据的聚合到完全聚合
  67.              * 将会调用merge()和terminate()

  69.              * COMPLETE: 如果出现了这个阶段，表示mapreduce只有map，没有reduce，所以map端就直接出结果了:从原始数据直接到完全聚合
  70.              * 将会调用 iterate()和terminate()
  71.              */

  73.             //map阶段读取sql列，输入为String基础数据格式
  74.             if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) {
  75.                 in putIO = (PrimitiveObjectInspector) parameters[0];
  76.             } else { //其余阶段，输入为Integer基础数据格式
  77.                 IntegerIO = (PrimitiveObjectInspector) parameters[0];
  78.             }

  80.             // 指定各个阶段输出数据格式都为Integer类型
  81.             outputIO = ObjectInspectorFactory.getReflectionObjectInspector(Integer.class,
  82.                     ObjectInspectorFactory.ObjectInspectorOptions.JAVA);
  83.             return outputIO;
  84.         }

  86.         /**
  87.          * 存储当前字符总数的类
  88.          */
  89.         static class LetterSumAgg implements AggregationBuffer {
  90.             int sum = 0;

  92.             void add(int num) {
  93.                 sum += num;
  94.             }
  95.         }

  97.         @Override
  98.         public AggregationBuffer getNewAggregationBuffer() throws HiveException {
  99.             LetterSumAgg result = new LetterSumAgg();
  100.             return result;
  101.         }

  103.         @Override
  104.         public void reset(AggregationBuffer aggregationBuffer) throws HiveException {
  105.             LetterSumAgg myAgg = new LetterSumAgg();
  106.         }

  108.         private boolean warned = false;

  110.         @Override
  111.         public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {
  112.             assert (parameters.length == 1);
  113.             if (parameters[0] != null) {
  114.                 LetterSumAgg myAgg = (LetterSumAgg) agg;
  115.                 Object p = in putIO.getPrimitiveJavaObject(parameters[0]);
  116.                 myAgg.add(String.valueOf(p).length());
  117.             }
  118.         }

  120.         @Override
  121.         public Object terminatePartial(AggregationBuffer agg) throws HiveException {
  122.             LetterSumAgg myAgg = (LetterSumAgg) agg;
  123.             total += myAgg.sum;
  124.             return total;
  125.         }

  127.         @Override
  128.         public void merge(AggregationBuffer agg, Object partial) throws HiveException {
  129.             if (partial != null) {
  130.                 LetterSumAgg myAgg1 = (LetterSumAgg) agg;
  131.                 Integer partialSum = (Integer) IntegerIO.getPrimitiveJavaObject(partial);
  132.                 LetterSumAgg myAgg2 = new LetterSumAgg();
  133.                 myAgg2.add(partialSum);
  134.                 myAgg1.add(myAgg2.sum);
  135.             }
  136.         }

  138.         @Override
  139.         public Object terminate(AggregationBuffer agg) throws HiveException {
  140.             LetterSumAgg myAgg = (LetterSumAgg) agg;
  141.             total = myAgg.sum;
  142.             return myAgg.sum;
  143.         }
  144.     }

  146. }


使用自定义函数
		
    
					
  1. 
						
    
						
    
							
						
    

							
    

    
							
    
								
								
    
									
									
    
										
										
    
											点击(此处)折叠或打开
										
    

									
    

									
    
										
										
      
											
											
    1. 
												ADD JAR ./hive-extension-examples-master/target/hive-extensions-1.0-SNAPSHOT-jar-with-dependencies.jar;  
											
      2. 

											
    2. 
												CREATE TEMPORARY FUNCTION letters as 'com.xxxx.udaf.TotalNumOfLettersGenericUDAF'; 
											
      3. 

											
    3. 
												SELECT letters(name) FROM people; 
											
      4. 

											
    4. 
												OK
											
      5. 

											
    5. 
												44
											
      6. 

											
    6. 
												Time taken: 20.688 seconds
											
      7. 

										
    

									
    

								
    

							
    

						
    

					
    2.


阅读(2926) | 评论(0) | 转发(0) |
0

上一篇：Hive中ObjectInspector的作用

下一篇：sed替换指定次数的某字符串

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6