Todos os produtos
Search

Este produto

    Realtime Compute for Apache Flink:Ingestão de banco de dados em tempo real

    Central de documentação

    Realtime Compute for Apache Flink:Ingestão de banco de dados em tempo real

    Última atualização: Aug 24, 2026

    O Realtime Compute for Apache Flink simplifica a ingestão de dados em tempo real ao gerenciar automaticamente a transição da sincronização completa para a incremental, a descoberta de metadados, a evolução de schema e a sincronização de todo o banco de dados. Este tópico mostra como criar rapidamente um job de ingestão que transmite dados do ApsaraDB RDS for MySQL para o Hologres.

    Contexto

    A seguir, visualize como esses bancos de dados e tabelas aparecem no console DMS.

    A instância RDS contém vários bancos de dados (tpc_ds, tpc_ds_large, user_db1, user_db2, user_db3, __recycle_bin__). O banco de dados user_db3 contém três tabelas: user03, user06 e user09. A tabela user03 possui duas colunas: id (int(11)) e name (varchar(255)).

    Siga estas etapas para desenvolver um job de ingestão que sincronize todas essas tabelas com o Hologres e mescle as tabelas de usuários fragmentadas em uma única tabela:

    Este tópico utiliza a ingestão de dados do Flink CDC para sincronizar todo o banco de dados e mesclar tabelas fragmentadas. Isso permite concluir a sincronização de dados completa e incremental, bem como a sincronização de alterações de schema em tempo real, com um único job.

    Pré-requisitos

    Preparar dados de teste do MySQL e um banco de dados Hologres

    1. Clique em tpc_ds.sql, user_db1.sql, user_db2.sql e user_db3.sql para baixar os arquivos de dados de teste para sua máquina local.

    2. No console DMS, prepare os dados de teste na sua instância ApsaraDB RDS for MySQL.

      1. Faça login na sua instância ApsaraDB RDS for MySQL usando o DMS.

      2. Na janela SQL Console, insira os seguintes comandos e clique em Execute.

        Os comandos abaixo criam quatro bancos de dados: tpc_ds, user_db1, user_db2 e user_db3.

        CREATE DATABASE tpc_ds;
CREATE DATABASE user_db1;
CREATE DATABASE user_db2;
CREATE DATABASE user_db3;

      3. Na barra de navegação superior, clique em Data Import.

      4. Na aba Batch Data Import, selecione o banco de dados de destino, faça upload do arquivo SQL correspondente, clique em Submit e, em seguida, clique em Execute Change. Na caixa de diálogo exibida, clique em Confirm Execution.

        Repita essa operação para importar os arquivos de dados correspondentes nos bancos de dados tpc_ds, user_db1, user_db2 e user_db3.

        Defina File Encoding como Auto Detect, Import Mode como Express Mode ou Safe Mode, e File Type como SQL Script, CSV ou Excel. Os anexos suportam formatos txt/sql/csv/xlsx/zip, até 5 GB.

    3. No console Hologres, crie um banco de dados chamado my_user para armazenar os dados da tabela de usuários mesclada.

      Para mais informações, consulte Criar um banco de dados.

    Configurar uma lista de permissões de IP

    Para permitir que o Flink acesse as instâncias ApsaraDB RDS for MySQL e Hologres, adicione o bloco CIDR do seu workspace Flink à lista de permissões de IP de ambas as instâncias.

    1. Obtenha o bloco CIDR do workspace Flink.

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

      2. Na lista de workspaces, localize o workspace desejado e escolha More > Workspace Details na coluna Actions.

      3. Na caixa de diálogo Workspace Details, visualize as informações de CIDR Block do vSwitch do Flink.

    2. Adicione o bloco CIDR do Flink à lista de permissões de IP da instância ApsaraDB RDS for MySQL.

      Para mais informações, consulte Configurar uma lista de permissões de IP.

      Na caixa de diálogo Modify Whitelist Group, insira o bloco CIDR totalmente gerenciado do Flink no campo Group Whitelist e clique em OK.

    3. Adicione o bloco CIDR do Flink à lista de permissões de IP da instância Hologres.

      Ao configurar uma conexão de dados no HoloWeb, defina Login Method como Password-free login for current user antes de configurar uma lista de permissões de IP para a conexão. Para mais informações, consulte Lista de permissões de IP.

      No HoloWeb, escolha Security Center na barra de navegação superior, clique em IP Whitelist no painel esquerdo e configure os seguintes parâmetros na caixa de diálogo Edit IP Whitelist:

      • Group: Selecione default

      • Database Restriction: Selecione ALL

      • User Restriction: Selecione ALL

      • IP Address: Insira o bloco CIDR do Flink em notação CIDR (por exemplo, 172.xx.0/19). Separe vários endereços IP com quebras de linha.

      Clique em OK para aplicar.

    Etapa 1: Desenvolver o job de ingestão

    1. Faça login no console de desenvolvimento Flink e crie um novo job.

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

      2. Clique em Blank Data Ingestion Draft.

        O Realtime Compute for Apache Flink oferece um conjunto rico de modelos de código, cada um com casos de uso específicos, exemplos de código e orientações. Clique em um modelo para conhecer os recursos do product e a sintaxe necessária para implementar sua lógica de negócios.

      3. Clique em Next.

      4. Na caixa de diálogo New Draft, configure os parâmetros.

        Parâmetro

        Descrição

        Exemplo

        Name

        Nome do job.

        Nota

        O nome do job deve ser exclusivo dentro do projeto atual.

        flink-test

        Location

        Pasta onde o arquivo de código do job será armazenado.

        Você também pode clicar no ícone 新建文件夹 ao lado de uma pasta existente para criar uma subpasta.

        Job Drafts

        Engine Version

        Versão do mecanismo Flink usada pelo job. Para obter informações sobre números de versão do mecanismo, compatibilidade de versões e datas importantes do ciclo de vida, consulte Versões do mecanismo.

        vvr-11.1-jdk11-flink-1.20

      5. Clique em OK.

    2. Copie o seguinte código de job para o editor de jobs.

      O código abaixo sincroniza todas as tabelas do banco de dados tpc_ds com o Hologres e mescla as tabelas de usuários fragmentadas em uma única tabela no Hologres:

      source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: tpc_ds.\.*,user_db[0-9]+.user[0-9]+
  server-id: 8601-8604
  # (Optional) Synchronize table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded chunks to prevent potential TaskManager OutOfMemory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to accelerate reading.
  scan.only.deserialize.captured.tables.changelog.enabled: true  

