在Spark SQL作业中使用UDAF

操作场景

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

约束限制

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

环境准备

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

开发流程

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.aidatalake.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.aidatalake.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:\AIDataLakeTest\MyUDAF\target”下名为“MyUDAF-1.0-SNAPSHOT.jar”。
4. 登录OBS控制台，将生成的Jar包文件上传到OBS路径下。
   

   Jar包文件上传的OBS桶所在的区域需与AI DataLake的工作空间区域相同，不可跨区域执行操作。

5. 创建UDAF函数。
   1. 在AI DataLake管理控制台左下角，单击“DataArts Studio”。
   2. 在“脚本开发”页面，新建Serverless Spark SQL脚本后，选择端点，Catalog和数据库。
      图10 新建Serverless Spark SQL脚本
      
   3. 在SQL编辑区域，输入创建UDAF函数的命令。

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

      或

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

      

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

   4. 单击“运行”。
6. 使用UDAF函数。

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

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

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

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

   Drop FUNCTION AvgFilterUDAFDemo;