Data Transmission Service:Use ETL para analisar pedidos em tempo real

Casos de uso

Para atender aos requisitos empresariais de processamento de dados em tempo real, o recurso de ETL oferece capacidades de extração, transformação e carregamento de dados em fluxo. Ele integra eficientemente grandes volumes de dados em tempo real, suporta operações de arrastar e soltar e adota uma abordagem de desenvolvimento low-code. Isso auxilia as empresas na análise de decisões de negócios, na aceleração da geração de relatórios e na computação de dados em tempo real. Durante a transformação digital, os casos de uso comuns para processamento de dados em tempo real incluem:

• Consolidação em tempo real de dados multirregionais ou heterogêneos: armazene dados de várias regiões ou fontes heterogêneas em um único banco de dados em tempo real para facilitar a gestão centralizada, eficiente e o suporte à tomada de decisão.

• Aceleração de relatórios: construa um sistema de relatórios em tempo real que melhore significativamente a eficiência da geração de relatórios e suporte mais cenários de análise em tempo real, atendendo às altas demandas por eficiência de relatórios durante a transformação digital.

• Computação em tempo real: limpe e processe dados de negócios em fluxo em tempo real para gerar valores de recursos e tags. Isso suporta modelos de negócios online, como criação de perfis de usuários, controle de riscos e sistemas de recomendação, além de painéis em tempo real e outros cenários de computação em fluxo.

Contexto

Este exemplo demonstra como usar o recurso de ETL de fluxo para unir dados de transações em tempo real (como ID do pedido, ID do cliente, código do product, valor da transação e horário da transação) com dados de dimensão de negócios (como código do product, preço unitário e nome do product). Os dados que atendem a condições específicas de filtro, como transações com valor superior a 3.000, são carregados em um data warehouse em tempo real. Isso permite a análise multidimensional dos dados de transação por dimensões como product e cliente. Você também pode usar ferramentas de visualização para criar painéis dinâmicos e obter insights baseados em dados.

Procedimento

Preparações

Antes de configurar a tarefa de ETL, armazene seus dados de transações em tempo real em uma tabela de fluxo e seus dados de dimensão de negócios em uma tabela de dimensão no banco de dados de origem do ApsaraDB RDS for MySQL.

Em seguida, crie uma tabela no banco de dados de destino do ApsaraDB RDS for MySQL com base nos seus requisitos de negócios.

create table test_orders(
   order_id bigint not null COMMENT 'Order ID',
     user_id bigint not null comment 'User ID',
     product_id bigint not null comment 'Product ID',
   total_price decimal(15,2) not null COMMENT 'Total order value',
   order_date TIMESTAMP  not null COMMENT 'Order date',
   PRIMARY KEY (order_id))

CREATE table product (
      product_id bigint not null comment 'Product ID',
      product_name varchar(20) comment 'Product name',
      product_price decimal(15,2) not null comment 'Product unit price')

create table test_orders(
 order_id bigint not null COMMENT 'Order ID',
 user_id bigint not null comment 'User ID',
 product_id bigint not null comment 'Product ID',
 total_price decimal(15,2) not null COMMENT 'Total order value',
 order_date TIMESTAMP  not null
 COMMENT 'Order date',
 product_id_2 bigint not null comment 'Product ID',
 product_name varchar(20) comment 'Product name',
 product_price decimal(15,2) not null comment 'Product unit price',
 PRIMARY KEY (order_id))

Etapa 1: Configurar o banco de dados de origem

1. acesse a página da lista de tarefas de ETL.

   1. Faça login no console do Data Transmission Service (DTS).

   2. No painel de navegação à esquerda, clique em ETL.

2. No canto superior esquerdo, clique em . Na caixa de diálogo Create Data Flow, insira um nome para a tarefa de ETL no campo Data Flow Name e defina Development Method como DAG.

3. clique em Confirm.

