在Spark SQL作业中使用UDAF

操作场景

DLI支持用户使用Hive UDAF（User Defined Aggregation Function，用户定义聚合函数）可对多行数据产生作用，通常与groupBy联合使用；等同于SQL中常用的SUM()，AVG()，也是聚合函数。

约束限制

• 在DLI Console上执行UDAF相关操作时，需要使用自建的SQL队列。
• 跨账号使用UDAF时，除了创建UDAF函数的用户，其他用户如果需要使用时，需要先进行授权才可使用对应的UDAF函数。

  授权操作参考如下：登录DLI管理控制台，选择“ 数据管理 > 程序包管理”页面，选择对应的UDAF Jar包，单击“操作”列中的“权限管理”，进入权限管理页面，单击右上角“授权”，勾选对应权限。

• 自定义函数中引用static类或接口时，必须要加上“try catch”异常捕获，否则可能会造成包冲突，导致函数功能异常。

环境准备

在进行UDAF开发前，请准备以下开发环境。

开发流程

DLI下UDAF函数开发流程参考如下：

图1 UDAF开发流程

操作步骤

1. 新建Maven工程，配置pom文件。以下通过IntelliJ IDEA 2020.2工具操作演示。
   1. 打开IntelliJ IDEA，选择“File > New > Project”。
      图2 新建Project
      
   2. 选择Maven，Project SDK选择1.8，单击“Next”。
      图3 配置Project SDK
      
   3. 定义样例工程名和配置样例工程存储路径，单击“Create",下一步单击弹窗中的“Finish”完成工程创建。
      图4 完成Project创建
      
   4. 在pom.xml文件中添加如下配置。

      <dependencies> 
               <dependency> 
                   <groupId>org.apache.hive</groupId> 
                   <artifactId>hive-exec</artifactId> 
                   <version>1.2.1</version> 
               </dependency> 
       </dependencies>

      图5 pom文件中添加配置
      
   5. 在工程路径的“src > main > java”文件夹上鼠标右键，选择“New > Package”，新建Package和类文件。
      Package根据需要定义，本示例定义为：“com.dli.demo”

      图6 新建Package
      

      在包路径下新建Java Class文件，本示例定义为：AvgFilterUDAFDemo。

      图7 创建类
      
2. 编写UDAF函数代码。UDAF函数实现，主要注意以下几点：
   • 自定义UDAF需要继承org.apache.hadoop.hive.ql.exec.UDAF和org.apache.hadoop.hive.ql.exec.UDAFEvaluator类。函数类需要继承UDAF类，计算类Evaluator实现UDAFEvaluator接口。
   • Evaluator需要实现UDAFEvaluator的init、iterate、terminatePartial、merge、terminate这几个函数。
     • init函数实现接口UDAFEvaluator的init函数。
     • iterate接收传入的参数，并进行内部的迭代。
     • terminatePartial无参数，其为iterate函数遍历结束后，返回遍历得到的数据，terminatePartial类似于 hadoop的Combiner。
     • merge接收terminatePartial的返回结果。
     • terminate返回最终的聚集函数结果。

     详细UDAF函数实现，可以参考如下样例代码：

     package com.dli.demo;
      
     import org.apache.hadoop.hive.ql.exec.UDAF;
     import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
      
     /***
      * @jdk jdk1.8.0
      * @version 1.0
      ***/
     public class AvgFilterUDAFDemo extends UDAF {
      
         /**
          * 定义静态内部类AvgFilter
          */
         public static class PartialResult
         {
             public Long sum;
         }
      
         public static class VarianceEvaluator implements UDAFEvaluator {
      
             //初始化PartialResult对象
             private AvgFilterUDAFDemo.PartialResult partial;
      
             //创建VarianceEvaluator无参构造函数
             public VarianceEvaluator(){
      
                 this.partial = new AvgFilterUDAFDemo.PartialResult();
      
                 init();
             }
      
             /**
              * init函数类似于构造函数，用于UDAF的初始化
              */
             @Override
             public void init() {
      
                 //设置sum初始值
                 this.partial.sum = 0L;
             }
      
             /**
              * iterate接收传入的参数，并进行内部的轮转。
              * @param x
              * @return
              */
             public void iterate(Long x) {
                 if (x == null) {
                     return;
                 }
                 AvgFilterUDAFDemo.PartialResult tmp9_6 = this.partial;
                 tmp9_6.sum = tmp9_6.sum | x;
             }
      
             /**
              * terminatePartial无参数，其为iterate函数遍历结束后，返回轮转数据，
              * terminatePartial类似于hadoop的Combiner
              * @return
              */
             public AvgFilterUDAFDemo.PartialResult terminatePartial()
             {
                 return this.partial;
             }
      
             /**
              * merge接收terminatePartial的返回结果，进行数据merge操作
              * @param
              * @return
              */
             public void merge(AvgFilterUDAFDemo.PartialResult pr)
             {
                 if (pr == null) {
                     return;
                 }
                 AvgFilterUDAFDemo.PartialResult tmp9_6 = this.partial;
                 tmp9_6.sum = tmp9_6.sum | pr.sum;
             }
      
             /**
              * terminate返回最终的聚集函数结果
              * @return
              */
             public Long terminate()
             {
                 if (this.partial.sum == null) {
                     return 0L;
                 }
                 return this.partial.sum;
             }
         }
     }

     图8 编写UDAF函数代码
     
