MaxCompute: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 {
    // 処理開始前に 1 回呼び出されます。初期化に使用します。
    @Override
    public void setup(ExecutionContext ctx) throws UDFException {}

    // 処理終了後に 1 回呼び出されます。クリーンアップに使用します。
    @Override
    public void close() throws UDFException {}

    // 各グループごとに新しい空の集約バッファーを作成します。
    abstract public Writable newBuffer();

    // Map フェーズ：各行ごとに 1 回呼び出されます。データをバッファーに累積します。
    // args は null になりませんが、args 内の個々の値は null になる場合があります（SQL NULL 入力を示します）。
    abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;

    // Combine／Reduce フェーズ：部分バッファー（partial）をメインバッファーにマージします。
    abstract public void merge(Writable buffer, Writable partial) throws UDFException;

    // Reduce フェーズ：完全にマージされたバッファーから最終的な出力値を抽出します。
    abstract public Writable terminate(Writable buffer) throws UDFException;
}

必須メソッドと任意メソッド：

iterate() および merge() は、同一グループ内で同じバッファーインスタンスを共有します。両方のメソッドから競合せずにデータを累積できるよう、バッファーを設計してください。

集約バッファー

集約バッファーは、分散パイプラインを通過するデータの途中結果を保持します。Writable インターフェイスを実装することで、シリアル化可能になります。

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

private static class MyBuffer implements Writable {
    // バッファーのフィールドをネットワーク転送用のバイトストリームにシリアル化します。
    @Override
    public void write(DataOutput out) throws IOException {
        // 各フィールドを固定順序で書き込みます。
    }

    // バイトストリームからフィールドをデシリアライズします。
    @Override
    public void readFields(DataInput in) throws IOException {
        // write() と同じ順序で各フィールドを読み込みます。
    }
}

フィールドの読み書きは、必ず同一の順序で行う必要があります。write() と readFields() の順序が一致しないと、デバッグが困難な静かなデータ破損が発生します。

例 1：平均値計算（AggrAvg）

この例では、MaxCompute Studio を使用して、DOUBLE 型のカラムの平均値を計算する UDAF AggrAvg を開発します。

バッファーには、累積和と行数の 2 つのフィールドが格納されます。iterate() は Map フェーズで各行の入力を累積し、merge() は Combine フェーズで異なるワーカーからの部分バッファーを統合し、terminate() は Reduce フェーズで総和を総数で除算して平均値を算出します。

入力データはスライス分割され、ワーカーに分散されます。odps.stage.mapper.split.size パラメーターを設定することで、各スライスのサイズを調整できます。各ワーカーは自身のスライス内のレコード数および合計値をカウント（中間結果）、ワーカーがスライス情報を収集した後、最終出力では r.sum / r.count を平均値として計算します。

ローカルでのデバッグ

UDAF を本番環境にデプロイする前に、ローカルマシンで実行してロジックを検証します。「UDF のローカル実行によるデバッグ」の手順については、「UDF の開発」をご参照ください。

上記図のパラメータ設定は参考用です。

UDAF のパッケージ化および登録

UDAF コードを JAR ファイルにパッケージ化し、MaxCompute プロジェクトにアップロードして UDAF を作成します。「Java プログラムのパッケージ化、パッケージのアップロード、および MaxCompute UDF の作成」の手順については、「Java プログラムのパッケージ化、パッケージのアップロード、および MaxCompute UDF の作成」をご参照ください。

この例では、UDAF を user_udaf として登録します。

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

SELECT user_udaf(col0) AS c0 FROM my_table;

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

例 2：文字列連結（AggrConcat）

AggrAvg ではシンプルな数値バッファーを使用しています。この例では、より現実的なパターンを示します。すなわち、複数の行から文字列値を収集し、区切り文字で連結する処理です。

バッファーには StringBuilder と区切り文字の文字列が格納されます。StringBuilder は直接シリアル化できないため、バッファーはこれを単純な String としてシリアル化します。これは、バッファーが非プリミティブ型を保持する場合の一般的なパターンです。write() でワイヤ安全な形式にシリアル化し、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;

// 入力：STRING 型のカラム、STRING 型の区切り文字。出力：連結された STRING。
@Resolve("string,string->string")
public class AggrConcat extends Aggregator {

  // バッファー：累積されたテキストおよび値間の区切り文字。
  private static class ConcatBuffer implements Writable {
    private StringBuilder sb = new StringBuilder();
    private String separator = ",";

    @Override
    public void write(DataOutput out) throws IOException {
      // StringBuilder をネットワーク転送に耐えるように String としてシリアル化します。
      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 フェーズ：null でない各値をバッファーに追加します。
  @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 フェーズ：他のワーカーからの部分バッファーをマージします。
  @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 フェーズ：連結された結果を返します。
  @Override
  public Writable terminate(Writable buffer) throws UDFException {
    ConcatBuffer buf = (ConcatBuffer) buffer;
    ret.set(buf.sb.toString());
    return ret;
  }
}

主な設計上の判断：

文字列の連結順序は、Map フェーズのパーティション割り当てが実行エンジンによって決定されるため、ワーカー間で保証されません。順序が重要な場合は、SQL ステートメントに ORDER BY 句を含め、シングルレデューサー戦略を採用するか、連結後の出力を別途ソートしてください。

インターネットアクセス

デフォルトでは、MaxCompute は UDAF のインターネットアクセスを許可していません。インターネットアクセスを有効にするには、業務要件に基づきネットワーク接続申請フォームを提出してください。承認後、MaxCompute のテクニカルサポートチームが接続の確立を支援します。「フォームの記入方法」については、「ネットワーク接続の手順」をご参照ください。

VPC アクセス

デフォルトでは、MaxCompute は UDAF の仮想プライベートクラウド（VPC）内リソースへのアクセスを許可していません。VPC アクセスを有効にするには、MaxCompute と VPC 間のネットワーク接続を確立します。「UDF を使用した VPC 内リソースへのアクセス」をご参照ください。

テーブル読み取りの制限

UDF、UDAF、およびユーザー定義のテーブル値関数（UDTF）は、以下のテーブルタイプからデータを読み取ることができません。

• スキーマ進化が実行されたテーブル

• 複合データ型を含むテーブル

• JSON データ型を含むテーブル

• トランザクショナルテーブル