MaxCompute:Java UDAF

Anda dapat menulis Java UDAF di IntelliJ IDEA dengan Maven atau MaxCompute Studio. Kode tersebut harus mencakup komponen-komponen berikut:

• Java package: Opsional.

  Anda dapat mengemas kelas Java yang Anda definisikan agar lebih mudah ditemukan dan digunakan kembali.

• Kelas dan anotasi yang diperlukan: Wajib.

  Anda harus mengimpor kelas com.aliyun.odps.udf.Aggregator dan menggunakan anotasi @Resolve (com.aliyun.odps.udf.annotation.Resolve). Kelas com.aliyun.odps.udf.UDFException bersifat opsional dan dapat digunakan untuk penanganan error. Jika Anda perlu menggunakan kelas terkait UDAF lainnya atau tipe data kompleks, impor kelas yang diperlukan sebagaimana dijelaskan dalam Ikhtisar MaxCompute UDF.

• Anotasi @Resolve: Wajib.

  Formatnya adalah @Resolve(<signature>), di mana signature adalah signature fungsi yang mendefinisikan tipe data parameter input dan nilai kembali. Signature fungsi UDAF tidak dapat ditentukan melalui refleksi dan hanya dapat diperoleh dengan menggunakan anotasi @Resolve, seperti @Resolve("smallint->varchar(10)"). Untuk informasi selengkapnya tentang anotasi @Resolve, lihat Anotasi @Resolve.

• Kelas Java kustom: Wajib.

  Kelas ini merupakan unit organisasi dari kode UDAF. Kelas ini mendefinisikan variabel dan metode yang mengimplementasikan logika bisnis Anda.

• Metode untuk kelas Java: Wajib.

  Kelas Java Anda harus memperluas kelas com.aliyun.odps.udf.Aggregator dan mengimplementasikan 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 {
      // Metode inisialisasi.
      @Override
      public void setup(ExecutionContext ctx) throws UDFException {
      }
      // Metode terminasi.
      @Override
      public void close() throws UDFException {
      }
      // Membuat buffer agregasi.
      abstract public Writable newBuffer();
      // Metode iterate.
      // buffer adalah buffer agregasi yang menyimpan data antara yang telah dirangkum. Dalam tugas map, buffer ini mengagregasi data untuk satu kelompok, dan metode ini dieksekusi sekali untuk setiap baris.
      // Writable[] merepresentasikan satu baris data, yang merujuk pada kolom input dalam kode. Misalnya, writable[0] merujuk pada kolom pertama, dan writable[1] merujuk pada kolom kedua.
      // args adalah parameter yang ditentukan saat memanggil UDAF dalam SQL. Array args itu sendiri tidak boleh null, tetapi elemennya bisa null, yang menunjukkan bahwa data input yang sesuai bernilai null.
      abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;
      // Metode terminate.
      abstract public Writable terminate(Writable buffer) throws UDFException;
      // Metode merge.
      abstract public void merge(Writable buffer, Writable partial) throws UDFException;
  }

  Metode iterate, merge, dan terminate adalah tiga metode inti yang mengimplementasikan logika utama UDAF. Anda juga harus mengimplementasikan buffer writable kustom.

  Buffer writable mengonversi objek dalam memori menjadi urutan byte (atau protokol transfer data lainnya) untuk memfasilitasi persistensi ke disk dan transmisi jaringan. Karena MaxCompute menggunakan komputasi terdistribusi untuk memproses fungsi agregat, data harus diserialisasi dan dideserialisasi agar dapat ditransfer antar worker.

  Saat menulis Java UDAF, Anda dapat menggunakan tipe Java atau tipe Java Writable. Untuk informasi selengkapnya tentang pemetaan antara tipe data yang didukung MaxCompute dan tipe data Java, lihat Tipe data.

Kode berikut memberikan contoh UDAF.

