Realtime Compute for Apache Flink:MySQL SQL connector

O conector MySQL oferece suporte aos itens a seguir.

• Processamento unificado de batch e stream, com suporte à leitura de dados completos e incrementais, eliminando a necessidade de manter dois processos separados.

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

• Alternância transparente da leitura de dados completos para a leitura incremental, com redução automática de escala para economizar recursos computacionais.

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

• Leitura de dados completos sem bloqueio, sem impacto nos serviços em produção.

• Suporte à leitura de logs de backup do ApsaraDB RDS for MySQL.

• Análise paralela de arquivos de binary log para redução da latência de leitura.

SQL

Você pode usar o conector MySQL em jobs SQL como tabela source, tabela de dimensão ou tabela sink.

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

  Parameter	Description	Required	Data type	Default value	Notes
  connector	O tipo da tabela.	Yes	STRING	None	Quando usada como tabela source, defina este parâmetro como mysql-cdc ou mysql. Os dois valores são equivalentes. Quando usada como tabela de dimensão ou tabela de destino, o valor deve ser mysql.
  hostname	O endereço IP ou hostname do banco de dados MySQL.	Yes	STRING	None	Recomendamos especificar um endereço de 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 utilize um endpoint público para acessar o banco de dados. Para mais informações, consulte Manage and operate workspaces e How can a fully managed Flink cluster access the Internet?.
  username	O nome de usuário do service de banco de dados MySQL.	Yes	STRING	None	Nenhuma observação.
  password	A senha do service de banco de dados MySQL.	Yes	STRING	None	Nenhuma observação.
  database-name	O nome do banco de dados MySQL.	Yes	STRING	None	Quando um banco de dados é usado como tabela source, é possível utilizar uma expressão regular no nome do banco de dados para ler dados de múltiplos bancos. Ao usar expressões regulares, não utilize os símbolos ^ e $ para corresponder ao início e ao fim da string. Para mais informações, consulte as observações do parâmetro table-name.
  table-name	O nome da tabela MySQL.	Yes	STRING	None	É possível usar uma expressão regular no nome da tabela source para ler dados de múltiplas tabelas. Ao usar expressões regulares, não utilize os símbolos ^ e $ para corresponder ao início e ao fim da string. Para mais informações, consulte a observação a seguir. Nota Quando uma tabela source CDC do MySQL faz a correspondência de nomes de tabela usando uma expressão regular, o conector concatena os valores de database-name e table-name especificados com a string \\. para formar uma expressão regular de caminho completo. Antes do VVR 8.0.1, o caractere . era utilizado. O conector usa essa expressão regular para corresponder aos nomes completamente qualificados das tabelas no banco de dados MySQL. Por exemplo, ao definir 'database-name'='db_.' e 'table-name'='tb_.+', o conector aplica a expressão regular db_.\\.tb_.+ para corresponder aos nomes completamente qualificados e determinar quais tabelas serão lidas. Antes do VVR 8.0.1, a expressão regular era db_.*.tb_.+.
  port	O número da porta do service de banco de dados MySQL.	No	INTEGER	3306	Nenhuma observação.

