# 在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开发前，请准备以下开发环境。 表1UDAF开发环境 | 准备项 | 说明 | |:---|:---| | 操作系统 | Windows系统，支持Windows7以上版本。 | | 安装JDK | JDK使用1.8版本（访问[Java官网](https://www.oracle.com/java/technologies/javase-downloads.html)）。 | | 安装和配置IntelliJ IDEA | [IntelliJ IDEA](https://www.jetbrains.com/idea/)为进行应用开发的工具，版本要求使用2019.1或其2019.1往后的版本。 | | 安装Maven | 开发环境的基本配置（[下载](https://maven.apache.org/download.cgi)并[安装 Maven](https://maven.apache.org/install.html)）。用于项目管理，贯穿软件开发生命周期。 | #### 开发流程 DLI下UDAF函数开发流程参考如下： 图1UDAF开发流程 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001487274748.png "点击放大") 表2开发流程说明 | 序号 | 阶段 | 操作界面 | 说明 | |:---|:---|:---|:---| | 1 | 新建Maven工程，配置pom文件 | IntelliJ IDEA | 参考[操作步骤]说明，编写UDAF函数代码。 | | 2 | 编写UDAF函数代码 | IntelliJ IDEA | 参考[操作步骤]说明，编写UDAF函数代码。 | | 3 | 调试，编译代码并导出Jar包 | IntelliJ IDEA | 参考[操作步骤]说明，编写UDAF函数代码。 | | 4 | 上传Jar包到OBS | OBS控制台 | 将生成的UDAF函数Jar包文件上传到OBS目录下。 | | 5 | 创建DLI程序包 | DLI控制台 | 选择刚上传到OBS的UDAF函数的Jar文件，由DLI进行纳管。 | | 6 | 创建DLI的UDAF函数 | DLI控制台 | 在DLI控制台的SQL作业管理界面创建使用的UDAF函数。 | | 7 | 验证和使用DLI的UDAF函数 | DLI控制台 | 在DLI作业中使用创建的UDAF函数。 | #### 操作步骤 1. 新建Maven工程，配置pom文件。以下通过IntelliJ IDEA 2020.2工具操作演示。 1. 打开IntelliJ IDEA，选择"File \> New \> Project"。 图2新建Project ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001538394645.png "点击放大") 2. 选择Maven，Project SDK选择1.8，单击"Next"。 图3配置Project SDK ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001487114864.png "点击放大") 3. 定义样例工程名和配置样例工程存储路径，单击"Create",下一步单击弹窗中的"Finish"完成工程创建。 图4完成Project创建 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001538514573.png "点击放大") 4. 在pom.xml文件中添加如下配置。 ``` org.apache.hive hive-exec 1.2.1 ``` 图5pom文件中添加配置 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001487434660.png "点击放大") 5. 在工程路径的"src \> main \> java"文件夹上鼠标右键，选择"New \> Package"，新建Package和类文件。 Package根据需要定义，本示例定义为："com.dli.demo" 图6新建Package ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001487594580.png "点击放大") 在包路径下新建Java Class文件，本示例定义为：AvgFilterUDAFDemo。 图7创建类 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001538354737.png "点击放大") 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函数代码 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001487274752.png "点击放大") 3. 编写调试完成代码后，通过IntelliJ IDEA工具编译代码并导出Jar包。 1. 单击工具右侧的"Maven"，参考下图分别单击"clean"、"compile"对代码进行编译。 编译成功后，单击"package"对代码进行打包。 图9导出jar包 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001538394649.png "点击放大") 2. 打包成功后，生成的Jar包会放到target目录下，以备后用。本示例将会生成到："D:\\DLITest\\MyUDAF\\target"下名为"MyUDAF-1.0-SNAPSHOT.jar"。 4. 登录OBS控制台，将生成的Jar包文件上传到OBS路径下。 ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png) 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函数，单击"执行"提交创建。 ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png) 如果该客户开启了自定义函数热加载功能，注册语句会发生变化。 详情可参考[注册函数](https://support.huaweicloud.com/sqlref-spark-dli/dli_08_0283.html)。 ``` 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; ```