Realtime Compute for Apache Flink:UDAF Java

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• IntelliJ IDEA installé

• Maven installé

• Environnement de développement Java configuré

Étapes

1. Téléchargez et décompressez le projet ASI_UDX_Demo sur votre machine locale. Le dossier décompressé ASI_UDX-main contient :

   • pom.xml : configuration du projet Maven, incluant les coordonnées, les dépendances et les règles de build.

   • \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java : exemple d'implémentation d'UDAF.

2. Dans IntelliJ IDEA, cliquez sur File > Open, puis sélectionnez le dossier ASI_UDX-main.

3. Ouvrez le fichier pom.xml situé dans le répertoire \ASI_UDX-main\ et configurez les dépendances. Le fichier inclut la dépendance minimale pour Flink 1.12 :

   • Si votre job n'a aucune dépendance supplémentaire, passez à l'étape suivante.

   • Si votre job nécessite des dépendances supplémentaires, ajoutez-les au fichier pom.xml.

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

   Utilisez la dernière version mineure de la version majeure d'Apache Flink correspondant à votre version Ververica Runtime (VVR). Pour les correspondances entre les versions VVR et Flink, consultez la rubrique Vue d'ensemble.

4. Ouvrez le fichier \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java et implémentez votre logique d'agrégation. L'exemple implémente une sommation cumulative :

   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. Depuis le répertoire contenant le fichier pom.xml, exécutez la commande suivante :

   mvn package -Dcheckstyle.skip

   L'UDAF est correctement empaquetée lorsque le fichier ASI_UDX-1.0-SNAPSHOT.jar apparaît dans le dossier \ASI_UDX-main\target\.

Méthode 1 : Utiliser une UDAF enregistrée (recommandée)

Enregistrez l'UDAF une fois, puis réutilisez-la dans plusieurs déploiements. Pour les étapes d'enregistrement, consultez la rubrique Gestion des UDF.

Une fois enregistrée sous le nom ASI_UDAF$MySum, appelez-la directement dans votre code SQL :

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;

Méthode 2 : Télécharger un JAR pour un déploiement spécifique

Sur la page Flink Data Studio > ETL, téléchargez le package JAR en utilisant l'option Additional dependency files située sous More configurations. Définissez ensuite un alias de fonction temporaire dans le code SQL du job.

Le JAR est limité à ce déploiement uniquement et ne peut pas être partagé avec d'autres déploiements.

Si la fonction temporaire est nommée mysum :

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;