Implementar ingestão em tempo real - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Este tópico aborda as melhores práticas para transmitir alterações em tempo real do MySQL ou Kafka para um data warehouse Hologres por meio de um job de ingestão de dados Change Data Capture (CDC) YAML.

Sincronizar do MySQL para o Hologres

Utilize a ingestão de dados CDC YAML para sincronizar informações com o Hologres e construir um data warehouse em tempo real. Essa abordagem combina as capacidades de processamento em tempo real do Flink com recursos do Hologres, como binlog, coexistência de armazenamento linha-coluna e forte isolamento de resources, garantindo eficiência e escalabilidade no processamento e na análise de dados. O conector do Hologres oferece suporte a:

Criação automática de tabelas e sincronização de schema
Armazenamento híbrido linha-coluna e leituras incrementais baseadas em Binlog
Mapeamento de tipos tolerante para lidar com alterações de schema na source sem causar falhas no job

O YAML abaixo define a configuração mínima necessária para sincronizar um banco de dados MySQL inteiro com o Hologres:

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: holo_test\.*
  server-id: 8601-8604

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

Parâmetros opcionais para configurar na primeira execução do job:

Parâmetro	Finalidade	Referência
`include-comments.enabled: true`	Sincroniza comentários de tabelas e campos	MySQL connector
`scan.incremental.snapshot.unbounded-chunk-first.enabled: true`	Prioriza shards ilimitados para evitar erros de OutOfMemory no TaskManager	MySQL connector
`scan.only.deserialize.captured.tables.changelog.enabled: true`	Ativa filtros de parsing para acelerar a leitura de dados	MySQL connector

Lidar com alterações de tipo na source

Como o conector do Hologres não suporta eventos de alteração de tipo de coluna, configure sink.type-normalize-strategy para mapear tipos do MySQL para tipos mais abrangentes no Hologres, evitando assim falhas no job diante de mudanças de schema. O valor padrão é STANDARD.

Estratégia	Comportamento
`STANDARD`	Mapeamento de tipos padrão
`BROADEN`	Mapeia para tipos compatíveis mais amplos
`ONLY_BIGINT_OR_TEXT`	Converte todos os tipos para `int8` ou `text`

Por exemplo, ao usar ONLY_BIGINT_OR_TEXT, se uma coluna no MySQL for alterada de INT para BIGINT, ambos os tipos serão mapeados como int8 no Hologres, permitindo que o job continue normalmente sem reportar erros de conversão.

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: holo_test\.*
  server-id: 8601-8604
  # (Optional) Synchronize table comments and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded shards to prevent potential TaskManager OutOfMemory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to accelerate data reads.
  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: ONLY_BIGINT_OR_TEXT

Para consultar a lista completa de mapeamentos de tipos suportados, consulte Mapeamento de tipos do conector Hologres para jobs YAML de ingestão de dados.

Gravar em tabelas particionadas

Ao utilizar o conector do Hologres como sink, é possível gravar dados diretamente em tabelas particionadas. Para mais detalhes, veja Gravar dados em tabelas particionadas.

Sincronizar do Kafka para o Hologres

Esta seção pressupõe que as alterações do MySQL já estejam sendo enviadas para o Kafka — por exemplo, utilizando a configuração descrita em Implementar distribuição em tempo real. O pipeline lê os dados do Kafka e os grava no Hologres.

Considere um tópico do Kafka chamado inventory contendo dados de duas tabelas, customers e products, no formato debezium-json. O job a seguir sincroniza ambas as tabelas com suas respectivas correspondentes no Hologres:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: 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: ONLY_BIGINT_OR_TEXT

transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Formatos suportados

A source do Kafka aceita três formatos de mensagem:

Formato	Observações
`json`	Formato JSON padrão
`canal-json`	Formato JSON compatível com Canal
`debezium-json`	Formato padrão para dados CDC originados do Debezium

Requisito de chave primária para debezium-json

Quando o formato utilizado é debezium-json, a source do Kafka não infere chaves primárias automaticamente. Defina a chave primária explicitamente usando uma regra de transform:

transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Tabelas distribuídas

Caso os dados de uma única tabela estejam espalhados por múltiplas partições, ou se for necessário mesclar tabelas fragmentadas (sharded) entre partições diferentes, defina debezium-json.distributed-tables ou canal-json.distributed-tables como true.

Inferência de schema

A source do Kafka oferece suporte a diversas políticas de inferência de schema. Configure a política desejada através do parâmetro schema.inference.strategy. Para mais informações, consulte Message Queue for Apache Kafka.