// Mengemas kelas Java yang didefinisikan dalam org.alidata.odps.udaf.examples.
package org.alidata.odps.udaf.examples;
// Mengimpor kelas dasar yang diperlukan.
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;
// Mendefinisikan kelas Java kustom.
// Menentukan anotasi @Resolve.
@Resolve("double->double")
public class AggrAvg extends Aggregator {
// Mengimplementasikan metode untuk kelas 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;
  }
}

Saat menulis Java UDAF, perhatikan hal-hal berikut:

• Menyertakan kelas dengan nama yang sama tetapi logika berbeda dalam file JAR UDAF yang berbeda dapat menyebabkan hasil yang tidak terduga atau kegagalan kompilasi. Misalnya, asumsikan UDAF1 dan UDAF2 masing-masing berkorespondensi dengan file resource JAR udaf1.jar dan udaf2.jar. Jika kedua file JAR tersebut berisi kelas bernama com.aliyun.UserFunction.class tetapi dengan implementasi berbeda, MaxCompute akan memuat salah satu kelas tersebut secara tidak terduga ketika UDAF1 dan UDAF2 dipanggil dalam pernyataan SQL yang sama.

• Dalam Java UDAF, parameter input dan nilai kembali harus berupa tipe objek, seperti String dan Long, bukan tipe primitif.

• Nilai NULL dalam SQL direpresentasikan sebagai NULL Java. Tipe primitif Java tidak dapat merepresentasikan nilai NULL dalam SQL dan tidak diperbolehkan.

Format anotasi @Resolve adalah sebagai berikut.

@Resolve(<signature>)

signature adalah string yang mengidentifikasi tipe data parameter input dan nilai kembali. Saat UDAF dieksekusi, tipe parameter input dan nilai kembalinya harus sesuai dengan tipe yang ditentukan dalam signature fungsi. Selama penguraian semantik, sistem akan memeriksa penggunaan yang tidak sesuai dengan signature fungsi dan melaporkan error jika terjadi ketidakcocokan tipe. Format spesifiknya adalah sebagai berikut.

'arg_type_list -> type'

Deskripsi:

• arg_type_list: Merepresentasikan tipe data parameter input. Beberapa parameter input dapat ditentukan, dipisahkan oleh koma (,). Tipe data yang didukung adalah BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(presisi,skala), CHAR, VARCHAR, tipe data kompleks (ARRAY, MAP, STRUCT), dan tipe data kompleks bersarang.

  arg_type_list juga mendukung tanda bintang (*) atau string kosong ('').

  • Jika arg_type_list berupa tanda bintang (*), artinya fungsi menerima sejumlah parameter input apa pun.

  • Jika arg_type_list berupa string kosong (''), artinya fungsi tidak memiliki parameter input.

  Untuk informasi selengkapnya tentang sintaksis lanjutan anotasi Resolve, lihat Parameter dinamis untuk UDAF dan UDTF.

• type: Merepresentasikan tipe data nilai kembali. UDAF hanya mengembalikan satu kolom. Tipe data yang didukung mencakup BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(presisi,skala), tipe data kompleks (ARRAY, MAP, STRUCT), dan tipe data kompleks bersarang.

Setelah Anda mengembangkan Java UDAF mengikuti proses pengembangan, Anda dapat memanggilnya dalam SQL MaxCompute sebagai berikut:

• Gunakan UDF dalam proyek MaxCompute: Caranya mirip dengan penggunaan fungsi bawaan. Anda dapat menggunakan fungsi yang didefinisikan pengguna dengan cara yang sama seperti menggunakan fungsi bawaan.

• Gunakan UDF lintas proyek: Gunakan UDF dari Proyek B di Proyek A. Pernyataan berikut menunjukkan contohnya: select B:udf_in_other_project(arg0, arg1) as res from table_t;. Untuk informasi selengkapnya tentang berbagi lintas proyek, lihat Akses resource lintas proyek berbasis paket.