Java UDAF

You can write a Java UDAF in IntelliJ IDEA with Maven or MaxCompute Studio. The code must include the following components:

• Java package: Optional.

  You can package the Java classes that you define to make them easier to find and reuse.

• Required classes and annotations: Required.

  You must import the com.aliyun.odps.udf.Aggregator class and use the @Resolve annotation (com.aliyun.odps.udf.annotation.Resolve). The com.aliyun.odps.udf.UDFException class is optional and can be used for error handling. If you need to use other UDAF-related classes or complex data types, import the required classes as described in Overview of MaxCompute UDFs.

• @Resolve annotation: Required.

  The format is @Resolve(<signature>), where signature is the function signature that defines the data types of the input parameters and the return value. The function signature of a UDAF cannot be determined through reflection and can only be obtained by using the @Resolve annotation, such as @Resolve("smallint->varchar(10)"). For more information about the @Resolve annotation, see @Resolve annotation.

• Custom Java class: Required.

  This class is the organizational unit of the UDAF code. It defines the variables and methods that implement your business logic.

• Methods for the Java class: Required.

  Your Java class must extend the com.aliyun.odps.udf.Aggregator class and implement the following methods.

  import com.aliyun.odps.udf.ContextFunction;
  import com.aliyun.odps.udf.ExecutionContext;
  import com.aliyun.odps.udf.UDFException;
  public abstract class Aggregator implements ContextFunction {
      // The initialization method.
      @Override
      public void setup(ExecutionContext ctx) throws UDFException {
      }
      // The termination method.
      @Override
      public void close() throws UDFException {
      }
      // Creates an aggregation buffer.
      abstract public Writable newBuffer();
      // The iterate method.
      // buffer is an aggregation buffer that holds intermediate, summarized data. In map tasks, it aggregates data for a group, and this method is executed once for each row.
      // Writable[] represents a row of data, which refers to the input columns in the code. For example, writable[0] refers to the first column, and writable[1] refers to the second column.
      // args are the parameters specified when calling the UDAF in SQL. The args array itself cannot be null, but its elements can be null, which indicates that the corresponding input data is null.
      abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;
      // The terminate method.
      abstract public Writable terminate(Writable buffer) throws UDFException;
      // The merge method.
      abstract public void merge(Writable buffer, Writable partial) throws UDFException;
  }

  The iterate, merge, and terminate methods are the three core methods that implement the main logic of a UDAF. You must also implement a custom writable buffer.

  A writable buffer converts in-memory objects into a byte sequence (or another data transfer protocol) to facilitate disk persistence and network transmission. Because MaxCompute uses distributed computing to process aggregate functions, it must serialize and deserialize data to transfer it between workers.

  When you write a Java UDAF, you can use Java types or Java Writable types. For more information about the mappings between data types supported by MaxCompute and Java data types, see Data types.

The following code provides an example of a UDAF.

// Package the defined Java class in org.alidata.odps.udaf.examples.
package org.alidata.odps.udaf.examples;
// Import the required base classes.
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import com.aliyun.odps.io.DoubleWritable;
import com.aliyun.odps.io.Writable;
import com.aliyun.odps.udf.Aggregator;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.annotation.Resolve;
// Define the custom Java class.
// Specify the @Resolve annotation.
@Resolve("double->double")
public class AggrAvg extends Aggregator {
// Implement the methods for the Java class.
  private static class AvgBuffer implements Writable {
    private double sum = 0;
    private long count = 0;
    @Override
    public void write(DataOutput out) throws IOException {
      out.writeDouble(sum);
      out.writeLong(count);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
      sum = in.readDouble();
      count = in.readLong();
    }
  }
  private DoubleWritable ret = new DoubleWritable();
  @Override
  public Writable newBuffer() {
    return new AvgBuffer();
  }
  @Override
  public void iterate(Writable buffer, Writable[] args) throws UDFException {
    DoubleWritable arg = (DoubleWritable) args[0];
    AvgBuffer buf = (AvgBuffer) buffer;
    if (arg != null) {
      buf.count += 1;
      buf.sum += arg.get();
    }
  }
  @Override
  public Writable terminate(Writable buffer) throws UDFException {
    AvgBuffer buf = (AvgBuffer) buffer;
    if (buf.count == 0) {
      ret.set(0);
    } else {
      ret.set(buf.sum / buf.count);
    }
    return ret;
  }
  @Override
  public void merge(Writable buffer, Writable partial) throws UDFException {
    AvgBuffer buf = (AvgBuffer) buffer;
    AvgBuffer p = (AvgBuffer) partial;
    buf.sum += p.sum;
    buf.count += p.count;
  }
}

When you write a Java UDAF, take note of the following points:

• Including classes with the same name but different logic in the JAR files of different UDAFs can lead to unexpected results or compilation failures. For example, assume UDAF1 and UDAF2 correspond to the resource JAR files udaf1.jar and udaf2.jar, respectively. If both JAR files contain a class named com.aliyun.UserFunction.class but with different implementations, MaxCompute loads one of the classes unpredictably when both UDAF1 and UDAF2 are called in the same SQL statement.

• In a Java UDAF, input parameters and the return value must be object types, such as String and Long, not primitive types.

• NULL values in SQL are represented by Java NULL. Java primitive types cannot represent NULL values in SQL and are not allowed.

The format of the @Resolve annotation is as follows.

@Resolve(<signature>)

The signature is a string that identifies the data types of the input parameters and return value. When a UDAF is executed, the types of its input parameters and return value must match the types specified in the function signature. During semantic parsing, the system checks for usages that do not conform to the function signature and reports an error if a type mismatch occurs. The specific format is as follows.

'arg_type_list -> type'

Description:

• arg_type_list: Represents the data types of the input parameters. Multiple input parameters can be specified, separated by commas (,). Supported data types are BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(precision,scale), CHAR, VARCHAR, complex data types (ARRAY, MAP, STRUCT), and nested complex data types.

  arg_type_list also supports an asterisk (*) or an empty string ('').

  • If arg_type_list is an asterisk (*), it indicates that the function accepts any number of input parameters.

  • If arg_type_list is an empty string (''), it indicates that the function has no input parameters.

  For more information about the extended syntax of the Resolve annotation, see Dynamic parameters for UDAFs and UDTFs.

• type: Represents the data type of the return value. A UDAF returns only one column. Supported data types include BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(precision,scale), complex data types (ARRAY, MAP, STRUCT), and nested complex data types.

After you develop the Java UDAF by following the development process, you can call it in MaxCompute SQL as follows:

• Use a user-defined function in the current MaxCompute project: The usage is similar to that of a built-in function.

• Use a user-defined function across projects: To use a user-defined function from Project B in Project A, use a cross-project sharing statement. For example: select B:udf_in_other_project(arg0, arg1) as res from table_t;. For more information about cross-project sharing, see Access resources across projects by using packages.