• Somente tabela source

  Parâmetro	Descrição	Obrigatório	Tipo de dado	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 automaticamente.	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 o formato de intervalo de IDs, como 5400-5408. Quando a leitura incremental está ativa, a leitura simultânea é suportada. Nesse caso, defina um intervalo de IDs para que cada leitor simultâneo utilize um ID diferente. Para mais informações, consulte Use Server ID.
  scan.incremental.snapshot.enabled	Indica se os snapshots incrementais devem ser ativados.	Não	BOOLEAN	true	Os snapshots incrementais são ativados por padrão. Trata-se de um novo mecanismo para leitura de snapshots completos de dados. Em comparação com o método antigo, os snapshots incrementais oferecem diversas vantagens, entre elas: O source consegue ler dados completos em paralelo. O source suporta checkpoints em nível de chunk durante a leitura de dados completos. O source não precisa adquirir um bloqueio de leitura global (FLUSH TABLES WITH read lock) durante a leitura de dados completos. Para que o source suporte leitura simultânea, cada leitor simultâneo precisa de um server ID único. Por isso, o server-id deve ser um intervalo, como 5400-6400, e o tamanho do intervalo deve ser 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	O tamanho de cada chunk em número de linhas.	Não	INTEGER	8096	Quando a leitura incremental de snapshot está ativa, a tabela é dividida em múltiplos chunks para leitura. Os dados de um chunk são armazenados em memória antes de serem lidos por completo. Quanto menos linhas cada chunk contiver, maior será o número total de chunks na tabela. Embora isso reduza a granularidade da recuperação de falhas, pode ocasionar erros de out-of-memory (OOM) e diminuir o throughput geral. Por isso, é necessário encontrar um equilíbrio e definir um tamanho de chunk adequado.
  scan.snapshot.fetch.size	O número máximo de registros a buscar de uma vez durante a leitura de dados completos de uma tabela.	Não	INTEGER	1024	Nenhuma.
  scan.startup.mode	O modo de inicialização para consumo de dados.	Não	STRING	initial	Valores válidos: initial (padrão): Na primeira inicialização ou em uma inicialização sem estado, o conector lê todos os dados históricos completos e, em seguida, passa a ler os dados mais recentes do binary log. latest-offset: Na primeira inicialização ou em uma inicialização sem estado, o conector não lê os dados históricos. A leitura começa a partir do final do binary log, ou seja, somente as alterações mais recentes feitas após o início do conector são lidas. earliest-offset: O conector não lê os dados históricos. A leitura começa a partir do binary log disponível mais antigo. specific-offset: O conector não lê os dados históricos. A leitura começa a partir de um offset específico do binary log. Especifique o offset configurando tanto scan.startup.specific-offset.file quanto scan.startup.specific-offset.pos, ou configure apenas scan.startup.specific-offset.gtid-set para iniciar a partir de um conjunto de GTIDs específico. timestamp: O conector não lê os dados históricos. A leitura do binary log começa a partir de um timestamp especificado, definido em milissegundos pelo parâmetro scan.startup.timestamp-millis. Importante Ao utilizar o modo de inicialização earliest-offset, specific-offset ou timestamp, certifique-se de que o schema da tabela correspondente não sofreu alterações entre a posição de consumo do binary log especificada e o momento de inicialização do job. Isso evita erros causados por incompatibilidades de schema.
  scan.startup.specific-offset.file	O nome do arquivo de binary log correspondente ao offset de início, usado no modo de inicialização specific-offset.	Não	STRING	Nenhum	Ao usar este parâmetro, defina scan.startup.mode como specific-offset. Exemplo de formato de nome de arquivo: mysql-bin.000003.
  scan.startup.specific-offset.pos	O offset dentro do arquivo de binary log especificado, correspondente ao offset de início no modo de inicialização specific-offset.	Não	INTEGER	Nenhum	Ao usar este parâmetro, defina scan.startup.mode como specific-offset.
  scan.startup.specific-offset.gtid-set	O conjunto de GTIDs correspondente ao offset de início, usado no modo de inicialização specific-offset.	Não	STRING	Nenhum	Ao usar este parâmetro, defina scan.startup.mode como specific-offset. Exemplo de formato de conjunto de GTIDs: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	O timestamp em milissegundos correspondente ao offset de início, usado no modo de inicialização timestamp.	Não	LONG	Nenhum	Ao usar este parâmetro, defina scan.startup.mode como timestamp. A unidade do timestamp é milissegundos. Importante Ao especificar um horário, o MySQL CDC tenta ler o evento inicial de cada arquivo de binary log para determinar seu timestamp e, em seguida, localiza o arquivo correspondente ao horário especificado. Certifique-se de que o arquivo de binary log correspondente ao timestamp indicado não foi removido do banco de dados e ainda pode ser lido.
  server-time-zone	O fuso horário de sessão utilizado pelo banco de dados.	Não	STRING	Se este parâmetro não for especificado, o sistema utiliza o fuso horário do ambiente de execução do job Flink como fuso horário do servidor de banco de dados. Esse é o fuso horário da zona selecionada.	Exemplo: Asia/Shanghai. Este parâmetro controla como o tipo TIMESTAMP do MySQL é convertido para o tipo STRING. Para mais informações, consulte Debezium temporal values.
  debezium.min.row.count.to.stream.results	Quando o número de linhas em uma tabela for maior que este valor, o modo de leitura em lotes é utilizado.	Não	INTEGER	1000	O Flink lê dados de uma tabela source do MySQL de uma das seguintes formas: Leitura completa: lê todos os dados da tabela diretamente na memória. Este método é rápido, mas consome a quantidade correspondente de memória. Se a tabela source for muito grande, há risco de erros OOM. Leitura em lotes: lê os dados em múltiplos lotes, com um determinado número de linhas por lote, até que todos os dados sejam lidos. Este método evita riscos de OOM na leitura de tabelas grandes, porém é relativamente mais lento.
  connect.timeout	O tempo máximo de espera por uma conexão ao servidor de banco de dados MySQL antes de uma nova tentativa.	Não	DURATION	30s	Nenhuma.
  connect.max-retries	O número máximo de tentativas após uma falha de conexão ao service de banco de dados MySQL.	Não	INTEGER	3	Nenhuma.
  connection.pool.size	O tamanho do pool de conexões do banco de dados.	Não	INTEGER	20	O pool de conexões do banco de dados permite reutilizar conexões existentes, reduzindo o número total de conexões abertas.
  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 não usar o protocolo SSL, configure 'jdbc.properties.useSSL' = 'false'. Para mais informações sobre os parâmetros de conexão suportados, consulte MySQL Configuration Properties.
  debezium.*	Parâmetros personalizados para o Debezium realizar a leitura de binary logs.	Não	STRING	Nenhum	É possível passar parâmetros personalizados para o Debezium. Por exemplo, use 'debezium.event.deserialization.failure.handling.mode'='ignore' para especificar a lógica de tratamento de erros de parsing. Aviso Não modifique parâmetros do Debezium de forma arbitrária, pois isso pode fazer com que o conector leia dados incorretamente. Por exemplo, o parâmetro debezium.binlog.buffer.size não pode ser configurado.
  heartbeat.interval	O intervalo no qual o source avança o offset do binary log por meio de eventos de heartbeat.	Não	DURATION	30s	Os eventos de heartbeat são usados para avançar o offset do binary log no source. Esse recurso é especialmente útil para tabelas do MySQL que são atualizadas com pouca frequência, pois nelas o offset do binary log não avança automaticamente. Os eventos de heartbeat empurram o offset para frente, evitando problemas causados por um offset de binary log expirado — o que pode fazer com que o job falhe de forma irrecuperável e exija uma reinicialização sem estado.
  scan.incremental.snapshot.chunk.key-column	Especifica a coluna a ser usada como coluna de divisão para o sharding durante a fase de snapshot.	Consulte a coluna Observações.	STRING	Nenhum	Obrigatório para tabelas sem chave primária. A coluna selecionada deve ser do tipo não nulo (NOT NULL). Opcional para tabelas com chave primária. Somente uma coluna pode ser selecionada a partir da chave primária.
  rds.region-id	O 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 mais informações sobre IDs de região, consulte Regions and zones. Importante Como a string GTID do MySQL CDC é gerada aleatoriamente e não é monotonicamente crescente como os offsets de arquivo de binary log, localizar um GTID em um arquivo exige baixar e analisar todos os logs arquivados do OSS. Esse processo consome muitos recursos e tempo, tornando inviáveis os recursos que dependem de offsets GTID. Por isso, o recurso de logs arquivados do OSS suporta apenas a inicialização a partir de um timestamp especificado ou de um offset de arquivo de binary log especificado. Ele não suporta a inicialização a partir de um GTID específico, nem cenários com failover primário/secundário nos logs arquivados, pois esses failovers dependem de GTIDs. Avalie cuidadosamente este recurso antes de utilizá-lo.
  rds.access-key-id	O 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 How do I view the AccessKey ID and AccessKey secret?. Importante Para evitar vazamento das informações do seu AccessKey, use o recurso de gerenciamento de segredos para especificar o AccessKey ID. Para mais informações, consulte Manage variables.
  rds.access-key-secret	O 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 How do I view the AccessKey ID and AccessKey secret? Importante Para evitar vazamento das informações do seu AccessKey, use o recurso de gerenciamento de segredos para especificar o AccessKey secret. Para mais informações, consulte Manage variables.
  rds.db-instance-id	O 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	O número do banco de dados primário da instância Alibaba Cloud ApsaraDB RDS for MySQL.	Não	STRING	Nenhum	Para mais informações sobre como obter o número do banco de dados primário, consulte ApsaraDB RDS for MySQL log backup. Suportado apenas no VVR 8.0.7 e versões posteriores. Nota Se este parâmetro não for especificado, o VVR 11.7 e versões posteriores consultam automaticamente o número do banco de dados primário com base nas informações de conexão do ApsaraDB RDS for MySQL.
  rds.download.timeout	O período de timeout para baixar um único log arquivado do OSS.	Não	DURATION	60s	Nenhuma.
  rds.endpoint	O endpoint de service para obtenção de informações de binary log do OSS.	Não	STRING	Nenhum	Para mais informações sobre os valores válidos, consulte Endpoints. Suportado apenas no VVR 8.0.8 e versões posteriores.
  scan.incremental.close-idle-reader.enabled	Indica se os leitores ociosos devem ser encerrados após a conclusão do snapshot.	Não	BOOLEAN	false	Suportado apenas no VVR 8.0.1 e versões posteriores. Para que esta configuração tenha efeito, 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 um fluxo de dados append-only.	Não	BOOLEAN	false	Valores válidos: true: Todos os tipos de mensagens, incluindo INSERT, DELETE, UPDATE_BEFORE e UPDATE_AFTER, são convertidos em mensagens INSERT. Ative esta opção apenas em cenários especiais, como quando é necessário salvar mensagens de exclusão da tabela upstream. false (padrão): Todos os tipos de mensagens são enviados para downstream sem alteração. Nota Suportado apenas no VVR 8.0.8 e versões posteriores.
  scan.only.deserialize.captured.tables.changelog.enabled	Na fase incremental, indica 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 é true no VVR 11.1 e versões posteriores.	Valores válidos: true: Desserializa apenas os dados de alteração das tabelas de destino para acelerar a leitura do binary log. false (padrão): Desserializa os dados de alteração de todas as tabelas. Nota Suportado apenas no VVR 8.0.7 e versões posteriores. Ao usar o VVR 8.0.8 ou versões anteriores, altere o nome do parâmetro para debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	Na fase incremental, indica se deve haver tentativa de análise dos eventos DDL de alteração sem bloqueio do RDS.	Não	BOOLEAN	false	Valores válidos: true: Analisa os eventos DDL de alteração sem bloqueio do RDS. false (padrão): Não analisa os eventos DDL de alteração sem bloqueio do RDS. Este é um recurso experimental. Antes de executar uma alteração online sem bloqueio, tire um snapshot do job Flink para fins de recuperação. Nota Suportado apenas no VVR 11.1 e versões posteriores.
  scan.incremental.snapshot.backfill.skip	Indica se o backfill deve ser ignorado durante a fase de leitura de snapshot.	Não	BOOLEAN	false	Valores válidos: true: Ignora o backfill durante a fase de leitura de snapshot. false (padrão): Não ignora o backfill durante a fase de leitura de snapshot. O backfill se aplica apenas durante a consulta de snapshot de um único chunk e não abrange toda a fase de leitura completa. Quando o backfill é ignorado, a consulta de snapshot de cada chunk lê os dados mais recentes da tabela naquele instante; atualizações que ocorrem em um chunk após sua leitura não são mescladas durante a fase de leitura completa e são lidas a partir do Binlog ao entrar na fase incremental. Por exemplo, uma atualização no chunk5 que ocorre enquanto ele está sendo capturado no snapshot é refletida diretamente no snapshot do chunk5; se o chunk5 for atualizado depois que o leitor já avançou para o chunk80, a atualização será aplicada posteriormente a partir do Binlog durante a fase incremental. Importante Quando ativado, as alterações que ocorrem durante ou após a varredura de um chunk ainda são entregues pelo Binlog na fase incremental e podem ser duplicadas. Apenas a semântica de at-least-once é garantida. Ative esta opção somente quando o sink downstream suportar escritas idempotentes por chave primária. Nota Suportado apenas no VVR 11.1 e versões posteriores.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Indica se os chunks ilimitados devem ser despachados primeiro durante a fase de leitura de snapshot.	Não	BOOELEAN	false	Valores válidos: true: Despacha os chunks ilimitados primeiro durante a fase de leitura de snapshot. false (padrão): Não despacha os chunks ilimitados primeiro durante a fase de leitura de snapshot. Este é um recurso experimental. Ativá-lo pode reduzir o risco de erros OOM no TaskManager durante a sincronização do último chunk na fase de snapshot. Adicione este parâmetro antes da primeira inicialização do job. Nota Suportado apenas no VVR 11.1 e versões posteriores.
  binlog.session.network.timeout	O timeout de leitura/gravação de rede para a conexão de binary log.	Não	DURATION	10m	Se definido como 0s, o timeout padrão do servidor MySQL é utilizado. Nota Suportado apenas no VVR 11.5 e versões posteriores.
  scan.rate-limit.records-per-second	Limita o número máximo de registros enviados pelo source por segundo.	Não	LONG	Nenhum	Aplicável a cenários onde é necessário limitar a taxa de leitura de dados. Este limite é efetivo tanto na fase completa quanto na fase incremental. A métrica numRecordsOutPerSecond do source reflete o número de registros produzidos por todo o fluxo de dados por segundo. Ajuste este parâmetro com base nessa métrica. Na fase de leitura de dados completos, geralmente é necessário reduzir o número de linhas lidas em cada lote. Para isso, reduza o valor do parâmetro scan.incremental.snapshot.chunk.size. Nota Suportado apenas no VVR 11.5 e versões posteriores.
  scan.binlog.tolerate.gtid-holes	Ao ativar este parâmetro, as lacunas na sequência GTID são ignoradas, permitindo que o job contorne eventos descontínuos e continue em execução.	Não	BOOLEAN	false	Antes de ativar este parâmetro, certifique-se de que o offset de início do job não expirou. Se o job iniciar a partir de um offset GTID já removido ou expirado, o motor silenciosamente ignorará os logs ausentes, o que resultará em perda de dados. Nota Este parâmetro é suportado apenas no VVR 11.6 e versões posteriores.
  scan.filter-unchanged-row.enabled	Indica se os eventos de alteração de campos não referenciados no SQL devem ser filtrados.	Não	BOOLEAN	false	Ao definir este parâmetro como true, uma tabela source CDC compara apenas as colunas referenciadas na instrução SQL ao receber um evento UPDATE. Se os valores das colunas referenciadas não mudarem, o evento não é enviado para downstream. Este parâmetro é efetivo em cenários de JOIN com múltiplos fluxos. Se uma tabela upstream atualizar um campo não referenciado na instrução SQL, eventos DELETE e INSERT desnecessários não são gerados. Isso reduz a amplificação de dados e a oscilação de resultados nos operadores downstream. Nota Este parâmetro é suportado apenas no VVR 11.8 e versões posteriores.
  scan.projection-pushdown.enabled	Indica se o pushdown de projeção deve ser ativado.	Não	BOOLEAN	false	Ao definir este parâmetro como true, uma tabela source CDC analisa e transmite apenas as colunas referenciadas na instrução SQL. As colunas não referenciadas não são serializadas nem transmitidas pela rede, reduzindo a sobrecarga de I/O e processamento. Este parâmetro é mais eficaz quando usado em conjunto com scan.filter-unchanged-row.enabled: o pushdown de projeção primeiro reduz o volume de dados, e em seguida as linhas sem alteração são filtradas, aliviando ainda mais a carga nos operadores downstream. Nota Este parâmetro é suportado apenas no VVR 11.8 e versões posteriores.

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

  Parâmetro	Descrição	Obrigatório	Tipo de dado	Valor padrão	Observações
  url	A URL JDBC do MySQL.	Não	STRING	Nenhum	O formato da URL é: jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	O número máximo de tentativas após uma falha na leitura de dados.	Não	INTEGER	3	Compatível apenas com VVR 6.0.7 e versões posteriores.
  lookup.cache.strategy	A 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 Dimension table JOIN statements. Nota Ao usar a política de cache LRU, você também deve configurar o parâmetro lookup.cache.max-rows.
  lookup.cache.max-rows	O número máximo de linhas em cache.	Não	INTEGER	100000	Se a política de cache LRU for selecionada, defina o tamanho do cache. Se a política de cache ALL for selecionada, não é necessário definir o tamanho do cache.
  lookup.cache.ttl	O tempo de vida (TTL) do cache.	Não	DURATION	10 s	A configuração de lookup.cache.ttl depende de lookup.cache.strategy: Se lookup.cache.strategy estiver definido como None, não é necessário configurar lookup.cache.ttl. Isso significa que o cache não expira. Se lookup.cache.strategy estiver definido como LRU, lookup.cache.ttl representa o TTL do cache. Por padrão, o cache não expira. Se lookup.cache.strategy estiver definido como ALL, lookup.cache.ttl representa o tempo de carregamento do cache. Por padrão, o cache não é recarregado. Use um formato de tempo, como 1min ou 10s.
  lookup.max-join-rows	O número máximo 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	Especifica se o filter pushdown deve ser ativado para a tabela de dimensão.	Não	BOOLEAN	false	Valores válidos: true: Ativa o filter pushdown 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 filter pushdown para a tabela de dimensão. Ao carregar dados da tabela do banco de dados MySQL, a tabela de dimensão carrega todos os dados. Nota Compatível apenas com 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 source do MySQL não suportam a ativação de filter pushdown. Se uma tabela Flink for usada tanto como tabela source quanto como tabela de dimensão e o filter pushdown estiver ativado para a tabela de dimensão, você deve definir explicitamente este item de configuração como false para a tabela source usando SQL Hints. Caso contrário, o job pode apresentar comportamento anormal.
  lookup.cache.cache-empty	Especifica se resultados de consulta vazios devem ser armazenados em cache.	Não	BOOLEAN	true	Aplica-se apenas quando a política de cache for LRU. Valores válidos: true (padrão): se uma consulta não retornar dados, um resultado vazio é armazenado em cache e retornado diretamente nas consultas subsequentes. false: se uma consulta não retornar dados, nenhum resultado é armazenado em cache e a próxima consulta busca os dados novamente. Nota Este parâmetro é compatível apenas com VVR 11.9.0 e versões posteriores.

