Realtime Compute for Apache Flink:Classificar e agregar dados com uma UDAF

Dados de amostra

A tabela electric_info contém dados de terminais de rede elétrica residencial, incluindo o ID do evento (event_id), o ID do usuário (user_id), o horário do evento (event_time) e o status do terminal (status). Você agregará os valores de status para cada usuário e os classificará por event_time.

• electric_info

  event_id	user_id	event_time	status
  1	1222	2023-06-30 11:14:00	LD
  2	1333	2023-06-30 11:12:00	LD
  3	1222	2023-06-30 11:11:00	TD
  4	1333	2023-06-30 11:12:00	LD
  5	1222	2023-06-30 11:15:00	TD
  6	1333	2023-06-30 11:18:00	LD
  7	1222	2023-06-30 11:19:00	TD
  8	1333	2023-06-30 11:10:00	TD
  9	1555	2023-06-30 11:16:00	TD
  10	1555	2023-06-30 11:17:00	LD

• Resultado esperado

  user_id	status
  1222	TD,LD,TD,TD
  1333	TD,LD,LD,LD
  1555	TD,LD

Etapa 1: Preparar a fonte de dados

Este exemplo utiliza o ApsaraDB RDS como fonte de dados.

1. Criar uma instância do ApsaraDB RDS for MySQL.

   Nota

   A instância do ApsaraDB RDS for MySQL deve estar na mesma VPC do workspace do Realtime Compute for Apache Flink. Se estiverem em VPCs diferentes, consulte Conectividade de rede.

2. Criar um banco de dados e uma conta.

   Crie um banco de dados chamado electric e crie uma conta com privilégios elevados ou uma conta regular com permissões de leitura e gravação no banco de dados electric.

3. Faça login na instância do ApsaraDB RDS for MySQL usando o Data Management (DMS), crie as tabelas electric_info e electric_info_SortListAgg no banco de dados electric e insira os dados.

   CREATE TABLE `electric_info` (
     event_id bigint NOT NULL PRIMARY KEY COMMENT 'Event ID',
     user_id bigint NOT NULL COMMENT 'User ID',
     event_time timestamp NOT NULL COMMENT 'Event time',
     status varchar(10) NOT NULL COMMENT 'User terminal status'
   );
   CREATE TABLE `electric_info_SortListAgg` (
     user_id bigint NOT NULL PRIMARY KEY COMMENT 'User ID',
     status_sort varchar(50) NULL COMMENT 'User terminal status sorted in ascending order by event time'
   );
   -- Prepare data
   INSERT INTO electric_info VALUES
   (1,1222,'2023-06-30 11:14','LD'),
   (2,1333,'2023-06-30 11:12','LD'),
   (3,1222,'2023-06-30 11:11','TD'),
   (4,1333,'2023-06-30 11:12','LD'),
   (5,1222,'2023-06-30 11:15','TD'),
   (6,1333,'2023-06-30 11:18','LD'),
   (7,1222,'2023-06-30 11:19','TD'),
   (8,1333,'2023-06-30 11:10','TD'),
   (9,1555,'2023-06-30 11:16','TD'),
   (10,1555,'2023-06-30 11:17','LD');

Etapa 2: Registrar a UDAF

1. Baixe o pacote ASI_UDX-1.0-SNAPSHOT.jar.

   O arquivo pom.xml está configurado com as dependências mínimas necessárias para esta função personalizada na versão 1.17.1 do Flink. Para mais informações sobre funções personalizadas, consulte Funções personalizadas.

