Realtime Compute for Apache Flink:Java UDAF'leri

Ön koşullar

Başlamadan önce şunlara sahip olduğunuzdan emin olun:

• IntelliJ IDEA yüklü

• Maven yüklü

• Java geliştirme ortamı yapılandırılmış

Adımlar

1. ASI_UDX_Demo projesini indirin ve yerel makinenizde sıkıştırılmış dosyayı açın. Sıkıştırması açılan ASI_UDX-main klasörü şunları içerir:

   • pom.xml — koordinatlar, bağımlılıklar ve derleme kuralları dahil Maven proje yapılandırması.

   • \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java — örnek UDAF uygulaması.

2. IntelliJ IDEA'da File > Open seçeneklerini seçin, ardından ASI_UDX-main klasörünü seçin.

3. \ASI_UDX-main\ dizinindeki pom.xml dosyasını açın ve bağımlılıkları yapılandırın. Dosya, Flink 1.12 için minimum bağımlılığı içerir:

   • İşinizde ek bağımlılık yoksa sonraki adıma geçin.

   • İşiniz ek bağımlılıklar gerektiriyorsa bunları pom.xml dosyasına ekleyin.

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

   Ververica Runtime (VVR) sürümünüzle eşleşen Apache Flink ana sürümü için en son küçük sürümü kullanın. VVR ile Flink sürüm eşlemeleri için bkz. Genel Bakış.

4. \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java dosyasını açın ve toplama mantığınızı uygulayın. Örnek, kümülatif toplamayı uygular:

   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;
           }
   
           /**
           * Supports retracting a message generated by an upstream operator.
           */
           public void retract(AccSum acc,long num){
               acc.sum -= num;
           }
   
           /**
           * Supports local-global two-stage aggregate optimization.
           */
           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;
                   }
               }
           }
       }
   }

5. pom.xml dosyasını içeren dizinden şu komutu çalıştırın:

   mvn package -Dcheckstyle.skip

   \ASI_UDX-main\target\ dizininde ASI_UDX-1.0-SNAPSHOT.jar dosyası göründüğünde UDAF başarıyla paketlenmiş olur.

Yöntem 1: Kayıtlı UDAF kullanma (önerilen)

UDAF'i bir kez kaydedin, ardından dağıtımlar arasında yeniden kullanın. Kayıt adımları için bkz. UDF'leri Yönetme.

ASI_UDAF$MySum olarak kaydedildikten sonra SQL kodunuzda doğrudan çağırabilirsiniz:

CREATE TEMPORARY TABLE ASI_UDAF_Source (
  a BIGINT NOT NULL
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE ASI_UDAF_Sink (
  sum  BIGINT
) WITH (
  'connector' = 'print'
);

INSERT INTO ASI_UDAF_Sink
SELECT `ASI_UDAF$MySum`(a)
FROM ASI_UDAF_Source;

Yöntem 2: Belirli bir dağıtıma JAR yükleme

Flink Data Studio > ETL sayfasında, More configurations altındaki Additional dependency files seçeneğini kullanarak JAR paketini yükleyin. Ardından işin SQL kodunda geçici bir işlev takma adı tanımlayın.

JAR yalnızca o dağıtımla sınırlıdır ve diğer dağıtımlarla paylaşılamaz.

Geçici işlev mysum olarak adlandırılırsa:

CREATE TEMPORARY TABLE ASI_UDAF_Source (
  a   BIGINT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE ASI_UDAF_Sink (
  sum  BIGINT
) WITH (
  'connector' = 'print'
);

CREATE TEMPORARY FUNCTION `mysum` AS 'ASI_UDAF.ASI_UDAF$MySum'; -- Create the temporary function mysum.

INSERT INTO ASI_UDAF_Sink
SELECT `mysum`(a)
FROM ASI_UDAF_Source;