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跨專案訪問資源。