MaxCompute:Java UDAFs

IntelliJ IDEAまたはMaxCompute StudioでMavenを使用して、UDAFをJavaで記述できます。 UDAFコードには次の情報を含めることができます。

• Javaパッケージ: オプション。

  定義されているJavaクラスは、今後使用するためにJARファイルにパッケージ化できます。

• 基本UDAFクラス: 必須。

  必要なUDAFクラスは、com.aliyun.odps.udf.Aggregatorおよびcom.aliyun.odps.udf.annotation.Resolveです。 com.aliyun.odps.udf.annotation.Resolveクラスは、@ Resolveアノテーションに対応しています。 com.aliyun.odps.udf.UDFException: オプション。 このクラスは、Javaクラスを初期化および終了するために使用されるメソッドに対応します。 他のUDAFクラスまたは複合データ型を使用する場合は、[概要] の手順に従って必要なクラスを追加します。

• @ Resolveアノテーション: 必須です。

  アノテーションは @ Resolve(<signature>) 形式です。 signatureは、入力パラメーターのデータ型とUDAFの戻り値を定義する関数シグネチャです。 UDAFは、リフレクション機能を使用して関数シグネチャを取得できません。 関数シグネチャは、@ Resolve("smallint->varchar(10)") などの @ Resolveアノテーションを使用してのみ取得できます。 @ Resolveアノテーションの詳細については、このトピックの「 @ Resolveアノテーション」をご参照ください。

• カスタムJavaクラス: 必須。

  カスタムJavaクラスは、UDAFコードの組織単位です。 このクラスは、ビジネス要件を満たすために使用される変数とメソッドを定義します。

• カスタムJavaクラスを実装するメソッド: required.

  Java UDAFsは、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 {
      // The initialization method. 
      @Override
      public void setup(ExecutionContext ctx) throws UDFException {
      }
      // The terminate method. 
      @Override
      public void close() throws UDFException {
      }
      // Create an aggregation buffer. 
      abstract public Writable newBuffer();
      // The iterate method. 
      // The buffer is an aggregation buffer, which stores the data that is aggregated in a specific phase. The aggregated data refers to the dataset that is obtained after GROUP BY is performed for different Map tasks. One buffer is created for each row of data that is aggregated. 
      // Writable[] indicates a row of data, which specifies the passed column in the code. For example, writable[0] indicates the first column, and writable[1] indicates the second column. 
      // args specifies the parameters that are used to call a UDAF in SQL. It cannot be NULL, but the values in args can be NULL, which indicates that the 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;
  }

  最も重要なメソッドは、iterate、merge、terminateです。 メソッドは、UDAFのメインロジックを実装するために使用されます。 さらに、ユーザー定義の書き込み可能バッファを実装する必要があります。

  ユーザ定義の書き込み可能バッファは、メモリ内のオブジェクトを、ディスク内の永続的な記憶およびネットワーク伝送のためのバイトシーケンス (または他のデータ伝送プロトコル) に変換する。 MaxComputeは分散コンピューティングを使用してUDAFを処理します。 したがって、MaxComputeは、異なるデバイス間でデータを送信する前に、データをシリアル化または逆シリアル化する必要があります。

  Java UDAFを作成するときは、Javaデータ型またはJava書き込み可能データ型を使用できます。 MaxComputeプロジェクトでサポートされているデータ型、Javaデータ型、およびJava書き込み可能データ型間のマッピングの詳細については、このトピックの「データ型」をご参照ください。

サンプルコード：

// Package the defined Java classes into a file named org.alidata.odps.udaf.examples. 
package org.alidata.odps.udaf.examples;
// The base UDAF 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;
// The custom Java class. 
// The @Resolve annotation. 
@Resolve("double->double")
public class AggrAvg extends Aggregator {
// The methods that are used to implement the custom 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;
  }
}

Java UDAFを作成する前に、次の点に注意してください。

• 同じ名前で異なるロジックを持つクラスを、異なるUDAFのJARファイルにパッケージ化しないことをお勧めします。 たとえば、UDAF 1のJARファイルの名前はudaf1.jar、UDAF 2のJARファイルの名前はudaf2.jarです。 両方のJARファイルにはcom.aliyun.UserFunction.classという名前のクラスが含まれていますが、このクラスのファイル内のロジックは異なります。 同じSQL文でUDAF 1とUDAF 2が呼び出された場合、MaxComputeは2つのファイルのいずれかからcom.aliyun.UserFunction.classを読み込みます。 その結果、UDAFは期待どおりに実行できず、コンパイルが失敗する可能性があります。

• Java UDAFの入力パラメーターまたは戻り値のデータ型がオブジェクトです。 Java UDAFコードで指定するデータ型の最初の文字は、Stringなどの大文字である必要があります。

• MaxCompute SQLのNULL値は、JavaではNULLで表されます。 MaxCompute SQLでは、Javaのプリミティブデータ型は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が空 ('') の場合、入力パラメーターは使用されません。

  @ Resolveアノテーションの構文拡張機能の詳細については、「UDAFおよびUDTFの動的パラメーター」をご参照ください。

• typeは、戻り値のデータ型を指定します。 UDAFの場合、値の1列のみが返されます。 BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision、scale) のデータ型がサポートされています。 ARRAY、MAP、STRUCTなどの複雑なデータ型、およびネストされた複雑なデータ型もサポートされています。

Java UDAFを開発した後、MaxCompute SQLを使用してこのUDAFを呼び出すことができます。 Java UDAFの開発方法の詳細については、「概要」の「開発プロセス」セクションをご参照ください。 次のいずれかのメソッドを使用して、Java UDAFを呼び出すことができます。

• MaxComputeプロジェクトでUDFを使用する: この方法は、組み込み関数を使用する方法と似ています。

• プロジェクト間でUDFを使用する: プロジェクトaでプロジェクトBのUDFを使用します。次のステートメントは、例を示します。select B:udf_in_other_project(arg0, arg1) as res from table_t; プロジェクト間共有の詳細については、「パッケージに基づくプロジェクト間リソースアクセス」をご参照ください。