Realtime Compute for Apache Flink:UDAFs Java

Pré-requisitos

Antes de começar, verifique se você tem:

• IntelliJ IDEA instalado

• Maven instalado

• Ambiente de desenvolvimento Java configurado

Etapas

1. Baixe e descompacte o ASI_UDX_Demo na sua máquina local. A pasta descompactada ASI_UDX-main contém:

   • pom.xml — configuração do projeto Maven, incluindo coordenadas, dependências e regras de build.

   • \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java — implementação de exemplo da UDAF.

2. No IntelliJ IDEA, selecione File > Open e escolha a pasta ASI_UDX-main.

3. Abra o arquivo pom.xml no diretório \ASI_UDX-main\ e configure as dependências. O arquivo já inclui a dependência mínima para o Flink 1.12:

   • Se o job não exigir dependências adicionais, pule para a próxima etapa.

   • Caso precise de mais dependências, adicione-as ao pom.xml.

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

   Use a versão secundária mais recente da versão principal do Apache Flink correspondente à sua versão do Ververica Runtime (VVR). Para consultar o mapeamento entre versões do VVR e do Flink, veja Overview.

4. Edite o arquivo \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java e implemente sua lógica de agregação. O exemplo abaixo faz uma soma cumulativa:

   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. No diretório onde está o pom.xml, execute:

   mvn package -Dcheckstyle.skip

   O empacotamento da UDAF será bem-sucedido quando o arquivo ASI_UDX-1.0-SNAPSHOT.jar aparecer em \ASI_UDX-main\target\.

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

Registre a UDAF uma única vez e reutilize-a em vários jobs. Para instruções de registro, consulte Gerencie UDFs.

Após registrá-la como ASI_UDAF$MySum, chame-a diretamente no 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: Fazer upload de um JAR para um job específico

Na página Data Studio > ETL do Flink, envie o pacote JAR pela opção Additional dependency files, em More configurations. Em seguida, defina um alias de função temporária no SQL do job.

Esse JAR fica restrito ao job em questão e não pode ser compartilhado com outros jobs.

Supondo que a função temporária se chame 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;