3. 编写调试完成代码后，通过IntelliJ IDEA工具编译代码并导出Jar包。
   1. 单击工具右侧的“Maven”，参考下图分别单击“clean”、“compile”对代码进行编译。

      编译成功后，单击“package”对代码进行打包。

      图9 导出jar包
      
   2. 打包成功后，生成的Jar包会放到target目录下，以备后用。本示例将会生成到：“D:\DLITest\MyUDAF\target”下名为“MyUDAF-1.0-SNAPSHOT.jar”。
4. 登录OBS控制台，将生成的Jar包文件上传到OBS路径下。
   

   Jar包文件上传的OBS桶所在的区域需与DLI的队列区域相同，不可跨区域执行操作。

5. （可选）可以将Jar包文件上传到DLI的程序包管理中，方便后续统一管理。
   1. 登录DLI管理控制台，单击“数据管理 > 程序包管理”。
   2. 在“程序包管理”页面，单击右上角的“创建”创建程序包。
   3. 在“创建程序包”对话框，配置以下参数。
      • 包类型：选择“JAR”。
      • OBS路径：程序包所在的OBS路径。
      • 分组设置和组名称根据情况选择设置，方便后续识别和管理程序包。
   4. 单击“确定”，完成创建程序包。
6. 创建UDAF函数。
   1. 登录登录DLI管理控制台，创建SQL队列和数据库。
   2. 登录DLI管理控制台，单击“SQL编辑器”，执行引擎选择“spark”，选择已创建的SQL队列和数据库。
   3. 在SQL编辑区域输入下列命令创建UDAF函数，单击“执行”提交创建。
      

      如果该客户开启了自定义函数热加载功能，注册语句会发生变化。

      CREATE FUNCTION AvgFilterUDAFDemo AS 'com.dli.demo.AvgFilterUDAFDemo' using jar 'obs://dli-test-obs01/MyUDAF-1.0-SNAPSHOT.jar';

      或

      CREATE OR REPLACE FUNCTION AvgFilterUDAFDemo AS 'com.dli.demo.AvgFilterUDAFDemo' using jar 'obs://dli-test-obs01/MyUDAF-1.0-SNAPSHOT.jar';

7. 重启原有SQL队列，使得创建的Function生效。
   1. 登录数据湖探索管理控制台，选择“资源管理”》“队列管理”，在对应“SQL队列”类型作业的“操作”列，单击“更多”》“重启”。
   2. 在“重启队列”界面，选择“确定”完成队列重启
8. 使用UDAF函数。

   在查询语句中使用6中创建的UDAF函数:

   select AvgFilterUDAFDemo(real_stock_rate) AS show_rate FROM dw_ad_estimate_real_stock_rate limit 1000;

9. （可选）删除UDAF函数。

   如果不再使用UDAF函数，可执行以下语句删除该函数:

   Drop FUNCTION AvgFilterUDAFDemo;