MaxCompute:Java UDAFs

Required methods

Extend com.aliyun.odps.udf.Aggregator 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 {
    // Called once before processing starts. Use for initialization.
    @Override
    public void setup(ExecutionContext ctx) throws UDFException {}

    // Called once after processing ends. Use for cleanup.
    @Override
    public void close() throws UDFException {}

    // Creates a new, empty aggregation buffer for each group.
    abstract public Writable newBuffer();

    // Map phase: called once per input row. Accumulate data into buffer.
    // args cannot be null, but individual values in args can be null (indicating SQL NULL input).
    abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;

    // Combine/reduce phase: merges a partial buffer (partial) into the main buffer.
    abstract public void merge(Writable buffer, Writable partial) throws UDFException;

    // Reduce phase: extracts the final output value from the fully merged buffer.
    abstract public Writable terminate(Writable buffer) throws UDFException;
}

Mandatory vs. optional methods:

iterate() and merge() share the same buffer instance within a group. Design your buffer so it can accumulate data from both without conflict.

Aggregation buffer

The aggregation buffer holds intermediate results as data flows through the distributed pipeline. Implement the Writable interface to make it serializable:

import com.aliyun.odps.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

private static class MyBuffer implements Writable {
    // Serializes buffer fields to a byte stream for network transmission.
    @Override
    public void write(DataOutput out) throws IOException {
        // Write each field in a fixed order.
    }

    // Deserializes fields back from the byte stream.
    @Override
    public void readFields(DataInput in) throws IOException {
        // Read each field in the same order as write().
    }
}

Read and write fields in the same order. A mismatch between write() and readFields() causes silent data corruption that is difficult to debug.

Example 1: average (AggrAvg)

This example develops AggrAvg, a UDAF that calculates the average of a DOUBLE column, using MaxCompute Studio.

The buffer stores two fields: a running sum and a row count. iterate() accumulates each input row during the map phase, merge() combines partial buffers from different workers during the combine phase, and terminate() divides the total sum by the total count during the reduce phase.

The input data is sliced and distributed to workers. You can configure the odps.stage.mapper.split.size parameter to adjust the size of each slice. Each worker counts the number of records and total in its slice (intermediate result), workers collect slice information, and the final output computes r.sum / r.count as the average.

Debug locally

Run the UDAF on your local machine to verify the logic before deploying. For debugging steps, see the "Perform a local run to debug the UDF" section in Develop a UDF.

The parameter settings in the preceding figure are for reference only.

Package and register the UDAF

Package the UDAF code into a JAR file, upload it to your MaxCompute project, and create the UDAF. For packaging steps, see the "Procedure" section in Package a Java program, upload the package, and create a MaxCompute UDF.

This example registers the UDAF as user_udaf.

+------------+------------+
| col0       | col1       |
+------------+------------+
| 1.2        | 2.0        |
| 1.6        | 2.1        |
+------------+------------+

SELECT user_udaf(col0) AS c0 FROM my_table;

+----+
| c0 |
+----+
| 1.4|
+----+

Example 2: string concatenation (AggrConcat)

AggrAvg uses a simple numeric buffer. This example shows a more realistic pattern: collecting string values from multiple rows and concatenating them with a separator.

The buffer stores a StringBuilder and a separator string. Because StringBuilder is not directly serializable, the buffer serializes it as a plain String. This is the pattern to follow whenever your buffer holds a non-primitive type — serialize to a wire-safe form in write() and reconstruct the in-memory form in readFields().

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import com.aliyun.odps.io.Text;
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;

// Input: STRING column, STRING separator. Output: concatenated STRING.
@Resolve("string,string->string")
public class AggrConcat extends Aggregator {

  // Buffer: accumulated text and the separator to use between values.
  private static class ConcatBuffer implements Writable {
    private StringBuilder sb = new StringBuilder();
    private String separator = ",";

    @Override
    public void write(DataOutput out) throws IOException {
      // Serialize StringBuilder as a String so it survives network transfer.
      out.writeUTF(sb.toString());
      out.writeUTF(separator);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
      sb = new StringBuilder(in.readUTF());
      separator = in.readUTF();
    }
  }

  private Text ret = new Text();

  @Override
  public Writable newBuffer() {
    return new ConcatBuffer();
  }

  // Map phase: append each non-null value to the buffer.
  @Override
  public void iterate(Writable buffer, Writable[] args) throws UDFException {
    Text value = (Text) args[0];
    Text sep   = (Text) args[1];
    ConcatBuffer buf = (ConcatBuffer) buffer;
    if (value != null) {
      if (sep != null) {
        buf.separator = sep.toString();
      }
      if (buf.sb.length() > 0) {
        buf.sb.append(buf.separator);
      }
      buf.sb.append(value.toString());
    }
  }

  // Combine phase: merge a partial buffer from another worker.
  @Override
  public void merge(Writable buffer, Writable partial) throws UDFException {
    ConcatBuffer buf = (ConcatBuffer) buffer;
    ConcatBuffer p   = (ConcatBuffer) partial;
    if (p.sb.length() > 0) {
      if (buf.sb.length() > 0) {
        buf.sb.append(buf.separator);
      }
      buf.sb.append(p.sb);
    }
  }

  // Reduce phase: return the concatenated result.
  @Override
  public Writable terminate(Writable buffer) throws UDFException {
    ConcatBuffer buf = (ConcatBuffer) buffer;
    ret.set(buf.sb.toString());
    return ret;
  }
}

Key design decisions:

String concatenation order is not guaranteed across workers because the map-phase partition assignment is determined by the execution engine, not by row order. If order matters, use an ORDER BY clause in your SQL statement and a single-reducer strategy, or sort the concatenated output afterward.

Internet access

By default, MaxCompute does not allow UDAFs to access the Internet. To enable Internet access, submit a network connection application form based on your business requirements. After approval, the MaxCompute technical support team will help you establish the connection. For instructions on filling in the form, see Network connection process.

VPC access

By default, MaxCompute does not allow UDAFs to access resources in a virtual private cloud (VPC). To enable VPC access, establish a network connection between MaxCompute and the VPC. See Use UDFs to access resources in VPCs.

Table read restrictions

UDFs, UDAFs, and user-defined table-valued functions (UDTFs) cannot read data from the following table types:

• Tables on which schema evolution has been performed

• Tables containing complex data types

• Tables containing JSON data types

• Transactional tables