Realtime Compute for Apache Flink:MySQL

O conector MySQL oferece suporte aos seguintes itens:

• Processamento unificado em lote e streaming que permite a leitura de dados completos e incrementais, eliminando a necessidade de manter dois processos separados.

• Leitura concorrente de dados completos para dimensionamento horizontal de desempenho.

• Transição contínua da leitura de dados completos para incrementais, com redução automática de escala para economizar recursos computacionais.

• Transferência de dados retomável durante a fase de leitura completa, aumentando a estabilidade.

• Leitura de dados completos sem bloqueio, sem impacto nos services online.

• Compatibilidade com a leitura de logs de backup do ApsaraDB RDS for MySQL.

• Análise paralela de arquivos de log binário para reduzir a latência de leitura.

SQL

Utilize o conector MySQL em jobs SQL como tabela de origem, tabela de dimensão ou tabela de destino.

Sintaxe

CREATE TEMPORARY TABLE mysqlcdc_source (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

Parâmetros WITH

• Geral

  Parâmetro	Descrição	Obrigatório	Tipo de dados	Valor padrão	Observações
  connector	O tipo da tabela.	Sim	STRING	Nenhum	Ao usar como tabela source, defina este parâmetro como mysql-cdc ou mysql. Ambos são equivalentes. Para tabelas de dimensão ou sink, o valor deve ser mysql.
  hostname	O endereço IP ou hostname do banco de dados MySQL.	Sim	STRING	Nenhum	Recomendamos especificar um endereço de virtual private cloud (VPC). Nota Se o banco de dados MySQL e o Realtime Compute for Apache Flink não estiverem na mesma VPC, estabeleça uma conexão de rede entre VPCs ou use um endpoint público para acessar o banco de dados. Para mais informações, consulte Gerenciar e operar workspaces e Como um cluster Flink totalmente gerenciado acessa a Internet?.
  username	O nome de usuário do service de banco de dados MySQL.	Sim	STRING	Nenhum	Nenhuma.
  password	A senha do service de banco de dados MySQL.	Sim	STRING	Nenhum	Nenhuma.
  database-name	O nome do banco de dados MySQL.	Sim	STRING	Nenhum	Para bancos de dados usados como tabela source, utilize expressões regulares no nome do banco para ler dados de múltiplos bancos. Ao empregar expressões regulares, evite os símbolos ^ e $ para corresponder ao início e fim da string. Consulte as observações do parâmetro table-name para mais detalhes.
  table-name	O nome da tabela MySQL.	Sim	STRING	Nenhum	É possível aplicar expressões regulares ao nome da tabela source para extrair dados de várias tabelas. Na leitura de múltiplas tabelas MySQL, envie várias instruções CTAS como um único job. Essa prática evita a ativação de múltiplos listeners de log binário, melhorando a performance e a eficiência. Para saber mais, veja Múltiplas instruções CTAS: Envio como job único. No uso de expressões regulares, não inclua os símbolos ^ e $ para delimitar o início e o fim da string. Veja a nota abaixo para mais informações. Nota Quando uma tabela source MySQL CDC corresponde nomes de tabelas via expressão regular, ela concatena database-name e table-name especificados com a string \\., formando uma expressão regular de caminho completo. Antes da VVR 8.0.1, utilizava-se o caractere .. O connector aplica essa expressão regular para identificar os nomes totalmente qualificados das tabelas no banco de dados MySQL. Por exemplo, ao definir 'database-name'='db_.' e 'table-name'='tb_.+', o connector utiliza a expressão regular db_.\\.tb_.+ para verificar os nomes completos das tabelas e determinar quais devem ser lidas. Em versões anteriores à VVR 8.0.1, a expressão regular era db_.*.tb_.+.
  port	O número da porta do service de banco de dados MySQL.	Não	INTEGER	3306	Nenhuma.

• Exclusivo para tabela source

  Parâmetro	Descrição	Obrigatório	Tipo de dados	Valor padrão	Observações
  server-id	ID numérico do cliente de banco de dados.	Não	STRING	Um valor aleatório entre 5400 e 6400 é gerado.	Este ID deve ser globalmente único no cluster MySQL. Defina um ID diferente para cada job que se conecta ao mesmo banco de dados. Este parâmetro também aceita um formato de intervalo de ID, como 5400-5408. Quando a leitura incremental está ativada, há suporte para leitura simultânea. Nesse caso, defina um intervalo de IDs para que cada leitor simultâneo utilize um ID distinto. Para mais informações, consulte Usar Server ID.
  scan.incremental.snapshot.enabled	Define se os snapshots incrementais devem ser ativados.	Não	BOOLEAN	true	Os snapshots incrementais vêm ativados por padrão. Esse mecanismo moderno de leitura de snapshots completos oferece diversas vantagens em relação ao método antigo: Leitura paralela dos dados completos pela source. Suporte a checkpoints no nível de chunk durante a leitura completa. Dispensa a aquisição de bloqueio global de leitura (FLUSH TABLES WITH read lock) na leitura completa. Para habilitar a leitura simultânea na source, cada leitor concorrente precisa de um server ID exclusivo. Por isso, o server-id deve ser definido como um intervalo, como 5400-6400, cujo tamanho seja maior ou igual à concorrência configurada. Nota Este item de configuração foi removido no Ververica Runtime (VVR) 11.1 e versões posteriores.
  scan.incremental.snapshot.chunk.size	Tamanho de cada chunk, medido em número de linhas.	Não	INTEGER	8096	Com a leitura de snapshot incremental ativada, a tabela é dividida em vários chunks para processamento. Os dados de cada chunk permanecem em cache na memória até que a leitura seja concluída. Chunks com menos linhas aumentam o número total de divisões na tabela. Embora isso reduza a granularidade da recuperação de falhas, pode provocar erros de falta de memória (OOM) e diminuir o throughput geral. Portanto, avalie as compensações e configure um tamanho de chunk adequado.
  scan.snapshot.fetch.size	Quantidade máxima de registros buscados por vez durante a leitura completa da tabela.	Não	INTEGER	1024	Nenhuma.
  scan.startup.mode	Modo de inicialização para consumo de dados.	Não	STRING	initial	Valores válidos: initial (padrão): Na primeira execução, o conector varre todos os dados históricos e depois lê os dados mais recentes do log binário. latest-offset: O conector ignora os dados históricos na primeira execução. A leitura começa pelo final do log binário, capturando apenas as alterações ocorridas após o início do conector. earliest-offset: Sem varredura de dados históricos. A leitura inicia no log binário mais antigo disponível. specific-offset: Ignora dados históricos e começa a partir de um offset específico do log binário. É possível definir esse ponto configurando scan.startup.specific-offset.file e scan.startup.specific-offset.pos, ou usando apenas scan.startup.specific-offset.gtid-set para iniciar a partir de um conjunto GTID específico. timestamp: Não lê dados históricos. Inicia a leitura do log binário a partir de um carimbo de data/hora definido pelo parâmetro scan.startup.timestamp-millis, em milissegundos. Importante Ao utilizar os modos de inicialização earliest-offset, specific-offset ou timestamp, garanta que o schema da tabela correspondente não sofra alterações entre a posição de consumo especificada no log binário e o momento de início do job. Essa precaução evita erros decorrentes de incompatibilidade de schema.
  scan.startup.specific-offset.file	Nome do arquivo de log binário que define o offset inicial no modo specific-offset.	Não	STRING	Nenhum	O uso deste parâmetro exige que scan.startup.mode esteja definido como specific-offset. Exemplo de formato de nome de arquivo: mysql-bin.000003.
  scan.startup.specific-offset.pos	Posição dentro do arquivo de log binário especificado para definir o offset inicial no modo specific-offset.	Não	INTEGER	Nenhum	Para usar este parâmetro, defina scan.startup.mode como specific-offset.
  scan.startup.specific-offset.gtid-set	Conjunto GTID que determina o offset inicial no modo specific-offset.	Não	STRING	Nenhum	Este parâmetro requer que scan.startup.mode esteja configurado como specific-offset. Exemplo de formato de conjunto GTID: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	Carimbo de data/hora em milissegundos para o offset inicial no modo timestamp.	Não	LONG	Nenhum	Ao utilizar este parâmetro, configure scan.startup.mode como timestamp. A unidade do carimbo de data/hora é milissegundos. Importante Ao especificar um horário, o MySQL CDC tenta ler o evento inicial de cada arquivo de log binário para identificar seu carimbo de data/hora e localizar o arquivo correspondente. Certifique-se de que o arquivo de log binário referente ao carimbo especificado não tenha sido limpo do banco de dados e ainda esteja acessível para leitura.
  server-time-zone	Fuso horário da sessão utilizado pelo banco de dados.	Não	STRING	Se omitido, o sistema adota o fuso horário do ambiente de execução do job Flink como fuso horário do servidor de banco de dados, correspondendo à zona selecionada.	Exemplo: Asia/Shanghai. Este parâmetro controla a conversão do tipo TIMESTAMP do MySQL para o tipo STRING. Para mais detalhes, consulte Valores temporais do Debezium.
  debezium.min.row.count.to.stream.results	Quando a quantidade de linhas da tabela excede este valor, o sistema utiliza o modo de leitura em lote.	Não	INTEGER	1000	O Flink lê dados de uma tabela source MySQL de duas formas: Leitura completa: Carrega todos os dados da tabela diretamente na memória. É rápida, mas consome memória proporcional ao volume. Tabelas muito grandes podem causar erros de OOM. Leitura em lote: Processa os dados em lotes sucessivos com um número definido de linhas até concluir a leitura. Evita riscos de OOM em tabelas extensas, embora seja mais lenta.
  connect.timeout	Tempo máximo de espera para estabelecer conexão com o servidor MySQL antes de tentar novamente.	Não	DURATION	30s	Nenhuma.
  connect.max-retries	Número máximo de tentativas após falha na conexão com o serviço de banco de dados MySQL.	Não	INTEGER	3	Nenhuma.
  connection.pool.size	Tamanho do pool de conexões do banco de dados.	Não	INTEGER	20	O pool de conexões permite reutilizar conexões existentes, reduzindo o número total de conexões abertas no banco de dados.
  jdbc.properties.*	Parâmetros de conexão personalizados na URL JDBC.	Não	STRING	Nenhum	É possível passar parâmetros de conexão personalizados. Por exemplo, para desativar o protocolo SSL, configure 'jdbc.properties.useSSL' = 'false'. Para consultar todos os parâmetros de conexão suportados, veja Propriedades de Configuração do MySQL.
  debezium.*	Parâmetros personalizados do Debezium para leitura de logs binários.	Não	STRING	Nenhum	Permite a passagem de parâmetros personalizados do Debezium. Por exemplo, use 'debezium.event.deserialization.failure.handling.mode'='ignore' para definir a lógica de tratamento de erros de parsing. Aviso Evite modificar parâmetros do Debezium indiscriminadamente, pois isso pode levar à leitura incorreta de dados pelo conector. Por exemplo, não é permitido configurar o parâmetro debezium.binlog.buffer.size.
  heartbeat.interval	Intervalo em que a source avança o offset do log binário por meio de eventos de heartbeat.	Não	DURATION	30s	Eventos de heartbeat mantêm o offset do log binário atualizado na source, sendo essenciais para tabelas MySQL com baixa frequência de atualização. Nessas tabelas, o offset não avança automaticamente. Os heartbeats forçam esse avanço, prevenindo problemas causados pela expiração do offset. Um offset expirado pode tornar o job irrecuperável, exigindo reinício sem estado.
  scan.incremental.snapshot.chunk.key-column	Define a coluna usada como chave de divisão para sharding durante a fase de snapshot.	Consulte a coluna Observações.	STRING	Nenhum	Obrigatório para tabelas sem chave primária. A coluna escolhida deve ser do tipo não nulo (NOT NULL). Opcional para tabelas com chave primária. Apenas uma coluna da chave primária pode ser selecionada.
  rds.region-id	ID da região da instância Alibaba Cloud ApsaraDB RDS for MySQL.	Obrigatório ao usar o recurso de leitura de logs arquivados do OSS.	STRING	Nenhum	Para obter a lista de IDs de região, consulte Regiões e zonas. Importante Como a string GTID do MySQL CDC é gerada aleatoriamente e não cresce monotonicamente como os offsets de arquivos de log binário, localizar um GTID exige baixar e analisar todos os logs arquivados no OSS. Esse processo consome muitos recursos e tempo, inviabilizando funcionalidades baseadas em offsets GTID. Assim, o recurso de logs arquivados do OSS suporta apenas início por carimbo de data/hora ou offset de arquivo de log binário. Não há suporte para início por GTID nem para cenários com failover primário/secundário nos logs arquivados, já que tais operações dependem de GTIDs. Avalie cuidadosamente antes de utilizar.
  rds.access-key-id	AccessKey ID da conta Alibaba Cloud ApsaraDB RDS for MySQL.	Obrigatório ao usar o recurso de leitura de logs arquivados do OSS.	STRING	Nenhum	Para mais informações, consulte Como visualizar o AccessKey ID e o AccessKey secret?. Importante Para evitar vazamento das suas credenciais, utilize o recurso de gerenciamento de segredos para especificar o AccessKey ID. Consulte Gerenciar variáveis.
  rds.access-key-secret	AccessKey secret da conta Alibaba Cloud ApsaraDB RDS for MySQL.	Obrigatório ao usar o recurso de leitura de logs arquivados do OSS.	STRING	Nenhum	Para mais informações, consulte Como visualizar o AccessKey ID e o AccessKey secret? Importante Para proteger suas credenciais contra vazamentos, utilize o recurso de gerenciamento de segredos para informar o AccessKey secret. Veja Gerenciar variáveis.
  rds.db-instance-id	ID da instância Alibaba Cloud ApsaraDB RDS for MySQL.	Obrigatório ao usar o recurso de leitura de logs arquivados do OSS.	STRING	Nenhum	Nenhuma.
  rds.main-db-id	Número do banco de dados principal da instância Alibaba Cloud ApsaraDB RDS for MySQL.	Não	STRING	Nenhum	Para saber como obter o número do banco de dados principal, consulte Backup de logs do ApsaraDB RDS for MySQL. Suportado apenas no VVR 8.0.7 e versões posteriores. Nota Se este parâmetro não for especificado, o VVR 11.1 e versões posteriores consultam automaticamente o número do banco de dados principal com base nas informações de conexão do ApsaraDB RDS for MySQL.
  rds.download.timeout	Tempo limite para download de um único log arquivado do OSS.	Não	DURATION	60s	Nenhuma.
  rds.endpoint	Endpoint de serviço para obtenção de informações de log binário do OSS.	Não	STRING	Nenhum	Para verificar os valores válidos, consulte Endpoints. Disponível apenas no VVR 8.0.8 e versões posteriores.
  scan.incremental.close-idle-reader.enabled	Define se leitores ociosos devem ser fechados após a conclusão do snapshot.	Não	BOOLEAN	false	Recurso disponível somente no VVR 8.0.1 e superior. Para que esta configuração funcione, defina execution.checkpointing.checkpoints-after-tasks-finish.enabled como true.
  scan.read-changelog-as-append-only.enabled	Indica se o fluxo de dados de changelog deve ser convertido para append-only.	Não	BOOLEAN	false	Valores aceitos: true: Converte todas as mensagens (INSERT, DELETE, UPDATE_BEFORE e UPDATE_AFTER) em mensagens INSERT. Ative esta opção apenas em cenários específicos, como quando for necessário preservar mensagens de exclusão da tabela upstream. false (padrão): Encaminha todos os tipos de mensagem downstream sem alteração. Nota Compatível apenas com VVR 8.0.8 ou superior.
  scan.only.deserialize.captured.tables.changelog.enabled	Na fase incremental, define se apenas os eventos de alteração das tabelas especificadas devem ser desserializados.	Não	BOOLEAN	O valor padrão é false nas versões VVR 8.x. O valor padrão passa a ser true a partir do VVR 11.1.	Opções disponíveis: true: Desserializa exclusivamente os dados de alteração das tabelas-alvo, acelerando a leitura do log binário. false (padrão): Desserializa os dados de alteração de todas as tabelas. Nota Recurso introduzido no VVR 8.0.7. No VVR 8.0.8 ou anterior, altere o nome do parâmetro para debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	Durante a fase incremental, indica se o sistema deve tentar analisar eventos DDL de alteração sem bloqueio do RDS.	Não	BOOLEAN	false	Valores possíveis: true: Analisa eventos DDL de alteração sem bloqueio do RDS. false (padrão): Ignora a análise de eventos DDL de alteração sem bloqueio do RDS. Trata-se de um recurso experimental. Antes de executar uma alteração online sem bloqueio, crie um snapshot do job Flink para fins de recuperação. Nota Disponível apenas no VVR 11.1 e versões posteriores.
  scan.incremental.snapshot.backfill.skip	Define se o backfill deve ser ignorado durante a fase de leitura de snapshot.	Não	BOOLEAN	false	Valores válidos: true: Ignora o backfill na fase de leitura de snapshot. false (padrão): Executa o backfill normalmente durante a leitura de snapshot. Ao pular o backfill, as alterações ocorridas na tabela durante a fase de snapshot são lidas posteriormente na fase incremental, em vez de serem mescladas ao snapshot. Importante Ignorar o backfill pode resultar em inconsistência de dados, pois alterações da fase de snapshot podem ser reprocessadas. Apenas a semântica at-least-once é garantida. Nota Suportado somente a partir do VVR 11.1.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Determina se chunks ilimitados devem ser despachados primeiro na fase de leitura de snapshot.	Não	BOOELEAN	false	Opções: true: Prioriza o despacho de chunks ilimitados na leitura de snapshot. false (padrão): Não prioriza chunks ilimitados na fase de snapshot. Recurso experimental. Sua ativação pode reduzir o risco de erros OOM no TaskManager ao sincronizar o último chunk durante o snapshot. Adicione este parâmetro antes da primeira execução do job. Nota Disponível apenas no VVR 11.1 e superior.
  binlog.session.network.timeout	Timeout de leitura/gravação de rede para a conexão do log binário.	Não	DURATION	10m	Se definido como 0s, utiliza o timeout padrão do servidor MySQL. Nota Compatível apenas com VVR 11.5 ou posterior.
  scan.rate-limit.records-per-second	Limita o número máximo de registros enviados pela source por segundo.	Não	LONG	Nenhum	Aplica-se a cenários que exigem limitação na leitura de dados. O limite vigora tanto na fase completa quanto na incremental. A métrica numRecordsOutPerSecond da source reflete a quantidade de registros emitidos por todo o fluxo de dados por segundo. Utilize essa métrica como base para ajustar este parâmetro. Na fase de leitura completa, geralmente é necessário reduzir o número de linhas lidas por lote. Diminua o valor do parâmetro scan.incremental.snapshot.chunk.size. Nota Recurso disponível apenas no VVR 11.5 e versões posteriores.
  scan.binlog.tolerate.gtid-holes	Ativar este parâmetro faz com que lacunas na sequência GTID sejam ignoradas, permitindo que o job contorne eventos descontínuos e continue executando.	Não	BOOLEAN	false	Antes de ativar este parâmetro, certifique-se de que o offset inicial do job não expirou. Se o job iniciar a partir de um offset GTID limpo ou expirado, o engine ignorará silenciosamente os logs ausentes, causando perda de dados. Nota Parâmetro suportado apenas no VVR 11.6 e versões posteriores.

• Parâmetros específicos para tabela de dimensão

  Parâmetro	Descrição	Obrigatório	Tipo de dados	Valor padrão	Observações
  url	URL JDBC do MySQL.	Não	STRING	Nenhum	Formato da URL: jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	Número máximo de tentativas após falha na leitura de dados.	Não	INTEGER	3	Suportado apenas no VVR 6.0.7 e versões posteriores.
  lookup.cache.strategy	Política de cache.	Não	STRING	Nenhum	As políticas de cache suportadas são None, LRU e ALL. Para mais informações sobre os valores, consulte Instruções JOIN de tabela de dimensão. Nota Ao usar a política de cache LRU, também é necessário configurar o parâmetro lookup.cache.max-rows.
  lookup.cache.max-rows	Número máximo de linhas em cache.	Não	INTEGER	100000	Se você selecionar a política de cache LRU, defina o tamanho do cache. Caso opte pela política de cache ALL, não é preciso definir o tamanho do cache.
  lookup.cache.ttl	Tempo de vida (TTL) do cache.	Não	DURATION	10 s	A configuração de lookup.cache.ttl depende de lookup.cache.strategy: Quando lookup.cache.strategy estiver definido como None, não configure lookup.cache.ttl. Isso significa que o cache não expira. Se lookup.cache.strategy for LRU, lookup.cache.ttl representa o TTL do cache. Por padrão, o cache não expira. Com lookup.cache.strategy definido como ALL, lookup.cache.ttl indica o tempo de carregamento do cache. Por padrão, o cache não é recarregado. Utilize um formato de tempo, como 1min ou 10s.
  lookup.max-join-rows	Quantidade máxima de resultados retornados quando um registro da tabela principal corresponde a registros na tabela de dimensão.	Não	INTEGER	1024	Nenhuma.
  lookup.filter-push-down.enabled	Define se o pushdown de filtro deve ser ativado para a tabela de dimensão.	Não	BOOLEAN	false	Valores válidos: true: Ativa o pushdown de filtro para a tabela de dimensão. Ao carregar dados da tabela do banco de dados MySQL, a tabela de dimensão filtra os dados antecipadamente com base nas condições definidas no job SQL. false (padrão): Desativa o pushdown de filtro para a tabela de dimensão. Durante o carregamento de dados da tabela do banco de dados MySQL, a tabela de dimensão carrega todos os dados. Nota Suportado apenas no VVR 8.0.7 e versões posteriores. Importante O pushdown de tabela de dimensão deve ser ativado somente quando uma tabela Flink for usada como tabela de dimensão. Tabelas de origem MySQL não suportam a ativação de pushdown de filtro. Se uma tabela Flink funcionar simultaneamente como tabela de origem e de dimensão, e o pushdown de filtro estiver ativado para a tabela de dimensão, defina explicitamente este item de configuração como false para a tabela de origem usando SQL Hints. Caso contrário, o job pode apresentar execução anormal.

• Apenas para tabelas sink

  Parâmetro	Descrição	Obrigatório	Tipo de dados	Valor padrão	Observações
  url	URL JDBC do MySQL.	Não	STRING	Nenhum	Formato da URL: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	Número máximo de tentativas após falha na gravação de dados.	Não	INTEGER	3	Nenhuma.
  sink.buffer-flush.batch-size	Quantidade de linhas em uma única gravação em lote.	Não	INTEGER	4096	Nenhuma.
  sink.buffer-flush.max-rows	Número de linhas de dados armazenadas em cache na memória.	Não	INTEGER	10000	Este parâmetro só entra em vigor após a especificação de uma chave primária.
  sink.buffer-flush.interval	Intervalo para liberação do cache. Se os dados no cache não atenderem às condições de saída após o tempo de espera especificado, o sistema libera automaticamente todos os dados do cache.	Não	DURATION	1s	Nenhuma.
  sink.ignore-delete	Define se as operações DELETE de dados devem ser ignoradas.	Não	BOOLEAN	false	Quando o stream gerado pelo Flink SQL inclui registros delete ou update-before, podem ocorrer inconsistências de dados se múltiplas tarefas de saída atualizarem campos diferentes da mesma tabela simultaneamente. Por exemplo, após a exclusão de um registro, outra tarefa atualiza apenas alguns campos. Os campos não atualizados assumirão valores nulos ou padrão, causando erros nos dados. Definir sink.ignore-delete como true permite ignorar operações DELETE e UPDATE_BEFORE upstream para evitar esses problemas. Nota UPDATE_BEFORE faz parte do mecanismo de retração do Flink, usado para "retrair" o valor antigo em uma operação de atualização. Com ignoreDelete = true, todos os registros do tipo DELETE e UPDATE_BEFORE são ignorados. Apenas registros INSERT e UPDATE_AFTER são processados.
  sink.ignore-null-when-update	Durante a atualização de dados, define se o campo correspondente deve ser atualizado para null ou se a atualização desse campo deve ser ignorada caso o valor do campo de dados recebido seja null.	Não	BOOLEAN	false	Valores válidos: true: Não atualiza o campo. Este parâmetro só pode ser definido como true quando uma chave primária estiver configurada para a tabela Flink. Quando definido como true: No VVR 8.0.6 e versões anteriores, a tabela sink não suporta gravação em lote. No VVR 8.0.7 e versões posteriores, a tabela sink suporta gravação em lote. A gravação em lote melhora significativamente a eficiência de escrita e o throughput geral, mas introduz latência de dados e risco de erros OOM. Portanto, avalie o custo-benefício com base no seu cenário de negócios. false: Atualiza o campo para null. Nota Este parâmetro é suportado apenas no VVR 8.0.5 e versões posteriores.

Mapeamento de tipos

• Tabelas source CDC

  Tipo de campo MySQL CDC	Tipo de campo Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  TINYINT UNSIGNED ZEROFILL
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  SMALLINT UNSIGNED ZEROFILL
  BIGINT	BIGINT
  INT UNSIGNED
  INT UNSIGNED ZEROFILL
  MEDIUMINT UNSIGNED
  MEDIUMINT UNSIGNED ZEROFILL
  BIGINT UNSIGNED	DECIMAL(20, 0)
  BIGINT UNSIGNED ZEROFILL
  SERIAL
  FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
  DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
  DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
  REAL [UNSIGNED] [ZEROFILL]
  NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
  DECIMAL(p, s) [UNSIGNED] [ZEROFILL]
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]	TIMESTAMP [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  Importante

  Não utilize o tipo TINYINT(1) no MySQL para armazenar valores diferentes de 0 e 1. Quando property-version=0, a tabela source MySQL CDC mapeia TINYINT(1) para o tipo BOOLEAN no Flink por padrão, o que pode causar imprecisões nos dados. Para utilizar o tipo TINYINT(1) com valores distintos de 0 e 1, consulte o parâmetro de configuração catalog.table.treat-tinyint1-as-boolean.

• Tabelas de dimensão e tabelas sink

  Tipo de campo MySQL	Tipo de campo Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) Nota onde p <= 38.
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  CHAR(n)	CHAR(n)
  VARCHAR(n)	VARCHAR(n)
  BIT(n)	BINARY(⌈n/8⌉)
  BINARY(n)	BINARY(n)
  VARBINARY(N)	VARBINARY(N)
  TINYTEXT	STRING
  TEXT
  MEDIUMTEXT
  LONGTEXT
  TINYBLOB	BYTES Importante O Flink suporta apenas registros do tipo BLOB do MySQL com tamanho igual ou inferior a 2.147.483.647 (2^31 - 1) bytes.
  BLOB
  MEDIUMBLOB
  LONGBLOB