sink:
  type: hologres
  name: Hologres Sink
  endpoint: ****.hologres.aliyuncs.com:80
  dbname: cdcyaml_test
  username: ${secret_values.holo-username}
  password: ${secret_values.holo-password}
  sink.type-normalize-strategy: BROADEN
  
route:
  # Merge and synchronize the sharded user tables to the my_user.users table.
  - source-table: user_db[0-9]+.user[0-9]+
    sink-table: my_user.users
      Nota

      As tabelas do banco de dados MySQL tpc_ds são mapeadas diretamente para tabelas com nomes idênticos no destino, portanto, a seção route não requer configuração extra de mapeamento. Para sincronizar as tabelas com um banco de dados de nome diferente, como ods_tps_ds, configure o módulo route da seguinte forma:

      route:
  # Merge and synchronize the sharded user tables to the my_user.users table.
  - source-table: user_db[0-9]+.user[0-9]+
    sink-table: my_user.users
  # Rename the database for all tables under tpc_ds and synchronize them to ods_tps_ds.
  - source-table: tpc_ds.\.*
    sink-table: ods_tps_ds.<>
    replace-symbol: <>

    Etapa 2: Iniciar o job

    1. Na página Development > Data Ingestion, clique em Deploy. Na caixa de diálogo exibida, clique em Confirm.

      Na caixa de diálogo, insira Remarks, defina Job Tags e selecione uma fila na lista suspensa Deployment Target (padrão: default-queue). Observação: A implantação entra em vigor na próxima vez que o job for iniciado.

    2. Na página O&M > Deployments, clique em Start na coluna Actions do job desejado. Configure os parâmetros conforme necessário. Para mais informações, consulte Iniciar um job.

    3. Clique em Start.

      Após o início do job, visualize suas informações de execução e status na página Job Operations. Os status do job incluem Running, Failed e Stopped. Use o filtro suspenso Stream Jobs na parte superior para filtrar a lista de jobs por tipo.

    Etapa 3: Verificar a sincronização completa

    1. Faça login no Console de Gerenciamento do Hologres.

    2. Na aba Metadata Management, verifique se as 24 tabelas e seus dados estão presentes no banco de dados tpc_ds da instância Hologres.

      Acesse Instance > tpc_ds > public > Tables para ver tabelas como call_center, catalog_sales, customer e store_sales. Selecione a tabela store_sales e clique na aba Data Preview para visualizar colunas como ss_sold_date, ss_sold_time, ss_item_sk e ss_customer.

    3. Na aba Metadata Management, verifique o schema da tabela users no banco de dados my_user.

      A seguir, descrevemos o schema e os dados da tabela sincronizada.

      • Schema da tabela

        Após a sincronização, a tabela users contém as seguintes colunas:

        • _db_name (text, chave primária)

        • _table_name (text, chave primária)

        • id (int4, chave primária)

        • name (varchar 255, anulável)

        As colunas _db_name, _table_name e id formam uma chave primária composta.

        O schema da tabela users inclui duas colunas adicionais não encontradas nas tabelas MySQL de source: _db_name e _table_name. Essas colunas indicam o banco de dados e a tabela de source de cada linha e fazem parte da chave primária composta, garantindo a unicidade dos dados após a mesclagem das tabelas fragmentadas.

      • Dados da tabela

        No canto superior direito da página de informações da tabela users, clique em Query Table. Insira o seguinte comando e clique em Run.

        select * from users order by _db_name,_table_name,id;

        Os resultados da consulta mostram que os dados do usuário foram totalmente sincronizados em três bancos de dados: user_db1 (user01, user04, user07), user_db2 (user02, user05, user08) e user_db3 (user03, user06, user09), totalizando 9 registros com as colunas _db_name, _table_name, id e name.

    Etapa 4: Verificar a sincronização incremental

    Após a conclusão da sincronização completa, o job muda automaticamente para a fase de sincronização incremental sem intervenção manual. Verifique o valor de currentEmitEventTimeLag na aba Monitoring and Alerts para determinar a fase de sincronização de dados.

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

    2. Clique em Console na coluna Actions do workspace desejado.

    3. Na página O&M > Deployments, clique no nome do job desejado.

    4. Clique na aba Alarm (ou Metrics).

    5. Examine o gráfico currentEmitEventTimeLag para determinar a fase de sincronização de dados.

      数据曲线

      • Um valor igual a 0 indica a fase de sincronização completa.

      • Um valor maior que 0 indica a fase de sincronização incremental.

    6. Verifique a sincronização de dados e alterações de schema em tempo real.

      A fonte MySQL CDC suporta sincronização de dados e schema em tempo real durante a fase incremental. Para verificar isso, modifique o schema e os dados de uma tabela de usuários fragmentada no MySQL após o job entrar nessa fase.

      1. Faça login na instância ApsaraDB RDS for MySQL usando o DMS.

      2. No banco de dados user_db2, execute os seguintes comandos para modificar o schema da tabela user02 e para inserir e atualizar dados.

        USE `user_db2`;
