Develop a data ingestion YAML job - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Contexto

Com configurações YAML, você define facilmente pipelines ETL complexos, que o sistema converte automaticamente em lógica de execução do Flink. A ingestão de dados com Flink CDC oferece uma solução robusta de integração baseada no Flink CDC. Essa solução suporta eficientemente a sincronização completa de banco de dados, de tabela única, de bancos e tabelas com sharding, além da descoberta automática de novas tabelas, tratamento de alterações de schema e colunas computadas personalizadas. Também há suporte para processamento ETL, filtragem por cláusula WHERE e poda de colunas. Essa abordagem declarativa simplifica significativamente o processo de integração de dados, aumentando a eficiência e a confiabilidade.

Vantagens do Flink CDC

No Realtime Compute for Apache Flink, desenvolva um job de ingestão de dados com Flink CDC, um job SQL ou um job DataStream para sincronizar dados. As seções a seguir descrevem as vantagens de usar um job de ingestão de dados com Flink CDC em comparação com as outras duas opções.

Flink CDC vs. Flink SQL

Jobs de ingestão de dados com Flink CDC e jobs SQL utilizam tipos de dados diferentes para transmissão:

Jobs SQL transmitem dados como RowData. Cada objeto RowData possui seu próprio tipo de alteração, que inclui quatro tipos principais: insert (+I), update before (-U), update after (+U) e delete (-D).
O Flink CDC usa SchemaChangeEvent para comunicar informações de mudança de schema, como criação de tabela, adição de coluna ou truncamento de tabela. Além disso, utiliza DataChangeEvent para comunicar alterações de dados, como inserções, atualizações e exclusões. Uma mensagem de atualização contém tanto o conteúdo anterior quanto o posterior, permitindo gravar os dados de alteração originais no sink.

A tabela a seguir lista as vantagens dos jobs de ingestão de dados com Flink CDC sobre os jobs SQL.

Ingestão com Flink CDC	Flink SQL
Detecção automática de schema e sincronização completa do banco de dados	Requer instruções `CREATE TABLE` e `INSERT` manuais
Suporta múltiplas políticas para alterações de schema	Não suporta alterações de schema
Preserva o changelog original	Quebra a estrutura original do changelog
Suporta leitura e escrita em múltiplas tabelas	Lê e escreve em uma única tabela

Em comparação com as instruções CTAS/CDAS, os jobs Flink CDC oferecem recursos mais avançados, incluindo:

Sincronização imediata de alterações de schema upstream, sem esperar que novas gravações de dados acionem a sincronização.
Preservação do changelog original, garantindo que as mensagens de atualização não sejam divididas.
Sincronização de mais tipos de alterações de schema, como TRUNCATE TABLE e DROP TABLE.
Mapeamento flexível de tabelas e definição de nomes de tabelas de sink.
Comportamentos de evolução de schema flexíveis e configuráveis.
Filtragem de dados usando cláusulas WHERE.
Suporte para poda de colunas.

Flink CDC vs. Flink DataStream

A tabela a seguir lista as vantagens dos jobs de ingestão de dados com Flink CDC sobre os jobs DataStream.

Ingestão com Flink CDC	Flink DataStream
Acessível para usuários de todos os níveis de habilidade.	Requer especialização em Java e sistemas distribuídos.
Oculta a complexidade subjacente e simplifica o desenvolvimento.	Exige conhecimento do framework Flink.
Formato YAML fácil de aprender.	Requer conhecimento de ferramentas como Maven para gerenciamento de dependências.
Alta reutilização para jobs existentes.	Difícil reutilização de código existente.

Limitações

Recomendamos o uso do Ververica Runtime (VVR) 11,1 para desenvolver jobs de ingestão de dados com Flink CDC. Caso precise usar o VVR 8.x, selecione o VVR 8.0.11.
Cada job suporta apenas uma source e um sink. Para ler de múltiplas sources ou gravar em múltiplos sinks, crie vários jobs Flink CDC.
Não é possível implantar jobs Flink CDC em um cluster de sessão.
Jobs de ingestão de dados com Flink CDC não suportam ajuste automático.

Conectores de ingestão de dados com Flink CDC

Para obter detalhes sobre os conectores de source e sink compatíveis com a ingestão de dados com Flink CDC, consulte Supported connectors.

Crie um job de ingestão de dados com Flink CDC

A partir de um modelo

Faça login no console do Realtime Compute for Apache Flink.
Na coluna Actions do workspace desejado, clique em Console.
No painel de navegação à esquerda, escolha Development > Data Ingestion.
Clique em e, em seguida, clique em New Draft with Template.
Selecione um modelo de sincronização de dados.

Vários modelos estão disponíveis, como MySQL para StarRocks, MySQL para Paimon e MySQL para Hologres.
Insira as informações do job (nome, local e versão do mecanismo) e clique em OK.
Configure as informações de source e sink para o job Flink CDC.

Para detalhes sobre a configuração de parâmetros, consulte a documentação dos conectores relevantes.

A partir de um job CTAS/CDAS

Importante

Se um job contiver várias instruções CXAS, o Flink detectará e converterá apenas a primeira.
Devido a diferenças no suporte a funções integradas entre o Flink SQL e o Flink CDC, as regras transform geradas podem não funcionar imediatamente. Revise-as e ajuste-as conforme necessário.
Se a source for MySQL e um job CTAS/CDAS original ainda estiver em execução, ajuste o server-id da source no job de ingestão de dados com Flink CDC para evitar conflitos.