2. O código de exemplo do ASI_UDAF mescla várias linhas em uma única linha e classifica os dados por uma coluna especificada. Você pode modificar o código conforme a necessidade do seu negócio.

   package ASI_UDAF;
   import org.apache.commons.lang3.StringUtils;
   import org.apache.flink.table.functions.AggregateFunction;
   import java.util.ArrayList;
   import java.util.Comparator;
   import java.util.Iterator;
   import java.util.List;
   public class ASI_UDAF{
   	/**Accumulator class*/
   	public static class AcList {
   		public  List<String> list;
   	}
   	/**Aggregate function class*/
   	public static class SortListAgg extends AggregateFunction<String,AcList> {
   		public String getValue(AcList asc) {
   			/**Sort the data in the list based on a specific rule*/
   			asc.list.sort(new Comparator<String>() {
   				@Override
   				public int compare(String o1, String o2) {
   					return Integer.parseInt(o1.split("#")[1]) - Integer.parseInt(o2.split("#")[1]);
   				}
   			});
   			/**Traverse the sorted list, extract the required fields, and join them into a string*/
   			List<String> ret = new ArrayList<String>();
   			Iterator<String> strlist = asc.list.iterator();
   			while (strlist.hasNext()) {
   				ret.add(strlist.next().split("#")[0]);
   			}
   			String str = StringUtils.join(ret, ',');
   			return str;
   		}
   		/**Method to create an accumulator*/
   		public AcList createAccumulator() {
   			AcList ac = new AcList();
   			List<String> list = new ArrayList<String>();
   			ac.list = list;
   			return ac;
   		}
   		/**Accumulation method: Add the input data to the accumulator*/
   		public void accumulate(AcList acc, String tuple1) {
   			acc.list.add(tuple1);
   		}
   		/**Retraction method*/
   		public void retract(AcList acc, String num) {
   		}
   	}
   }

3. Registre a UDAF.

   Registrar uma UDAF permite reutilizar o código em outros jobs. Para UDAFs em Java, você também pode fazer upload do JAR como arquivo de dependência. Para mais informações, consulte Funções de agregação definidas pelo usuário (UDAFs).

   1. Faça login no console do Realtime Compute for Apache Flink.

   2. Localize o workspace desejado e clique em Console na coluna Actions.

   3. No painel de navegação à esquerda, escolha Development > ETL.

   4. Na aba UDFs, clique em Register UDF Artifact.

4. Na seção Click to select, faça upload do arquivo JAR da Etapa 1 e clique em OK.

   A caixa de diálogo oferece dois métodos de registro: Upload File e External URL. Você também deve especificar um UDF Name e, opcionalmente, fazer upload de um dependency file.

   Nota

   • O arquivo JAR da sua UDF é enviado para o diretório sql-artifacts do OSS Bucket associado ao workspace.

   • O console do Realtime Compute for Apache Flink analisa o arquivo JAR da UDF e detecta classes que utilizam as interfaces UDF, UDAF e UDTF do Flink. Os nomes das classes são extraídos automaticamente e preenchidos no campo Function Name.

5. Na caixa de diálogo Manage Functions, clique em Create Functions.

   A UDF registrada aparece na lista de UDFs no lado esquerdo da página do editor SQL.

Etapa 3: Criar um job Flink

1. Na página Development > ETL, clique em New.

2. Clique em Blank Stream Draft.

3. Clique em Next.

4. Na caixa de diálogo New Draft, configure as definições do job.

   Parâmetro	Descrição
   Name	Um nome exclusivo para o job. Nota O nome do job deve ser exclusivo no projeto atual.
   Location	O local de armazenamento do job. Você também pode clicar no ícone ao lado de uma pasta existente para criar uma subpasta.
   Engine version	A versão do engine Flink para o job. Deve corresponder à versão especificada no arquivo pom.xml. Para detalhes sobre versões do engine, mapeamento de versões e informações de ciclo de vida, consulte Versões do engine.

