云原生大数据计算服务 MaxCompute：Java UDAF

您可以通过IntelliJ IDEA（Maven）或MaxCompute Studio工具使用Java语言编写UDAF代码，代码中需要包含如下信息：

• Java包（Package）：可选。

  您可以将定义的Java类打包，为后续查找和使用类提供方便。

• 继承UDAF类：必选。

  必须携带的UDAF类为com.aliyun.odps.udf.Aggregator和com.aliyun.odps.udf.annotation.Resolve（对应@Resolve注解）。com.aliyun.odps.udf.UDFException（可选，对应实现Java类初始化和结束的方法）。当您需要使用其他UDAF类或者需要用到复杂数据类型时，请根据MaxCompute UDF概述添加需要的类。

• @Resolve注解：必选。

  格式为@Resolve(<signature>)。signature为函数签名，用于定义函数的输入参数和返回值的数据类型。UDAF无法通过反射分析获取函数签名，只能通过@Resolve注解方式获取函数签名，例如@Resolve("smallint->varchar(10)")。更多@Resolve注解信息，请参见@Resolve注解。

• 自定义Java类：必选。

  UDAF代码的组织单位，定义了实现业务需求的变量及方法。

• 实现Java类的方法：必选。

  实现Java类需要继承com.aliyun.odps.udf.Aggregator类并实现如下方法。

  import com.aliyun.odps.udf.ContextFunction;
  import com.aliyun.odps.udf.ExecutionContext;
  import com.aliyun.odps.udf.UDFException;
  public abstract class Aggregator implements ContextFunction {
      //初始化方法。
      @Override
      public void setup(ExecutionContext ctx) throws UDFException {
      }
      //结束方法。
      @Override
      public void close() throws UDFException {
      }
      //创建聚合Buffer。
      abstract public Writable newBuffer();
      //iterate方法。
      //buffer为聚合buffer，是指一个阶段性的汇总数据，即在不同的Map任务中，group by后得出的数据（可理解为一个集合），每行执行一次。
      //Writable[]表示一行数据，在代码中指代传入的列。例如writable[0]表示第一列，writable[1]表示第二列。
      //args为SQL中调用UDAF时指定的参数，不能为NULL，但是args里面的元素可以为NULL，代表对应的输入数据是NULL。
      abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;
      //terminate方法。
      abstract public Writable terminate(Writable buffer) throws UDFException;
      //merge方法。
      abstract public void merge(Writable buffer, Writable partial) throws UDFException;
  }

  其中：iterate、merge和terminate是最重要的三个方法，UDAF的主要逻辑依赖于这三个方法的实现。此外，还需要您实现自定义的Writable buffer。

  Writable buffer将内存中的对象转换成字节序列（或其他数据传输协议）以便于储存到磁盘（持久化）和网络传输。因为MaxCompute使用分布式计算的方式来处理聚合函数，因此需要知道如何序列化和反序列化数据，以便于数据在不同的设备之间进行传输。

  编写Java UDAF时可以使用Java Type或Java Writable Type，MaxCompute项目支持处理的数据类型与Java数据类型的详细映射关系，请参见数据类型。

UDAF代码示例如下。

//将定义的Java类组织在org.alidata.odps.udaf.examples包中。
package org.alidata.odps.udaf.examples;
//继承UDAF类。
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;
//自定义Java类。
//@Resolve注解。
@Resolve("double->double")
public class AggrAvg extends Aggregator {
//实现Java类的方法。
  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;
  }
}

在编写Java UDAF时，您需要注意：

• 不同UDAF JAR包中不建议存在类名相同但实现逻辑不一样的类。例如UDAF1、UDAF2分别对应资源JAR包udaf1.jar、udaf2.jar，两个JAR包里都包含名称为com.aliyun.UserFunction.class的类但实现逻辑不一样，当同一条SQL语句中同时调用UDAF1和UDAF2时，MaxCompute会随机加载其中一个类，此时会导致UDAF执行结果不符合预期甚至编译失败。

• Java UDAF中输入或返回值的数据类型是对象，数据类型首字母必须大写，例如String。

• SQL中的NULL值通过Java中的NULL表示。Java Primitive Type无法表示SQL中的NULL值，不允许使用。

@Resolve注解格式如下。

@Resolve(<signature>)

signature为字符串，用于标识输入参数和返回值的数据类型。执行UDAF时，UDAF函数的输入参数和返回值类型要与函数签名指定的类型一致。查询语义解析阶段会检查不符合函数签名定义的用法，检查到类型不匹配时会报错。具体格式如下。

'arg_type_list -> type'

其中：

• arg_type_list：表示输入参数的数据类型。输入参数可以为多个，用英文逗号（,）分隔。支持的数据类型为BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision,scale)、CHAR、VARCHAR、复杂数据类型（ARRAY、MAP、STRUCT）或复杂数据类型嵌套。

  arg_type_list还支持星号（*）或为空（''）：

  • 当arg_type_list为星号（*）时，表示输入参数为任意个数。

  • 当arg_type_list为空（''）时，表示无输入参数。

• type：表示返回值的数据类型。UDAF只返回一列。支持的数据类型为：BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision,scale)、复杂数据类型（ARRAY、MAP、STRUCT）或复杂数据类型嵌套。

按照开发流程，完成Java UDAF开发后，您即可通过MaxCompute SQL调用Java UDAF。调用方法如下：

• 在归属MaxCompute项目中使用自定义函数：使用方法与内建函数类似，您可以参照内建函数的使用方法使用自定义函数。

• 跨项目使用自定义函数：即在项目A中使用项目B的自定义函数，跨项目分享语句示例：select B:udf_in_other_project(arg0, arg1) as res from table_t;。更多跨项目分享信息，请参见基于Package跨项目访问资源。