4. configure a tabela de fluxo e a tabela de dimensão.

   1. Configurar a tabela de fluxo

      1. No painel à esquerda, arraste o nó Input/Dimension Table (MySQL) para a tela à direita.

      2. clique em no nó Input/Dimension Table MySQL-1 na tela e configure o banco de dados de origem.

      3. Na aba Node Configuration, defina os seguintes parâmetros.

         Parâmetro	Descrição
         Data Source Name	Para facilitar a identificação, recomendamos usar um nome descritivo e não exclusivo.
         Region	Selecione a região do banco de dados de origem. Nota As tarefas de ETL estão disponíveis apenas nas seguintes regiões: china (Hangzhou), china (Shanghai), china (Qingdao), china (Beijing), china (Zhangjiakou), china (Shenzhen), china (Guangzhou) e china (Hong Kong).
         Instances	Selecione o nome da instância que hospeda o banco de dados de origem. Você também pode clique em em Create Instance para criar uma nova instância. Para mais informações, consulte Databases supported by DMS.
         Node Type	Selecione o tipo da tabela de origem. Neste exemplo, selecione Stream Table. Stream Table: uma tabela que muda em tempo real. Pode ser unida a uma tabela de dimensão para enriquecimento de dados. Dimension Tables: uma tabela atualizada com pouca frequência (não em tempo real). Geralmente é usada para unir com dados em tempo real e criar uma tabela ampla para análise.
         Convert Format	Ao processar dados, o ETL converte um fluxo em uma tabela dinâmica e executa consultas contínuas sobre ela. A tabela dinâmica é constantemente modificada por operações INSERT, UPDATE e DELETE, o que gera uma nova tabela dinâmica. Ao gravar dados no banco de dados de destino, essa nova tabela dinâmica é convertida novamente em um fluxo. É necessário especifique um conversion format para codificar as alterações: Upsert Stream: os dados na tabela dinâmica podem ser modificados por operações INSERT, UPDATE e DELETE. Quando convertidos em fluxo, as operações INSERT e UPDATE são codificadas como mensagens upsert, e as operações DELETE são codificadas como mensagens delete. Nota Este formato de codificação exige que a tabela dinâmica tenha uma chave exclusiva, que pode ser uma chave composta. Append-Only Stream: os dados na tabela dinâmica podem ser modificados apenas por operações INSERT. Quando convertidos em fluxo, apenas os dados inseridos são enviados.
         Select Databases and Tables	Selecione os bancos de dados e as tabelas de origem a serem transformados.

      4. Após configurar o nó, a aba Output Fields é exibida. Na coluna Column Name, selecione os campos necessários.

      5. Neste exemplo, utiliza-se uma tabela de fluxo. clique em na aba Time Attribute e defina os seguintes parâmetros.

         Parâmetro	Descrição
         Select Event Time Watermark	Selecione um campo de tempo da tabela de fluxo. Uma tabela de fluxo geralmente possui um campo de tempo que indica quando os dados foram gerados, como um carimbo de data/hora específico do negócio, por exemplo ordertime.
         Latency of Event Time Watermark	Insira a latência máxima de dados tolerável. Devido às condições de rede, os dados podem não chegar ao service de ETL na ordem em que foram gerados, causando atrasos. O ETL não pode esperar indefinidamente por dados atrasados. Portanto, é necessário especifique um período de latência para lidar com dados fora de ordem. Por exemplo, se os dados das 10:00 chegaram, mas os das 9:59 não, o ETL aguarda apenas até as 10:00 mais a latência especificada. Se os dados das 9:59 chegarem após esse período, o ETL os descartará.
         Processing Time	O tempo de processamento é a hora local do servidor quando o ETL processa os dados. É necessário defina uma coluna para armazenar esse valor. Ele é usado principalmente para cálculos de operadores; por exemplo, uma união temporal usa o tempo de processamento para buscar a versão mais recente de uma tabela comum.

      Nota

      A configuração estará concluída quando o ícone de ponto de exclamação desaparecer do lado direito do nó do banco de dados de origem.

   2. Configurar a tabela de dimensão

      1. No painel à esquerda, arraste o nó Input/Dimension Table (MySQL) para a tela à direita.

      2. clique em no nó Input/Dimension Table MySQL-2 na tela e configure o banco de dados de origem.

      3. Na aba Node Configuration, defina os seguintes parâmetros.

         Parâmetro	Descrição
         Data Source Name	Para facilitar a identificação, recomendamos usar um nome descritivo e não exclusivo.
         Region	Selecione a região do banco de dados de origem.
         Instances	Selecione o nome da instância que hospeda o banco de dados de origem. Você também pode clique em em Create Instance para criar uma nova instância. Para mais informações, consulte Databases supported by DMS.
         Node Type	Selecione o tipo da tabela de origem. Neste exemplo, selecione Dimension Tables.
         Select Databases and Tables	Selecione os bancos de dados e as tabelas de origem a serem transformados.

      4. Após configurar o nó, a aba Output Fields é exibida. Na coluna Column Name, selecione os campos necessários.

      Nota

      A configuração estará concluída quando o ícone de ponto de exclamação desaparecer do lado direito do nó do banco de dados de origem.