5. Escreva as instruções DDL e DML.

   -- Create the temporary table electric_info.
   CREATE TEMPORARY TABLE electric_info (
     event_id bigint not null,
     `user_id` bigint not null,
     event_time timestamp(6) not null,
     status string not null,
     primary key(event_id) not enforced
   ) WITH (
     'connector' = 'mysql',
     'hostname' = 'rm-bp1s1xgll21******.mysql.rds.aliyuncs.com',
     'port' = '3306',
     'username' = 'your_username',
     'password' = '${secret_values.mysql_pw}',
     'database-name' = 'electric',
     'table-name' = 'electric_info'
   );
   CREATE TEMPORARY TABLE electric_info_sortlistagg (
     `user_id` bigint not null,
     status_sort varchar(50) not null,
     primary key(user_id) not enforced
   ) WITH (
     'connector' = 'mysql',
     'hostname' = 'rm-bp1s1xgll21******.mysql.rds.aliyuncs.com',
     'port' = '3306',
     'username' = 'your_username',
     'password' = '${secret_values.mysql_pw}',
     'database-name' = 'electric',
     'table-name' = 'electric_info_sortlistagg'
   );
   -- Aggregate data from the electric_info table and insert it into the electric_info_sortlistagg table.
   -- Pass a concatenated string of status and event_time as a parameter to the registered custom function ASI_UDAF$SortListAgg.
   INSERT INTO electric_info_sortlistagg
   SELECT `user_id`, `ASI_UDAF$SortListAgg`(CONCAT(status,'#',CAST(UNIX_TIMESTAMP(event_time) as STRING)))
   FROM electric_info GROUP BY user_id;

   A tabela a seguir descreve os parâmetros. Modifique-os conforme necessário. Para mais informações sobre os parâmetros do conector MySQL, consulte Conector MySQL.

   Parâmetro	Descrição	Observações
   connector	O tipo de conector.	Neste exemplo, o valor é fixo como mysql.
   hostname	O endereço IP ou nome do host do banco de dados MySQL.	Este exemplo utiliza o endpoint interno da instância do ApsaraDB RDS for MySQL.
   username	O nome de usuário do serviço de banco de dados MySQL.	Nenhuma.
   password	A senha do serviço de banco de dados MySQL.	Este exemplo utiliza uma variável chamada mysql_pw para a senha, a fim de evitar riscos de segurança. Para mais informações, consulte Variáveis.
   database-name	O nome do banco de dados MySQL.	Este exemplo utiliza electric, o banco de dados criado na Etapa 1: Preparar a fonte de dados.
   table-name	O nome da tabela MySQL.	Neste exemplo, defina como electric_info ou electric_info_sortlistagg.
   port	O número da porta do serviço de banco de dados MySQL.	Nenhuma.

6. (Opcional) No canto superior direito, clique em Validate e Debug. Para mais informações sobre esses recursos, consulte Visão geral do desenvolvimento de jobs.

7. Clique em Deploy e, em seguida, clique em Confirm.

8. Na página O&M > Deployments, localize o job desejado, clique em Start na coluna Actions e selecione Initial Mode.

Etapa 4: Consultar o resultado

No ApsaraDB RDS, execute a seguinte instrução para visualizar os resultados agregados e classificados.

SELECT * FROM `electric_info_sortlistagg`;

A saída confirma que os status de cada usuário foram agregados e classificados corretamente: user_id=1222 corresponde a status_sort=TD,LD,TD,TD; user_id=1333 corresponde a status_sort=TD,LD,LD,LD; e user_id=1555 corresponde a status_sort=TD,LD.

Referências

• Para informações sobre as funções integradas suportadas pelo Flink, consulte Funções integradas.

• Para informações sobre como implantar e iniciar jobs, consulte Implantar um job e Iniciar um job.

• Para modificar os parâmetros de execução do job, consulte Configurar a implantação do job. Alguns parâmetros podem ser atualizados dinamicamente para reduzir o tempo de inatividade causado pela interrupção e reinício dos jobs. Para mais informações, consulte Dimensionamento dinâmico e atualização de parâmetros.

• Para informações sobre como usar funções personalizadas em Python em jobs SQL, consulte Funções personalizadas.