MaxCompute:Java UDAFs

Metode wajib

Perluas com.aliyun.odps.udf.Aggregator dan implementasikan metode-metode berikut:

import com.aliyun.odps.udf.ContextFunction;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDFException;

public abstract class Aggregator implements ContextFunction {
    // Dipanggil sekali sebelum pemrosesan dimulai. Gunakan untuk inisialisasi.
    @Override
    public void setup(ExecutionContext ctx) throws UDFException {}

    // Dipanggil sekali setelah pemrosesan selesai. Gunakan untuk pembersihan.
    @Override
    public void close() throws UDFException {}

    // Membuat buffer agregasi baru yang kosong untuk setiap kelompok.
    abstract public Writable newBuffer();

    // Fase Map: dipanggil sekali per baris input. Akumulasikan data ke dalam buffer.
    // args tidak boleh null, tetapi nilai individual dalam args bisa null (menunjukkan input SQL NULL).
    abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;

    // Fase Combine/reduce: menggabungkan buffer parsial (partial) ke dalam buffer utama.
    abstract public void merge(Writable buffer, Writable partial) throws UDFException;

    // Fase Reduce: mengekstraksi nilai output akhir dari buffer yang telah digabung sepenuhnya.
    abstract public Writable terminate(Writable buffer) throws UDFException;
}

Metode wajib vs opsional:

iterate() dan merge() berbagi instance buffer yang sama dalam satu kelompok. Rancang buffer Anda agar dapat mengakumulasi data dari keduanya tanpa konflik.

Buffer agregasi

Buffer agregasi menyimpan hasil antara saat data mengalir melalui pipeline terdistribusi. Implementasikan antarmuka Writable agar dapat diserialisasi:

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

private static class MyBuffer implements Writable {
    // Menyerialisasi field buffer ke aliran byte untuk transmisi jaringan.
    @Override
    public void write(DataOutput out) throws IOException {
        // Tulis setiap field dalam urutan tetap.
    }

    // Mendeserialisasi field kembali dari aliran byte.
    @Override
    public void readFields(DataInput in) throws IOException {
        // Baca setiap field dalam urutan yang sama seperti write().
    }
}

Baca dan tulis field dalam urutan yang sama. Ketidaksesuaian antara write() dan readFields() menyebabkan korupsi data diam-diam yang sulit didebug.

Contoh 1: rata-rata (AggrAvg)

Contoh ini mengembangkan AggrAvg, sebuah UDAF yang menghitung rata-rata kolom DOUBLE, menggunakan MaxCompute Studio.

Buffer menyimpan dua field: jumlah berjalan dan jumlah baris. iterate() mengakumulasi setiap baris input selama fase map, merge() menggabungkan buffer parsial dari worker berbeda selama fase combine, dan terminate() membagi jumlah total dengan jumlah baris selama fase reduce.

Data input diiris dan didistribusikan ke worker. Anda dapat mengonfigurasi parameter odps.stage.mapper.split.size untuk menyesuaikan ukuran setiap irisan. Setiap worker menghitung jumlah catatan dan total dalam irisannya (hasil antara), worker mengumpulkan informasi irisan, dan output akhir menghitung r.sum / r.count sebagai rata-rata.

Debug secara lokal

Jalankan UDAF di mesin lokal Anda untuk memverifikasi logika sebelum deployment. Untuk langkah debugging, lihat bagian "Perform a local run to debug the UDF" dalam Develop a UDF.

Pengaturan parameter pada gambar di atas hanya untuk referensi.

Paket dan daftarkan UDAF

Paket kode UDAF ke dalam file JAR, unggah ke proyek MaxCompute Anda, dan buat UDAF tersebut. Untuk langkah pengemasan, lihat bagian "Procedure" dalam Package a Java program, upload the package, and create a MaxCompute UDF.

Contoh ini mendaftarkan UDAF sebagai user_udaf.

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

SELECT user_udaf(col0) AS c0 FROM my_table;

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

Contoh 2: penggabungan string (AggrConcat)

AggrAvg menggunakan buffer numerik sederhana. Contoh ini menunjukkan pola yang lebih realistis: mengumpulkan nilai string dari beberapa baris dan menggabungkannya dengan pemisah.

Buffer menyimpan StringBuilder dan string pemisah. Karena StringBuilder tidak dapat diserialisasi secara langsung, buffer menyerialisasikannya sebagai String biasa. Ini adalah pola yang harus diikuti ketika buffer Anda menyimpan tipe non-primitif—serialisasi ke bentuk yang aman untuk transmisi dalam write() dan rekonstruksi bentuk dalam memori dalam 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: kolom STRING, pemisah STRING. Output: STRING yang digabungkan.
@Resolve("string,string->string")
public class AggrConcat extends Aggregator {

  // Buffer: teks yang terakumulasi dan pemisah yang digunakan antar nilai.
  private static class ConcatBuffer implements Writable {
    private StringBuilder sb = new StringBuilder();
    private String separator = ",";

    @Override
    public void write(DataOutput out) throws IOException {
      // Serialisasi StringBuilder sebagai String agar bertahan dalam transfer jaringan.
      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();
  }

  // Fase Map: tambahkan setiap nilai non-null ke dalam 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());
    }
  }

  // Fase Combine: gabungkan buffer parsial dari worker lain.
  @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);
    }
  }

  // Fase Reduce: kembalikan hasil penggabungan.
  @Override
  public Writable terminate(Writable buffer) throws UDFException {
    ConcatBuffer buf = (ConcatBuffer) buffer;
    ret.set(buf.sb.toString());
    return ret;
  }
}

Keputusan desain utama:

Urutan penggabungan string tidak dijamin antar worker karena penugasan partisi fase map ditentukan oleh mesin eksekusi, bukan oleh urutan baris. Jika urutan penting, gunakan klausa ORDER BY dalam pernyataan SQL Anda dan strategi single-reducer, atau urutkan output penggabungan setelahnya.

Akses Internet

Secara default, MaxCompute tidak mengizinkan UDAF mengakses Internet. Untuk mengaktifkan akses Internet, ajukan formulir permohonan koneksi jaringan sesuai kebutuhan bisnis Anda. Setelah disetujui, tim dukungan teknis MaxCompute akan membantu Anda membuat koneksi tersebut. Untuk petunjuk pengisian formulir, lihat Proses koneksi jaringan.

Akses VPC

Secara default, MaxCompute tidak mengizinkan UDAF mengakses resource di virtual private cloud (VPC). Untuk mengaktifkan akses VPC, buat koneksi jaringan antara MaxCompute dan VPC. Lihat Gunakan UDF untuk mengakses resource di VPC.

Batasan pembacaan tabel

UDF, UDAF, dan fungsi bernilai tabel yang didefinisikan pengguna (UDTF) tidak dapat membaca data dari jenis tabel berikut:

• Tabel yang telah menjalani evolusi skema

• Tabel yang berisi tipe data kompleks

• Tabel yang berisi tipe data JSON

• Tabel transaksional