Realtime Compute for Apache Flink:UDAFs Java

Pré-requisitos

Antes de começar, certifique-se de ter:

• IntelliJ IDEA instalado

• Maven instalado

• Ambiente de desenvolvimento Java configurado

Etapas

1. Baixe e descompacte o ASI_UDX_Demo em 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 de 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:

   • Caso seu job não exija dependências adicionais, pule para a próxima etapa.

   • Se forem necessárias 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>

   Utilize a versão secundária mais recente da versão principal do Apache Flink correspondente à sua versão do Ververica Runtime (VVR). Para verificar o mapeamento entre versões do VVR e do Flink, consulte Visão geral.

4. Abra \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java e implemente sua lógica de agregação. O exemplo abaixo realiza 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á localizado 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 deployments. Para instruções de registro, consulte Gerencie UDFs.

Após registrá-la como ASI_UDAF$MySum, chame-a diretamente no seu 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 deployment específico

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

Esse JAR fica restrito exclusivamente àquele deployment, não podendo ser compartilhado com outros.

Se a função temporária for nomeada como 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;