• Apenas para tabelas sink

  Parâmetro	Descrição	Obrigatório	Tipo de dado	Valor padrão	Observações
  url	A URL JDBC do MySQL.	Não	STRING	Nenhum	O formato da URL é: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	O número máximo de tentativas após uma falha na gravação de dados.	Não	INTEGER	3	Nenhuma.
  sink.buffer-flush.batch-size	O número de linhas em uma única gravação em lote.	Não	INTEGER	4096	Nenhuma.
  sink.buffer-flush.max-rows	O 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	O intervalo para liberação do cache. Se os dados em cache não atenderem às condições de saída após o tempo de espera especificado, o sistema grava automaticamente todos os dados do cache.	Não	DURATION	1s	Nenhuma.
  sink.ignore-delete	Especifica se as operações DELETE de dados devem ser ignoradas.	Não	BOOLEAN	false	Quando o stream gerado pelo Flink SQL inclui registros de delete ou update-before, se múltiplas tarefas de saída atualizam campos diferentes da mesma tabela simultaneamente, pode ocorrer inconsistência de dados. Por exemplo, após a exclusão de um registro, outra tarefa atualiza apenas alguns campos. Os campos não atualizados passam a ser nulos ou assumem valores padrão, causando erros nos dados. Ao definir sink.ignore-delete como true, você ignora as 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 "retratar" o valor antigo em uma operação de atualização. Quando ignoreDelete = true, todos os registros do tipo DELETE e UPDATE_BEFORE são ignorados. Apenas os registros INSERT e UPDATE_AFTER são processados.
  sink.ignore-delete-mode	A estratégia para tratar registros do tipo delete após as operações DELETE serem ignoradas.	Não	STRING	ALL	Valores válidos: ALL: Ignora os registros -D e -U. REAL_DELETE: Ignora apenas os registros -D. UPDATE_BEFORE: Ignora apenas os registros -U. Nota Esta opção é compatível apenas com o engine Realtime Compute VVR 11.8 e versões posteriores. Válido somente quando sink.ignore-delete=true. Configurar este parâmetro isoladamente resulta em erro.
  sink.ignore-null-when-update	Ao atualizar dados, especifica se o campo correspondente deve ser atualizado para nulo ou se a atualização desse campo deve ser ignorada quando o valor do campo de entrada for nulo.	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: Para VVR 8.0.6 e versões anteriores, a tabela sink não suporta gravação em lote. Para VVR 8.0.7 e versões posteriores, a tabela sink suporta gravação em lote. A gravação em lote pode melhorar significativamente a eficiência de gravação e o throughput geral, mas introduz latência nos dados e risco de erros de OOM. Portanto, avalie essa opção com base no seu cenário de negócios. false: Atualiza o campo para nulo. Nota Este parâmetro é compatível apenas com VVR 8.0.5 e versões posteriores.
  sink.force-batch-on-non-primary-table	Especifica se os registros devem ser obrigatoriamente armazenados em buffer e inseridos em lotes via JDBC Batch ao gravar dados em uma tabela sink do MySQL que não possui chave primária.	Não	BOOLEAN	false	Este parâmetro só entra em vigor em tabelas sem chave primária. Ao definir este parâmetro como true, os eventos DELETE e UPDATE_BEFORE são descartados silenciosamente, e os eventos INSERT e UPDATE_AFTER são gravados em lote como instruções INSERT. Recomenda-se ativar este parâmetro somente se os dados forem apenas de inserção (append-only) ou se a semântica descrita for aceitável para o seu caso. Nota Este parâmetro é compatível apenas com VVR 11.8 e versões posteriores. Ative este parâmetro somente se os dados forem apenas de inserção (append-only) ou se a semântica descrita for aceitável para o seu caso. Caso contrário, os dados podem ficar inconsistentes.

Mapeamento de tipos

• Tabelas source CDC

  MySQL CDC field type	Flink field type
  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 CDC do MySQL mapeia TINYINT(1) para o tipo BOOLEAN no Flink por padrão. Isso pode causar imprecisões nos dados. Para usar o tipo TINYINT(1) para armazenar valores diferentes de 0 e 1, consulte o parâmetro de configuração catalog.table.treat-tinyint1-as-boolean.

• Tabelas de dimensão e tabelas sink

  MySQL field type	Flink field type
  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 where 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 Flink only supports MySQL BLOB type records that are less than or equal to 2,147,483,647 (2^31 - 1) bytes.
  BLOB
  MEDIUMBLOB
  LONGBLOB

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

SET 'table.optimizer.source-merge.enabled' = 'true';

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    'rds.region-id' = 'cn-beijing',
    'rds.access-key-id' = 'your_access_key_id',
    'rds.access-key-secret' = 'your_access_key_secret',
    'rds.db-instance-id' = 'rm-xxxxxxxx',  // The ID of the database instance.
    'rds.main-db-id' = '12345678', // The ID of the primary database.
    'rds.endpoint' = 'rds.aliyuncs.com'
    ...
)