Use Flink CDC to synchronize data to a Delta table - MaxCompute: Central de Documentação do Alibaba Cloud

O MaxCompute oferece um conector Flink Change Data Capture (CDC) que sincroniza dados de alteração de fontes como MySQL para tabelas padrão do MaxCompute ou Delta Tables em tempo real. Use o conector de pipeline do Flink CDC para configurar um pipeline de extração, transformação e carga (ETL) em streaming do MySQL para o MaxCompute.

O conector oferece suporte a:

Criação automática de tabelas com base na estrutura da tabela de origem
Evolução de schema: alterações no schema da tabela de origem são propagadas para o MaxCompute
Sincronização completa de banco de dados em várias tabelas
Roteamento de tabelas: mapeia tabelas de origem para diferentes nomes de tabelas de destino, incluindo mesclagem de banco de dados fragmentado

Pré-requisitos

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

Um projeto do MaxCompute. Para criar um, consulte Create a MaxCompute project.
Docker e Docker Compose instalados em sua máquina
Java Runtime Environment (JRE) compatível com Flink 1.18.0

Observações de uso

Seleção de tipo de tabela: Se a tabela de origem tiver uma chave primária, o sistema criará uma Delta Table automaticamente. Caso contrário, será criada uma tabela padrão do MaxCompute.
Comportamento de gravação para tabelas padrão: O sistema ignora operações DELETE e trata operações UPDATE como operações INSERT.
Semântica de entrega: Há suporte apenas à semântica at-least-once. Para implementar gravações idempotentes, use uma Delta Table. A chave primária na Delta Table gerencia a deduplicação.
Evolução de schema: Alterações no schema da tabela de origem são propagadas para o MaxCompute. Novas colunas são sempre adicionadas ao final. Os tipos de dados das colunas só podem ser alterados para um tipo compatível. Para obter detalhes, consulte Change the data type of a column.
Mesclagem de banco de dados fragmentado: É possível mesclar várias tabelas de origem em uma única tabela de destino usando expressões regulares. Ainda não há suporte para mesclagem de tabelas com valores de chave primária duplicados entre as origens.

Primeiros passos

Este tutorial configura um trabalho completo de ETL em streaming que sincroniza dados de alteração do MySQL para o MaxCompute. Ele abrange sincronização completa de banco de dados, evolução de schema e roteamento de tabelas.

Configure os ambientes

Configure um cluster Flink no modo standalone

Baixe o arquivo flink-1.18.0-bin-scala_2.12.tgz e descompacte-o para obter o diretório flink-1.18.0. No diretório flink-1.18.0, execute o comando abaixo para definir FLINK_HOME:
```
export FLINK_HOME=$(pwd)
```

Abra o arquivo $FLINK_HOME/conf/flink-conf.yaml em um editor de texto e adicione os seguintes parâmetros:

# Enable checkpointing. Run a checkpoint every 3 seconds.
# For production, set the checkpoint interval to at least 30 seconds.
execution.checkpointing.interval: 3000

# The Flink CDC pipeline connector relies on Flink's communication mechanism
# for data synchronization. Increase the timeout to prevent connection drops.
pekko.ask.timeout: 60s

Inicie o cluster Flink:
```
./bin/start-cluster.sh
```
Após a inicialização, abra http://localhost:8081/ em um navegador para acessar a interface web do Flink. Execute start-cluster.sh várias vezes para adicionar TaskManagers.

Configure um ambiente MySQL

Este tutorial usa o Docker Compose para executar o MySQL.

Crie um arquivo chamado docker-compose.yaml com o seguinte conteúdo:

Parâmetro	Descrição
`version`	Versão do Docker Compose
`image`	Versão da imagem Docker. Defina como `debezium/example-mysql:1.1`
`ports`	Mapeamento de porta do MySQL
`environment`	Senha root do MySQL e credenciais de usuário

version: '2.1'
services:
  mysql:
    image: debezium/example-mysql:1.1
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw

No diretório que contém o arquivo docker-compose.yaml, inicie os contêineres:
```
docker-compose up -d
```
Execute docker ps para confirmar se todos os contêineres estão em execução.

Prepare dados no MySQL

Conecte-se ao contêiner MySQL:

docker-compose exec mysql mysql -uroot -p123456

Crie o banco de dados app_db e preencha três tabelas:

