Calling UDTFs in Spark SQL Jobs

Scenario

You can use Hive User-Defined Table-Generating Functions (UDTF) to customize table-valued functions. Hive UDTFs are used for the one-in-multiple-out data operations. UDTF reads a row of data and output multiple values.

Constraints

• To perform UDTF-related operations on DLI, you need to create a SQL queue instead of using the default queue.
• When UDTFs are used by multiple accounts, other users, except the user who creates them, need to be authorized before using the UDTF. The authorization operations are as follows:

  Log in to the DLI console and choose Data Management > Package Management. On the displayed page, select your UDTF Jar package and click Manage Permissions in the Operation column. On the permission management page, click Grant Permission in the upper right corner and select the required permissions.

• If you use a static class or interface in a UDF, add try catch to capture exceptions. Otherwise, package conflicts may occur.

Environment Preparations

Before you start, set up the development environment.

Development Process

The process of developing a UDTF is as follows:

Figure 1 Development process

Table 2 Process description

No.	Phase	Software Portal	Description
1	Create a Maven project and configure the POM file.	IntelliJ IDEA	Write UDTF code by referring the steps in Procedure.
2	Write UDTF code.
3	Debug, compile, and pack the code into a Jar package.
4	Upload the Jar package to OBS.	OBS console	Upload the UDTF Jar file to an OBS directory.
5	Create the UDTF on DLI.	DLI console	Create a UDTF on the SQL job management page of the DLI console.
6	Verify and use the UDTF on DLI.	DLI console	Use the UDTF in your DLI job.

Procedure

1. Create a Maven project and configure the POM file. This step uses IntelliJ IDEA 2020.2 as an example.
   1. Start IntelliJ IDEA and choose File > New > Project.
      Figure 2 Creating a project
      
   2. Choose Maven, set Project SDK to 1.8, and click Next.
      Figure 3 Choosing Maven
      
   3. Set the project name, configure the storage path, and click Finish.
      Figure 4 Creating a project
      
   4. Add the following content to the pom.xml file.

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

      Figure 5 Adding configurations to the POM file
      
   5. Choose src > main and right-click the java folder. Choose New > Package to create a package and a class file.
      Figure 6 Creating a package and a class file
      

      Set the package name as you need. In this example, set Package to com.huawei.demo. Then, press Enter.

      Figure 7 Customizing a package
      

      Create a Java Class file in the package path. In this example, the Java Class file is UDTFSplit.

      Figure 8 Creating a Java class file
      
2. Write UDTF code. For sample code, see Sample Code.

   The UDTF class must inherit org.apache.hadoop.hive.ql.udf.generic.GenericUDTF to implement the initialize, process, and close methods.

   1. Call the initialize method in the UDTF. This method returns the information about the returned data rows of the UDTF, such as the number and type.
   2. Call the process method to process data. Each time forward() is called in the process method, a row is generated.
      If multiple columns are generated, you can put the values in an array and pass the array to the forward() function.

      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. Call the close method to clear methods that need to be closed.
3. Use IntelliJ IDEA to compile the code and pack it into the JAR package.
   1. Click Maven in the tool bar on the right, and click clean and compile to compile the code.
      After the compilation is successful, click package.

      Figure 9 Compiling and packaging
      

      The generated JAR package is stored in the target directory. In this example, MyUDTF-1.0-SNAPSHOT.jar is stored in D:\MyUDTF\target.

      Figure 10 Generating a JAR file
      
4. Log in to the OBS console and upload the file to the OBS path.
   

   The region of the OBS bucket to which the Jar package is uploaded must be the same as the region of the DLI queue. Cross-region operations are not allowed.

5. (Optional) Upload the file to DLI for package management.
   1. Log in to the DLI management console and choose Data Management > Package Management.
   2. On the Package Management page, click Create in the upper right corner.
   3. In the Create Package dialog, set the following parameters:
      1. Type: Select JAR.
      2. OBS Path: Specify the OBS path for storing the package.
      3. Set Group and Group Name as required for package identification and management.
   4. Click OK.
      Figure 11 Creating a package
      
6. Create the UDTF on DLI.
   1. Log in to the DLI console, choose SQL Editor. Set Engine to spark, and select the created SQL queue and database.
      Figure 12 Selecting the queue and database
      
   2. In the SQL editing area, enter the path of the JAR file to be uploaded to create a UDTF and click Execute.

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

7. Restart the original SQL queue for the added function to take effect.
   1. Log in to the DLI management console and choose Resources > Queue Management from the navigation pane. In the Operation column of the SQL queue job, click Restart.
   2. In the Restart dialog box, click OK.
8. Verify and use the UDTF on DLI.

   Use the UDTF created in 6 in the SELECT statement as follows:

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

   Figure 13 Execution result
   
9. (Optional) Delete the UDTF.

   If this function is no longer used, run the following statement to delete the function:

   Drop FUNCTION mytestsplit;

Sample Code

The complete UDTFSplit.java code is as follows:

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);
    }

}