Ingestão de dados

Utilize o conector MySQL como source de dados em um job YAML de ingestão de dados.

Sintaxe

source:
   type: mysql
   name: MySQL Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: xxx

Itens de configuração

Reutilizar um catálogo existente

A partir do VVR 11.5, é possível referenciar diretamente um catálogo MySQL integrado criado na página Data Management em um job de ingestão de dados do Flink CDC. Isso reduz o esforço manual na escrita das propriedades de conexão.

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog

Atualmente, os jobs de ingestão de dados suportam a reutilização automática dos seguintes parâmetros de catálogo MySQL:

• hostname

• port

• username

• password

• catalog.table.metadata-columns

• catalog.table.treat-tinyint1-as-boolean

Para substituir qualquer um desses parâmetros reutilizados automaticamente, escreva explicitamente o parâmetro YAML correspondente. O parâmetro escrito explicitamente tem prioridade maior.

Mapeamento de tipos

A tabela a seguir apresenta o mapeamento de tipos de dados para ingestão de dados.

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium configuration
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Enable parsing filter
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Parse only the change events of specified tables.
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium configuration
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Enable parsing filter
  scan.only.deserialize.captured.tables.changelog.enabled: true

Crie um programa de API DataStream e utilize o MySqlSource. Os exemplos de código e dependências do pom estão disponíveis abaixo:

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // set captured database
        .tableList("yourDatabaseName.yourTableName") // set captured table
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // enable checkpoint
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
    env.execute("Print MySQL Snapshot + Binlog");
  }
}

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>

Ao compilar o MySqlSource, especifique os seguintes parâmetros no código:

As dependências do pom devem especificar os seguintes parâmetros: