MaxCompute：Java UDAFs

Anda dapat menggunakan Maven di IntelliJ IDEA atau MaxCompute Studio untuk menulis UDAF dalam Java. Kode UDAF dapat mencakup informasi berikut:

• Paket Java: opsional.

  Anda dapat mengemas kelas Java yang telah didefinisikan ke dalam file JAR untuk digunakan di masa mendatang.

• Kelas dasar UDAF: wajib.

  Kelas UDAF yang diperlukan adalah com.aliyun.odps.udf.Aggregator dan com.aliyun.odps.udf.annotation.Resolve. Kelas com.aliyun.odps.udf.annotation.Resolve sesuai dengan anotasi @Resolve. Kelas com.aliyun.odps.udf.UDFException bersifat opsional dan digunakan untuk metode inisialisasi serta menghentikan tugas pada kelas Java. Jika Anda ingin menggunakan kelas UDAF lainnya atau tipe data kompleks, tambahkan kelas yang diperlukan sesuai petunjuk dalam Ikhtisar.

• Anotasi @Resolve: wajib.

  Anotasi ini memiliki format @Resolve(<signature>). signature adalah tanda tangan fungsi yang mendefinisikan tipe data parameter input dan nilai balikan dari UDAF. UDAF tidak dapat memperoleh tanda tangan fungsi melalui fitur refleksi. Anda hanya dapat memperoleh tanda tangan fungsi melalui anotasi @Resolve, seperti @Resolve("smallint->varchar(10)"). Untuk informasi lebih lanjut tentang anotasi @Resolve, lihat Anotasi @Resolve dalam topik ini.

• Kelas Java kustom: wajib.

  Kelas Java kustom merupakan unit organisasi dari kode UDAF. Kelas ini mendefinisikan variabel dan metode yang digunakan untuk memenuhi kebutuhan bisnis Anda.

• Metode untuk mengimplementasikan kelas Java kustom: wajib.

  Java UDAF harus mewarisi kelas com.aliyun.odps.udf.Aggregator dan mengimplementasikan 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 terminate. 
      @Override
      public void close() throws UDFException {
      }
      // Buat buffer agregasi. 
      abstract public Writable newBuffer();
      // Metode iterate. 
      // Buffer adalah buffer agregasi, yang menyimpan data yang diagregasi pada fase tertentu. Data teragregasi merujuk pada dataset yang diperoleh setelah GROUP BY dilakukan untuk tugas Map yang berbeda. Satu buffer dibuat untuk setiap baris data yang diagregasi. 
      // Writable[] menunjukkan satu baris data, yang menentukan kolom yang dilewati dalam kode. Misalnya, writable[0] menunjukkan kolom pertama, dan writable[1] menunjukkan kolom kedua. 
      // args menentukan parameter yang digunakan untuk memanggil UDAF dalam SQL. Ini tidak boleh NULL, tetapi nilai dalam args bisa NULL, yang menunjukkan bahwa data input adalah 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 utama yang harus diimplementasikan adalah iterate, merge, dan terminate. Metode-metode ini digunakan untuk mengimplementasikan logika inti UDAF. Selain itu, Anda harus mengimplementasikan buffer writable yang ditentukan pengguna.

  Buffer writable yang ditentukan pengguna mengonversi objek dalam memori menjadi urutan byte (atau protokol transmisi data lainnya) untuk penyimpanan persisten di disk dan transmisi jaringan. MaxCompute menggunakan komputasi terdistribusi untuk memproses UDAF. Oleh karena itu, MaxCompute harus melakukan serialisasi atau deserialisasi data sebelum data dapat ditransmisikan antar perangkat yang berbeda.

  Saat menulis Java UDAF, Anda dapat menggunakan tipe data Java atau tipe data writable Java. Untuk informasi lebih lanjut tentang pemetaan antara tipe data yang didukung oleh proyek MaxCompute, tipe data Java, dan tipe data writable Java, lihat Tipe Data dalam topik ini.

Contoh kode:

// Kemas kelas Java yang didefinisikan ke dalam file bernama org.alidata.odps.udaf.examples. 
package org.alidata.odps.udaf.examples;
// Kelas dasar 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;
// Kelas Java kustom. 
// Anotasi @Resolve. 
@Resolve("double->double")
public class AggrAvg extends Aggregator {
// Metode yang digunakan untuk mengimplementasikan kelas Java kustom. 
  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;
  }
}

Sebelum menulis Java UDAF, perhatikan hal-hal berikut:

• Kami sarankan Anda tidak mengemas kelas dengan nama yang sama tetapi logika berbeda ke dalam file JAR UDAF yang berbeda. Sebagai contoh, file JAR UDAF 1 bernama udaf1.jar dan file JAR UDAF 2 bernama udaf2.jar. Kedua file JAR tersebut berisi kelas bernama com.aliyun.UserFunction.class, tetapi kelas tersebut memiliki logika yang berbeda dalam file-file tersebut. Jika UDAF 1 dan UDAF 2 dipanggil dalam pernyataan SQL yang sama, MaxCompute memuat com.aliyun.UserFunction.class dari salah satu dari dua file tersebut. Akibatnya, UDAF tidak dapat berjalan sesuai harapan dan kompilasi mungkin gagal.

• Tipe data parameter input atau nilai balikan dalam Java UDAF adalah objek. Huruf pertama dari tipe data yang Anda tentukan dalam kode Java UDAF harus huruf besar, seperti String.

• Nilai NULL dalam MaxCompute SQL direpresentasikan oleh NULL dalam Java. Tipe data primitif dalam Java tidak dapat merepresentasikan nilai NULL dalam MaxCompute SQL. Oleh karena itu, tipe data ini tidak dapat digunakan.

Format anotasi @Resolve:

@Resolve(<signature>)

Parameter signature adalah string yang menentukan tipe data parameter input dan nilai balikan. Saat menjalankan UDAF, tipe data parameter input dan nilai balikan UDAF harus konsisten dengan tipe data yang ditentukan dalam tanda tangan fungsi. Konsistensi tipe data diperiksa selama penguraian semantik. Jika tipe data tidak konsisten, kesalahan akan dikembalikan. Format tanda tangan:

'arg_type_list -> type'

• arg_type_list: menentukan tipe data parameter input. Jika beberapa parameter input digunakan, tentukan beberapa tipe data dan pisahkan dengan koma (,). Tipe data berikut didukung: BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(presisi,skala), CHAR, VARCHAR, tipe data kompleks (ARRAY, MAP, dan STRUCT), dan tipe data kompleks bersarang.

  arg_type_list dapat direpresentasikan oleh asterisk (*) atau dibiarkan kosong ('').

  • Jika arg_type_list direpresentasikan oleh asterisk (*), jumlah parameter input acak diperbolehkan.

  • Jika arg_type_list dibiarkan kosong (''), tidak ada parameter input yang digunakan.

  Untuk informasi lebih lanjut tentang ekstensi sintaksis anotasi @Resolve, lihat Parameter Dinamis UDAF dan UDTF.

• type menentukan tipe data nilai balikan. Untuk UDAF, hanya satu kolom nilai yang dikembalikan. Tipe data berikut didukung: BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, dan DECIMAL(presisi, skala). Tipe data kompleks seperti ARRAY, MAP, dan STRUCT, serta tipe data kompleks bersarang juga didukung.

Setelah mengembangkan Java UDAF, Anda dapat menggunakan MaxCompute SQL untuk memanggil UDAF ini. Untuk informasi lebih lanjut tentang cara mengembangkan Java UDAF, lihat bagian "Proses Pengembangan" dalam Ikhtisar. Anda dapat menggunakan salah satu metode berikut untuk memanggil Java UDAF:

• Gunakan UDF dalam proyek MaxCompute: Metodenya mirip dengan penggunaan fungsi bawaan.

• Gunakan UDF lintas proyek: Gunakan UDF Proyek B dalam Proyek A. Contoh pernyataan berikut menunjukkan contohnya: select B:udf_in_other_project(arg0, arg1) as res from table_t;. Untuk informasi lebih lanjut tentang berbagi lintas proyek, lihat Akses Sumber Daya Lintas Proyek Berdasarkan Paket.