Crie o banco de dados:
```
CREATE DATABASE app_db;
USE app_db;
```

Crie e preencha a tabela orders:

CREATE TABLE `orders` (
  `id` INT NOT NULL,
  `price` DECIMAL(10,2) NOT NULL,
  PRIMARY KEY (`id`)
);

INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);

Crie e preencha a tabela shipments:

CREATE TABLE `shipments` (
  `id` INT NOT NULL,
  `city` VARCHAR(255) NOT NULL,
  PRIMARY KEY (`id`)
);

INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');

Crie e preencha a tabela products:

CREATE TABLE `products` (
  `id` INT NOT NULL,
  `product` VARCHAR(255) NOT NULL,
  PRIMARY KEY (`id`)
);

INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');

Envie um trabalho de pipeline YAML

Baixe os pacotes JAR necessários:
- Pacote Flink CDC: Baixe o arquivo flink-cdc-3.1.1-bin.tar.gz e descompacte-o para obter o diretório flink-cdc-3.1.1. Mova o conteúdo dos diretórios bin, lib, log e conf para os diretórios correspondentes em flink-1.18.0.
- Pacotes de conectores: Baixe os seguintes JARs e coloque-os no diretório flink-1.18.0/lib: > Nota: Os links de download acima apontam apenas para versões lançadas. Para usar uma versão SNAPSHOT, compile o código source da branch master ou release localmente.
  - MySQL pipeline connector 3.1.1
  - MaxCompute pipeline connector 3.1.1
- Driver JDBC: Baixe o arquivo mysql-connector-java-8.0.27.jar. Passe-o para a CLI do Flink CDC com --jar ou coloque-o em $FLINK_HOME/lib e reinicie o cluster. Os conectores CDC não incluem mais drivers JDBC.

Crie o arquivo YAML do pipeline. O exemplo a seguir (mysql-to-maxcompute.yaml) sincroniza todas as tabelas do banco de dados app_db:

Espaço reservado	Descrição
`${your_accessId}`	Seu AccessKey ID. Obtenha-o na página AccessKey Pair.
`${your_accessKey}`	O AccessKey secret correspondente ao seu AccessKey ID.
`${your_maxcompute_endpoint}`	O endpoint do MaxCompute para sua região e rede. Consulte Endpoints.
`${your_project}`	O nome do seu projeto MaxCompute. Encontre-o no console do MaxCompute em Workspace > Projects.

################################################################################
# Description: Sync MySQL all tables to MaxCompute
################################################################################
source:
  type: mysql
  hostname: localhost
  port: 3306
  username: root
  password: 123456
  tables: app_db\.*
  server-id: 5400-5404
  server-time-zone: UTC

# Configure the accessId, accessKey, endpoint, and project parameters.
sink:
  type: maxcompute
  name: MaxComputeSink
  accessId: ${your_accessId}
  accessKey: ${your_accessKey}
  endpoint: ${your_maxcompute_endpoint}
  project: ${your_project}
  bucketsNum: 8

pipeline:
  name: Sync MySQL Database to MaxCompute
  parallelism: 1

Substitua os seguintes espaços reservados: Para todos os parâmetros de origem disponíveis, consulte Conector MySQL. Para todos os parâmetros de destino disponíveis, consulte Configuração do conector.

Envie o trabalho de pipeline para o cluster Flink standalone:
```
./bin/flink-cdc.sh mysql-to-maxcompute.yaml
```
Um envio bem-sucedido retorna uma saída semelhante a:
```
Pipeline has been submitted to cluster.
Job ID: f9f9689866946e25bf151ecc179ef46f
Job Description: Sync MySQL Database to MaxCompute
```
A interface web do Flink mostra um trabalho em execução chamado Sync MySQL Database to MaxCompute.

No MaxCompute, verifique se as tabelas foram criadas e se os dados foram gravados:

-- Query the orders table.
read orders;

-- Expected result:
+------------+------------+
| id         | price      |
+------------+------------+
| 1          | 4          |
| 2          | 100        |
+------------+------------+

-- Query the shipments table.
read shipments;

-- Expected result:
+------------+------------+
| id         | city       |
+------------+------------+
| 1          | beijing    |
| 2          | xian       |
+------------+------------+

-- Query the products table.
read products;

-- Expected result:
+------------+------------+
| id         | product    |
+------------+------------+
| 3          | Peanut     |
| 1          | Beer       |
| 2          | Cap        |
+------------+------------+