Etapa 2: Configurar o nó Table Join

1. No painel à esquerda, arraste o nó Table Join para uma área em branco na tela.

2. Passe o ponteiro sobre os nós configurados da tabela de fluxo e da tabela de dimensão. clique em no círculo de cada nó e arraste uma linha de conexão até o nó Table Join-1.

3. clique em no nó Table Join-1 na tela para configurar o nó.

   1. Na aba Node Settings, defina os seguintes parâmetros.

      Seção	Parâmetro	Descrição
      Conversion Name	Enter Transformation Name	Para facilitar a identificação, recomendamos usar um nome descritivo e não exclusivo.
      JOIN Settings	Left Table in JOIN Clause	Selecione a tabela a ser colocada no lado esquerdo do operador JOIN. Esta tabela serve como tabela primária. Neste exemplo, selecione a tabela de fluxo.
      	Temporal Join Time Attribute (Regular joins apply if not selected)	Selecione o atributo de tempo da tabela de fluxo usado para unir com a tabela temporal. Se você não selecionar um atributo, o DTS execute uma união comum. Neste exemplo, selecione Based on Processing Time. Nota Uma tabela temporal, também conhecida como tabela dinâmica, é uma visão parametrizada de uma tabela que registra alterações de dados ao longo do tempo. As tabelas temporais incluem tabelas versionadas que mostram versões históricas dos dados e tabelas comuns que mostram apenas a versão mais recente dos dados. Uma união temporal exige que a tabela de fluxo tenha um atributo de tempo e que a tabela da direita tenha uma chave primária. Se a tabela da direita for uma tabela de dimensão, a Join condition deve incluir a chave primária da tabela de dimensão. Based on Event Time Watermark: usa o tempo de evento da tabela de fluxo para unir com a versão correspondente na tabela versionada. Based on Processing Time: usa o tempo de processamento da tabela de fluxo para unir com a versão mais recente da tabela comum.
      	Select JOIN Operation	Selecione uma operação de união. Neste exemplo, selecione Inner Join. Inner Join: retorna registros que possuem valores correspondentes em ambas as tabelas. Left Join: retorna todos os registros da tabela da esquerda e os registros correspondentes da tabela da direita. Right Join: retorna todos os registros da tabela da direita e os registros correspondentes da tabela da esquerda.
      JOIN Condition	+ Add condition	clique em em + Add Condition para selecionar os campos da condição de união. Nota O campo à esquerda do sinal de igual (=) pertence à tabela da esquerda na união, e o campo à direita pertence à tabela da direita.

4. Após configurar a condição de união, clique em na aba Output Fields. Na coluna Column Name, selecione os campos necessários.

