Realtime Compute for Apache Flink:Java UDAFs

Definisi

UDAF mengagregasi beberapa nilai menjadi satu nilai, membentuk pemetaan many-to-one antara input dan output. Beberapa nilai input diagregasi untuk menghasilkan satu nilai output. Untuk informasi selengkapnya, lihat User-defined Functions.

Buat UDAF

1. Unduh dan ekstrak ASI_UDX_Demo ke mesin lokal Anda.

   Setelah mengekstrak paket tersebut, folder ASI_UDX-main akan dihasilkan. Folder tersebut berisi item-item berikut:

   • pom.xml: file konfigurasi tingkat proyek yang menjelaskan koordinat Maven, dependensi, aturan yang harus diikuti pengembang, sistem manajemen cacat, organisasi, lisensi proyek, serta semua faktor terkait proyek lainnya.

   • \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java: kode Java untuk contoh UDAF.

2. Di IntelliJ IDEA, pilih File > Open, lalu pilih folder ASI_UDX-main yang telah diekstrak.

3. Klik ganda file pom.xml di direktori \ASI_UDX-main\ dan konfigurasikan informasi dalam file tersebut sesuai kebutuhan.

   Dalam contoh ini, file pom.xml berisi dependensi minimum yang diperlukan untuk mengembangkan user-defined functions untuk Flink 1.12. Jika bisnis Anda:

   • Tidak memiliki dependensi lain, Anda dapat melanjutkan ke langkah berikutnya tanpa mengonfigurasi file pom.xml.

   • Memiliki dependensi lain, tambahkan dependensi yang diperlukan ke file pom.xml.

   Contoh berikut menunjukkan dependensi minimum untuk Flink 1.12:

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>1.12.7</version>
            <scope>provided</scope>
        </dependency>
   </dependencies>

   Catatan

   Kami menyarankan Anda menggunakan versi minor terbaru dari versi utama Apache Flink yang sesuai dengan versi Ververica Runtime (VVR) pada penerapan Anda. Untuk informasi lebih lanjut tentang pemetaan versi antara VVR dan Apache Flink, lihat Ikhtisar.

4. Klik ganda \ASI_UDX-main\src\main\java\ASI_UDAF untuk membukanya, lalu konfigurasikan ASI_UDAF.java sesuai kebutuhan.

   Contoh ini menunjukkan kode untuk operasi penjumlahan kumulatif dalam file ASI_UDAF.java.

   package ASI_UDAF;
   
   import org.apache.flink.table.functions.AggregateFunction;
   
   import java.util.Iterator;
   
   public class ASI_UDAF{
       public static class AccSum{
           public long sum;
       }
   
       public static class MySum extends AggregateFunction<Long, AccSum>{
   
           @Override
           public Long getValue(AccSum acSum){
               return acSum.sum;
           }
   
           @Override
           public AccSum createAccumulator(){
               AccSum acCount= new AccSum();
               acCount.sum=0;
               return acCount;
           }
   
           public void accumulate(AccSum acc,long num){
               acc.sum += num;
           }
   
           /**
           * Mendukung penarikan kembali paket yang dihasilkan oleh operator hulu.
           */
           public void retract(AccSum acc,long num){
               acc.sum -= num;
           }
   
           /**
           * Mendukung optimasi agregasi dua tahap local-global.
           */
           public void merge(AccSum acc,Iterable<AccSum> it){
               Iterator<AccSum> iter=it.iterator();
               while(iter.hasNext()){
                   AccSum accSum=iter.next();
                   if(null!=accSum){
                       acc.sum+=accSum.sum;
                   }
               }
           }
       }
   }

   UDAF ini melakukan operasi penjumlahan kumulatif sederhana. Misalnya, jika nilai input untuk kunci grup yang sama (bidang GROUP BY) adalah 1, 2, dan 3, output dapat berupa salah satu dari dua kasus berikut:

   • Jika miniBatch tidak diaktifkan, nilai kembali adalah 1, 3, dan 6. Secara default, miniBatch tidak diaktifkan.

   • Jika miniBatch diaktifkan, nilai kembali adalah 6 dan jumlah nilai kembali antara tidak dapat ditentukan karena bervariasi berdasarkan pengaturan miniBatch dan distribusi data masukan.

   Catatan

   Untuk informasi selengkapnya tentang miniBatch, lihat Optimize Flink SQL.

5. Buka direktori tempat file pom.xml disimpan, lalu jalankan perintah berikut untuk membuat paket:

   mvn package -Dcheckstyle.skip

   Jika paket ASI_UDX-1.0-SNAPSHOT.jar muncul di direktori \ASI_UDX-main\target\, berarti UDAF telah berhasil dibuat.

Gunakan UDAF