ALTER TABLE `user02` ADD COLUMN `age` INT;   -- Add the age column.
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30); -- Insert a row that includes the age data.
UPDATE `user05` SET name='JARK' WHERE id=15;  -- Update another table and change the name to uppercase.

      3. No console Hologres, verifique as alterações no schema e nos dados da tabela users.

        No canto superior direito da página de informações da tabela users, clique em Query Table, insira o seguinte comando e clique em Run.

        select * from users order by _db_name,_table_name,id;

        A alteração de schema em user02 e as modificações de dados são propagadas em tempo real, mesmo que as tabelas fragmentadas tenham schemas diferentes. A tabela users do Hologres agora mostra a nova coluna age, o registro inserido para Tony e o registro atualizado para JARK.

        Os resultados da consulta mostram 11 linhas da tabela users em user_db1, user_db2 e user_db3. As alterações da sincronização incremental são refletidas: o usuário Tony (id 27) em user_db2 tem age atualizado para 30, e outro registro tem name alterado para JARK. As linhas restantes têm valores de idade iguais a \N, confirmando que a sincronização incremental está funcionando.

    (Opcional) Etapa 5: Configurar recursos do job

    Para obter melhor desempenho, ajuste os recursos do job, como concorrência, memória do TaskManager e CUs, com base no volume de dados.

    1. Na página O&M > Deployments, clique no nome do job desejado.

    2. Na aba Configuration, clique em Edit no canto superior direito da seção Resources.

    3. Defina manualmente parâmetros de recursos, como memória do TaskManager e concorrência.

    4. No lado direito da seção Resources, clique em Save.

    5. Reinicie o job.

      As alterações na configuração de recursos só entram em vigor após reiniciar o job.

    Documentos relacionados