Etapa 3: Configurar o nó Table Record Filter

1. No painel à esquerda, arraste o nó Table Record Filter para uma área em branco na tela.

2. Passe o ponteiro sobre o nó Table Join-1 configurado. clique em no círculo e arraste uma linha de conexão do nó Table Join-1 para o nó Table Record Filter-1.

3. clique em no nó Table Record Filter-1 na tela para configurar o nó.

   1. Na seção Conversion Name, insira um nome para a transformação.

      Nota

      Para facilitar a identificação, recomendamos usar um nome descritivo e não exclusivo.

   2. Na seção WHERE Condition, configure a condição WHERE usando um dos seguintes métodos:

      • Insira diretamente a condição WHERE. Por exemplo, insira total_price > 3000.00 para filtrar registros onde o total_price da tabela unida seja maior que 3000.00.

      • clique em nas opções nas seções Input Fields ou Operator para construir a condição WHERE.

Etapa 4: Configurar o banco de dados de destino

1. No lado esquerdo da página, arraste o nó output MySQL para a área em branco da tela no lado direito da página.

2. Passe o ponteiro sobre o nó Table Record Filter-1 configurado. clique em no círculo e arraste uma linha de conexão do nó Table Record Filter-1 para o nó Output MySQL-1.

3. clique em no nó Output MySQL-1 na tela para configurar o banco de dados de destino.

   1. Na aba Node Configuration, defina os seguintes parâmetros.

      Parâmetro	Descrição
      Data Source Name	Para facilitar a identificação, recomendamos usar um nome descritivo e não exclusivo.
      Region	Selecione a região do banco de dados de destino. Nota As tarefas de ETL estão disponíveis apenas nas seguintes regiões: china (Hangzhou), china (Shanghai), china (Qingdao), china (Beijing), china (Zhangjiakou), china (Shenzhen), china (Guangzhou) e china (Hong Kong).
      Instances	Selecione o nome da instância que hospeda o banco de dados de destino. Você também pode clique em em Create Instance para criar uma nova instância. Para mais informações, consulte Databases supported by DMS.
      Table Mapping	Selecione a tabela de destino onde os dados transformados serão armazenados. Na seção Select Destination Table, clique em na tabela de destino.

4. Na coluna Column Name, selecione os parâmetros necessários.

Etapa 5: Pré-verificar e iniciar a tarefa

1. Após concluir a configuração, clique em Generate Flink SQL Validation. O ETL gera e valida o Flink SQL.

2. Após a conclusão da validação, clique em View ETL Validation Details para visualize a instrução Flink SQL gerada e os resultados da validação em uma caixa de diálogo. Após confirme que as informações estão corretas, clique em Close.

   Nota

   Se a validação falhar, solucione o problema com base nas causas exibidas nos resultados.

3. clique em Next: Save Task Settings and Precheck. A tarefa de ETL só pode ser iniciada após passar na pré-verificação. Se a pré-verificação falhar, clique em View Details ao lado do item com falha, corrija o problema com base nas informações fornecidas e execute a pré-verificação novamente.

4. Após a conclusão da pré-verificação, clique em Next: Purchase Instance na parte inferior da página.

5. Na página Purchase, selecione uma Instance Class e especifique as Compute Units (CUs). Em seguida, leia e selecione Data Transmission Service (Pay-as-you-go) Service Terms e Service Terms for Public Preview.

   Nota

   Durante a prévia pública, cada usuário pode criar e usar duas instâncias de ETL gratuitamente.

6. clique em Buy and Start para iniciar a tarefa de ETL.

Resultados da tarefa

Neste exemplo, após o início da tarefa de ETL (por exemplo, em 1º de agosto), quando um registro atualizado na tabela de transações em tempo real test_orders atende à condição de filtro (total_price > 3000.00), o registro é sincronizado com a tabela de destino test_orders_new.