在Spark SQL作业中使用UDTF

操作场景

DLI支持用户使用Hive UDTF（User-Defined Table-Generating Functions）自定义表值函数，UDTF用于解决一进多出业务场景，即其输入与输出是一对多的关系，读入一行数据，输出多个值。

约束限制

• 在DLI Console上执行UDTF相关操作时，需要使用自建的SQL队列。
• 不同的IAM用户使用UDTF时，除了创建UDTF函数的用户，其他用户如果需要使用时，需要先进行授权才可使用对应的UDTF函数。授权操作参考如下：

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

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

环境准备

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

开发流程

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

图1 UDTF开发流程

表2 开发流程说明

序号	阶段	操作界面	说明
1	新建Maven工程，配置pom文件	IntelliJ IDEA	参考操作步骤说明，编写UDTF函数代码。
2	编写UDTF函数代码
3	调试，编译代码并导出Jar包
4	上传Jar包到OBS	OBS控制台	将生成的UDTF函数Jar包文件上传到OBS目录下。
5	创建DLI的UDTF函数	DLI控制台	在DLI控制台的SQL作业管理界面创建使用的UDTF函数。
6	验证和使用DLI的UDTF函数	DLI控制台	在DLI作业中使用创建的UDTF函数。

操作步骤

1. 新建Maven工程，配置pom文件。以下通过IntelliJ IDEA 2020.2工具操作演示。
   1. 打开IntelliJ IDEA，选择“File > New > Project”。
      图2 新建Project
      
   2. 选择Maven，Project SDK选择1.8，单击“Next”。
      图3 选择Maven
      
   3. 定义样例工程名和配置样例工程存储路径，单击“Finish”完成工程创建。
      图4 创建工程
      
   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和类文件。
      图6 新建Package和类文件
      

      Package根据需要定义，本示例定义为：“com.huawei.demo”，完成后回车。

      图7 自定义Package
      

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

      图8 新建Java Class文件
      
2. 编写UDTF函数代码。完整样例代码请参考样例代码。

   UDTF的类需要继承“org.apache.hadoop.hive.ql.udf.generic.GenericUDTF”，实现initialize，process，close三个方法。

   1. UDTF首先会调用initialize方法，此方法返回UDTF的返回行的信息，如，返回个数，类型等。
   2. 初始化完成后，会调用process方法，真正处理在process函数中，在process中，每一次forward()调用产生一行。
      如果产生多列可以将多个列的值放在一个数组中，然后将该数组传入到forward()函数。

      public void process(Object[] args) throws HiveException {
              // TODO Auto-generated method stub
              if(args.length == 0){
                  return;
              }
              String input = args[0].toString();
              if(StringUtils.isEmpty(input)){
                  return;
              }
              String[] test = input.split(";");
              for (int i = 0; i < test.length; i++) {
                  try {
                      String[] result = test[i].split(":");
                      forward(result);
                  } catch (Exception e) {
                      continue;
                  }
              }
      
          }

   3. 最后调用close方法，对需要清理的方法进行清理。
3. 编写调试完成代码后，通过IntelliJ IDEA工具编译代码并导出Jar包。
   1. 单击工具右侧的“Maven”，参考下图分别单击“clean”、“compile”对代码进行编译。
      编译成功后，单击“package”对代码进行打包。

      图9 编译打包
      

      打包成功后，生成的Jar包会放到target目录下，以备后用。本示例将会生成到：“D:\MyUDTF\target”下名为“MyUDTF-1.0-SNAPSHOT.jar”。

      图10 生成Jar包
      
4. 登录OBS控制台，将生成的Jar包文件上传到OBS路径下。
   

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

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

      CREATE FUNCTION mytestsplit AS 'com.huawei.demo.UDTFSplit' using jar 'obs://dli-test-obs01/MyUDTF-1.0-SNAPSHOT.jar';

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

   在查询语句中使用6中创建的UDTF函数，如：

   select mytestsplit('abc:123\;efd:567\;utf:890');

   图13 执行结果
   
9. （可选）删除UDTF函数。

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

   Drop FUNCTION mytestsplit;

样例代码

UDTFSplit.java完整的样例代码参考如下所示：

import java.util.ArrayList;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

public class UDTFSplit extends GenericUDTF {

    @Override
    public void close() throws HiveException {
        // TODO Auto-generated method stub

    }

    @Override
    public void process(Object[] args) throws HiveException {
        // TODO Auto-generated method stub
        if(args.length == 0){
            return;
        }
        String input = args[0].toString();
        if(StringUtils.isEmpty(input)){
            return;
        }
        String[] test = input.split(";");
        for (int i = 0; i < test.length; i++) {
            try {
                String[] result = test[i].split(":");
                forward(result);
            } catch (Exception e) {
                continue;
            }
        }

    }

    @Override
    public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {
        if (args.length != 1) {
            throw new UDFArgumentLengthException("ExplodeMap takes only one argument");
        }
        if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {
            throw new UDFArgumentException("ExplodeMap takes string as a parameter");
        }

        ArrayList<String> fieldNames = new ArrayList<String>();
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldNames.add("col1");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldNames.add("col2");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

}