Realtime Compute for Apache Flink:UDAF en Java

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

• IntelliJ IDEA instalado

• Maven instalado

• Entorno de desarrollo de Java configurado

Pasos

1. Descargue y descomprima ASI_UDX_Demo en su computadora local. La carpeta descomprimida ASI_UDX-main contiene:

   • pom.xml — configuración del proyecto Maven, incluidas coordenadas, dependencias y reglas de compilación.

   • \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java — implementación de UDAF de ejemplo.

2. En IntelliJ IDEA, seleccione File > Open y luego seleccione la carpeta ASI_UDX-main.

3. Abra pom.xml en el directorio \ASI_UDX-main\ y configure las dependencias. El archivo incluye la dependencia mínima para Flink 1.12:

   • Si su trabajo no requiere dependencias adicionales, omita este paso y continúe con el siguiente.

   • Si su trabajo necesita dependencias adicionales, agréguelas a pom.xml.

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

   Utilice la versión menor más reciente de la versión principal de Apache Flink que corresponda a su versión de Ververica Runtime (VVR). Para consultar la correspondencia entre versiones de VVR y Flink, vea Overview.

4. Abra \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java e implemente su lógica de agregación. El ejemplo implementa una suma acumulativa:

   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. Desde el directorio que contiene pom.xml, ejecute:

   mvn package -Dcheckstyle.skip

   La UDAF se habrá empaquetado correctamente cuando aparezca ASI_UDX-1.0-SNAPSHOT.jar en \ASI_UDX-main\target\.

Método 1: Usar una UDAF registrada (recomendado)

Registre la UDAF una sola vez y reutilícela en distintos despliegues. Para consultar los pasos de registro, vea Manage UDFs.

Una vez registrada como ASI_UDAF$MySum, invoquela directamente en su 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étodo 2: Cargar un JAR en un despliegue específico

En la página Data Studio > ETL de Flink, cargue el paquete JAR mediante Additional dependency files bajo More configurations. Después, defina un alias de función temporal en el SQL del trabajo.

El JAR queda restringido a ese despliegue y no se puede compartir con otros.

Si la función temporal se llama 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;