Sincronize alterações em tempo real

Com o pipeline em execução, as alterações no MySQL são refletidas no MaxCompute em tempo real. Os exemplos a seguir usam a tabela orders para demonstrar operações de inserção, alteração de schema, atualização e exclusão.

Conecte-se ao contêiner MySQL:

docker-compose exec mysql mysql -uroot -p123456

Insira uma linha na tabela orders:

INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

Execute read orders; no MaxCompute. A nova linha aparece:

+------------+------------+
| id         | price      |
+------------+------------+
| 3          | 100        |
| 1          | 4          |
| 2          | 100        |
+------------+------------+

Adicione uma coluna à tabela orders no MySQL:

ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

Execute read orders; no MaxCompute. A nova coluna amount é propagada automaticamente:

+------------+------------+------------+
| id         | price      | amount     |
+------------+------------+------------+
| 3          | 100        | NULL       |
| 1          | 4          | NULL       |
| 2          | 100        | NULL       |
+------------+------------+------------+

Atualize uma linha na tabela orders:

UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

Execute read orders; no MaxCompute:

+------------+------------+------------+
| id         | price      | amount     |
+------------+------------+------------+
| 3          | 100        | NULL       |
| 1          | 100        | 100.00     |
| 2          | 100        | NULL       |
+------------+------------+------------+

Exclua uma linha da tabela orders:

DELETE FROM app_db.orders WHERE id=2;

Execute read orders; no MaxCompute:

+------------+------------+------------+
| id         | price      | amount     |
+------------+------------+------------+
| 3          | 100        | NULL       |
| 1          | 100        | 100.00     |
+------------+------------+------------+

Roteie tabelas

O Flink CDC permite rotear schemas e dados de tabelas de origem para diferentes nomes de tabelas de destino. Use este recurso para migrações de banco de dados, renomeação de tabelas ou mesclagem de bancos de dados fragmentados.

O exemplo YAML a seguir roteia cada tabela de origem para uma tabela de destino com um nome diferente:

################################################################################
# Description: Sync MySQL all tables to MaxCompute
################################################################################
source:
  type: mysql
  hostname: localhost
  port: 3306
  username: root
  password: 123456
  tables: app_db\.*
  server-id: 5400-5404
  server-time-zone: UTC

# Configure the accessId, accessKey, endpoint, and project parameters.
sink:
  type: maxcompute
  name: MaxComputeSink
  accessId: ${your_accessId}
  accessKey: ${your_accessKey}
  endpoint: ${your_maxcompute_endpoint}
  project: ${your_project}
  bucketsNum: 8

route:
  - source-table: app_db.orders
    sink-table: ods_db.ods_orders
  - source-table: app_db.shipments
    sink-table: ods_db.ods_shipments
  - source-table: app_db.products
    sink-table: ods_db.ods_products

pipeline:
  name: Sync MySQL Database to MaxCompute
  parallelism: 1

O campo source-table aceita expressões regulares. Para mesclar várias tabelas fragmentadas em uma única tabela de destino:

route:
  - source-table: app_db.order\.*
    sink-table: ods_db.ods_orders

Essa configuração agrega dados de app_db.order01, app_db.order02, app_db.order03 e tabelas semelhantes em ods_db.ods_orders.

Atualmente, não há suporte para cenários com valores de chave primária duplicados em várias tabelas de origem. O suporte será adicionado em versões futuras do conector.

Para todos os parâmetros de roteamento, consulte Roteamento.

Limpeza

Após concluir o tutorial, pare todos os componentes em execução.

Pare os contêineres Docker. Execute o seguinte comando no diretório que contém docker-compose.yaml:
```
docker-compose down
```
Pare o cluster Flink. Execute o seguinte comando no diretório flink-1.18.0:
```
./bin/stop-cluster.sh
```

Apêndice

Configuração do conector

O conector de destino do MaxCompute aceita os seguintes parâmetros de configuração.

Os dados são liberados para o MaxCompute quando o buffer de uma partição ou bucket excede o limiar de tamanho configurado ou quando um checkpoint é acionado.