Faça login no console do Realtime Compute for Apache Flink.
Na coluna Actions do workspace desejado, clique em Console.
No painel de navegação à esquerda, escolha Development > Data Ingestion.
Clique em e, em seguida, clique em New Draft from CTAS/CDAS. Selecione o job CTAS ou CDAS desejado e clique em OK.

A página de seleção exibe apenas jobs CTAS e CDAS válidos. Jobs ETL comuns e rascunhos com erros de sintaxe não aparecem na lista.
Insira as informações do job (nome, local e versão do mecanismo) e clique em OK.

A partir do Flink CDC open-source

Faça login no console do Realtime Compute for Apache Flink.
Na coluna Actions do workspace desejado, clique em Console.
No painel de navegação à esquerda, escolha Development > Data Ingestion.
Clique em e selecione New Draft. Insira o name e a engine version e clique em Create.
Cole o código YAML do seu job Flink CDC open-source no editor.
(Opcional) Clique em Validate.

Essa opção verifica erros de sintaxe, problemas de conectividade de rede e permissões.

Do zero

Faça login no console do Realtime Compute for Apache Flink.
Na coluna Actions do workspace desejado, clique em Console.
No painel de navegação à esquerda, escolha Development > Data Ingestion.
Clique em e selecione New Draft. Insira o name e a engine version e clique em Create.

Configure o job Flink CDC usando YAML. Exemplo:

# Required
source:
  # The type of the data source.
  type: <Replace with your source connector type>
  # The configuration for the data source. For details about configuration items, see the documentation for the corresponding connector.
  ...
# Required
sink:
  # The type of the sink.
  type: <Replace with your sink connector type>
  # The configuration for the sink. For details about configuration items, see the documentation for the corresponding connector.
  ...
# Optional
transform:
  # A transform rule for the flink_test.customers table.
  - source-table: flink_test.customers
    # Projection configuration. Specifies the columns to synchronize and performs data transformations.
    projection: id, username, UPPER(username) as username1, age, (age + 1) as age1, test_col1, __schema_name__ || '.' || __table_name__ identifier_name
    # Filter condition. Synchronizes only data where id is greater than 10.
    filter: id > 10
    # Description of the transform rule.
    description: append calculated columns based on source table
# Optional
route:
  # A route rule that specifies the mapping between the source table and the sink table.
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # Description of the route rule.
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # Description of the route rule.
    description: sync customers_suffix table
# Optional
pipeline:
  # The name of the job.
  name: MySQL to Hologres Pipeline

Nota

Em um job Flink CDC, separe uma chave e seu valor com dois pontos e um espaço. O formato é Key: Value.

A tabela a seguir descreve os blocos de código.

Obrigatório	Módulo	Descrição
Sim	source	Ponto inicial do pipeline de dados. O Flink CDC captura dados de alteração da source. Nota Para obter informações sobre os sistemas de source compatíveis atualmente, consulte Source and sink modules. Para detalhes sobre os itens de configuração da source, veja a documentação do conector correspondente. Use variáveis para gerenciar informações confidenciais. Para mais informações, consulte Variable Management.
Sim	sink	Ponto final do pipeline de dados. O Flink CDC transmite as alterações de dados capturadas para o sistema de sink. Nota Para obter informações sobre os sistemas de sink compatíveis atualmente, consulte Source and sink modules. Para detalhes sobre os itens de configuração do sink, veja a documentação do conector correspondente. Use variáveis para gerenciar informações confidenciais. Para mais informações, consulte Variable Management.
Não	pipeline (pipeline de dados)	Define configurações básicas para todo o job de pipeline de dados, como o nome do pipeline.
	transform	Especifica regras de transformação de dados para operar nos dados durante o fluxo pelo pipeline do Flink. Suporta processamento ETL, filtragem por cláusula `WHERE`, poda de colunas e colunas computadas. Se for necessário transformar dados de alteração brutos capturados pelo Flink CDC para adequá-los a sistemas downstream específicos, use o bloco `transform`.
	route	Sem essa configuração, o job executa a sincronização completa do banco de dados ou da tabela de destino por padrão. Em alguns casos, envie dados de alteração capturados para destinos diferentes com base em regras específicas. O módulo `route` permite especificar flexivelmente o mapeamento entre source e sink, direcionando dados para diferentes alvos.

Para mais informações sobre a sintaxe e a configuração de cada módulo, consulte Reference for Flink CDC data ingestion jobs.

O código a seguir exemplifica a sincronização de todas as tabelas do banco de dados app_db no MySQL para um banco de dados no Hologres.

source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404
  # (Optional) Synchronize data from tables newly created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize dispatching unbounded chunks to prevent potential TaskManager OutOfMemory issues.
  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: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}
pipeline:
  name: Sync MySQL Database to Hologres

(Opcional) Clique em Validate.

Essa opção verifica erros de sintaxe, problemas de conectividade de rede e permissões.

Tópicos relacionados

Após concluir o desenvolvimento de um job Flink CDC, implante-o. Para mais informações, consulte Deploy a job.
Para criar rapidamente um job Flink CDC que sincronize dados de um banco de dados MySQL para o StarRocks, consulte Tutorial: Create a Flink CDC data ingestion job.