Parâmetro	Obrigatório	Padrão	Tipo	Descrição
`type`	Sim	—	String	O tipo de conector. Defina como `maxcompute`.
`name`	Não	—	String	O nome de exibição do destino.
`accessId`	Sim	—	String	Seu AccessKey ID. Obtenha-o na página AccessKey Pair.
`accessKey`	Sim	—	String	O AccessKey secret correspondente ao seu AccessKey ID.
`endpoint`	Sim	—	String	O endpoint do MaxCompute. Configure com base na sua região e método de conexão de rede. Consulte Endpoints.
`project`	Sim	—	String	O nome do projeto MaxCompute. Encontre-o no console do MaxCompute em Workspace > Projects.
`tunnelEndpoint`	Não	—	String	O endpoint do MaxCompute Tunnel. O roteamento automático funciona na maioria dos casos. Configure este parâmetro apenas ao usar um ambiente de rede especial, como um proxy.
`quotaName`	Não	—	String	O nome do grupo de recursos exclusivo para o MaxCompute Tunnel. Se não definido, o sistema usará um grupo de recursos compartilhados.
`stsToken`	Não	—	String	Necessário ao usar um token STS emitido por uma função RAM para autenticação.
`bucketsNum`	Não	16	Integer	O número de buckets para criação automática de Delta Table. Consulte Near-real-time data warehouse overview.
`compressAlgorithm`	Não	`zlib`	String	O algoritmo de compressão para gravações. Valores válidos: `raw` (sem compressão), `zlib`, `snappy`.
`totalBatchSize`	Não	`64MB`	String	O tamanho máximo de buffer por partição ou tabela não particionada. Quando excedido, os dados são liberados para o MaxCompute. Cada partição ou tabela não particionada mantém um buffer independente.
`bucketBatchSize`	Não	`4MB`	String	O tamanho máximo de buffer por bucket. Aplica-se apenas a gravações em Delta Table. Quando excedido, os dados são liberados para o MaxCompute. Cada bucket mantém um buffer independente.
`numCommitThreads`	Não	16	Integer	O número máximo de partições ou tabelas processadas simultaneamente durante um checkpoint.
`numFlushConcurrent`	Não	4	Integer	O número máximo de buckets gravados no MaxCompute simultaneamente. Aplica-se apenas a gravações em Delta Table.
`retryTimes`	Não	3	Integer	O número máximo de tentativas em caso de erros de rede.
`sleepMillis`	Não	—	Long	A duração da espera entre novas tentativas em caso de erros de rede. Unidade: milissegundos.

Mapeamentos de localização de tabela

Quando o conector cria uma tabela automaticamente, ele mapeia as informações de localização da origem para o MaxCompute da seguinte forma.

Importante

Se o projeto MaxCompute não oferecer suporte a modelos de schema, cada tarefa de sincronização poderá sincronizar apenas um banco de dados MySQL. O conector ignora as informações TableId.namespace para outras fontes de dados.

Objeto no Flink CDC	Localização no MaxCompute	Localização no MySQL
Projeto (do arquivo de configuração)	Projeto	—
`TableId.namespace`	Schema (requer suporte a modelo de schema; ignorado se o projeto não oferecer suporte a modelos de schema)	Banco de dados
`TableId.tableName`	Tabela	Tabela

Mapeamentos de tipo de dados

A tabela a seguir mostra como os tipos de dados do Flink são mapeados para os tipos de dados do MaxCompute.

Tipo de dados Flink	Tipo de dados MaxCompute	Observação
CHAR/VARCHAR	STRING	O MaxCompute armazena todos os tipos de string como STRING.
BOOLEAN	BOOLEAN	—
BINARY/VARBINARY	BINARY	—
DECIMAL	DECIMAL	—
TINYINT	TINYINT	—
SMALLINT	SMALLINT	—
INTEGER	INTEGER	—
BIGINT	BIGINT	—
FLOAT	FLOAT	—
DOUBLE	DOUBLE	—
TIME_WITHOUT_TIME_ZONE	STRING	O MaxCompute não possui um tipo TIME nativo. Os valores de tempo são armazenados como strings.
DATE	DATE	—
TIMESTAMP_WITHOUT_TIME_ZONE	TIMESTAMP_NTZ	Armazenado sem informações de fuso horário.
TIMESTAMP_WITH_LOCAL_TIME_ZONE	TIMESTAMP	—
TIMESTAMP_WITH_TIME_ZONE	TIMESTAMP	—
ARRAY	ARRAY	—
MAP	MAP	—
ROW	STRUCT	—