AnalyticDB for MySQL V3.0 connector - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

O conector do AnalyticDB for MySQL V3.0 permite ler e gravar em clusters AnalyticDB for MySQL V3.0 usando o Flink SQL. O AnalyticDB for MySQL é um service de data warehousing nativo de cloud que oferece suporte a gravações em tempo real com alto throughput, análises de baixa latência e operações complexas de extração, transformação e carga (ETL).

O conector oferece suporte aos seguintes tipos de tabela e recursos:

Item	Descrição
Tipos de tabela	Tabela source, tabela de dimensão e tabela sink. Nota As tabelas source exigem o Ververica Runtime (VVR) 8.0.4 ou posterior. Para obter os parâmetros da tabela source, consulte Use Flink to subscribe to binary logs.
Modos de execução	Modo streaming e modo em lote
Formato de dados	N/D
Métricas	N/D
Tipo de API	API SQL
Atualize e exclua dados em tabelas sink	Suportado

Pré-requisitos

Antes de começar, certifique-se de ter:

Um cluster e uma tabela do AnalyticDB for MySQL. Consulte Crie um cluster e CREATE TABLE
Uma lista de permissões de endereços IP configurada para o cluster. Consulte Configure an IP address whitelist

Sintaxe

CREATE TEMPORARY TABLE adb_table (
  `id` INT,
  `num` BIGINT,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'adb3.0',
  'url' = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName' = '<yourUsername>',
  'password' = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

Importante

A chave primária definida na instrução DDL do Flink deve corresponder à chave primária da tabela física no AnalyticDB for MySQL: mesmos campos, mesmos nomes, presentes em ambas. Chaves primárias divergentes podem causar corrupção de dados.

Comportamentos principais

Chave primária e modo de gravação

Quando a DDL define uma chave primária, o conector opera no modo upsert, em que chaves primárias duplicadas acionam uma atualização em vez de uma nova inserção. O comportamento exato do SQL depende do parâmetro replaceMode:

replace — usa REPLACE INTO. Uma chave primária duplicada sobrescreve toda a linha existente.
upsert — usa INSERT INTO ... ON DUPLICATE KEY UPDATE. Apenas os campos especificados são atualizados; os demais campos mantêm seus valores atuais.
insert — usa INSERT IGNORE INTO. O sistema ignora silenciosamente as chaves primárias duplicadas e mantém a linha existente.

Quando não há uma chave primária definida, o conector sempre usa INSERT IGNORE INTO.

O replaceMode exige o AnalyticDB for MySQL V3.1.3.5 ou posterior e o VVR 11.2 ou posterior para os valores de string ( replace , upsert , insert ). As versões anteriores do VVR usam true (equivalente a replace ) e false (equivalente a upsert ). O VVR 11.2 e posteriores permanecem compatíveis com true e false .

Buffer de gravação

O conector armazena registros em buffer na memória e os libera em lotes. A liberação ocorre quando uma das seguintes condições é atendida:

O número de registros em buffer atinge batchSize (padrão: 1.000) ou bufferSize (padrão: 1.000)
O tempo desde a última liberação atinge flushIntervalMs (padrão: 3.000 ms)

Tanto batchSize quanto bufferSize só têm efeito quando há uma chave primária definida.

Cache de tabela de dimensão

O conector oferece suporte a três políticas de cache para tabelas de dimensão. O cache reduz as consultas à tabela física, mas exige memória adicional.

Política de cache	Comportamento	Quando usar
`ALL` (padrão)	Todos os dados da tabela de dimensão são carregados na memória antes do início do job. As consultas subsequentes acessam apenas o cache. Os dados são recarregados após a expiração de `cacheTTLMs`.	A tabela de dimensão é pequena e a ausência de chaves é comum
`LRU`	As linhas acessadas com frequência são armazenadas em cache. Em caso de falta no cache, o conector consulta a tabela física e atualiza o cache. As entradas expiram após `cacheTTLMs`.	A tabela de dimensão é grande, mas o acesso é concentrado em um subconjunto de chaves
`None`	Sem cache. Cada consulta acessa a tabela física diretamente.	Baixo volume de consultas ou requisitos rígidos de atualização de dados

Importante

Com o cache ALL, o conector carrega de forma assíncrona a tabela de dimensão inteira na memória. Você deve aumentar a memória do nó usado para unir tabelas — o aumento necessário é de pelo menos o dobro do tamanho da tabela remota — para evitar erros de falta de memória (OOM). Monitore o uso de memória do nó durante a execução do job.

Parâmetros na cláusula WITH

Parâmetros comuns

Parâmetro	Tipo	Obrigatório	Padrão	Descrição
`connector`	String	Sim	—	Defina como `adb3.0`
`url`	String	Sim	—	URL do Java Database Connectivity (JDBC) do banco de dados, no formato `jdbc:mysql://<endpoint>:<port>/<databaseName>`. Para obter o endpoint e a porta, selecione a seção Network Information da página do seu cluster em AnalyticDB for MySQL console.
`userName`	String	Sim	—	Nome de usuário para acesso ao banco de dados
`password`	String	Sim	—	Senha para acesso ao banco de dados
`tableName`	String	Sim	—	Nome da tabela de destino no banco de dados
`maxRetryTimes`	Integer	Não	10	Número máximo de tentativas em caso de falha de leitura ou gravação

Parâmetros da tabela sink

Parâmetro	Tipo	Obrigatório	Padrão	Descrição
`batchSize`	Integer	Não	1000	Número de registros gravados por lote. Só tem efeito quando há uma chave primária definida.
`bufferSize`	Integer	Não	1000	Número máximo de registros armazenados em buffer na memória antes que uma liberação seja acionada. Só tem efeito quando há uma chave primária definida.
`flushIntervalMs`	Integer	Não	3000	Tempo máximo (em milissegundos) entre as liberações. Quando esse intervalo é atingido, todos os registros em buffer são gravados independentemente do tamanho do buffer.
`ignoreDelete`	Boolean	Não	false	Indica se as operações de exclusão devem ser ignoradas. Defina como `true` para ignorar exclusões; `false` para aplicá-las.
`replaceMode`	String para o VVR 11.2 e posterior. Boolean para versões anteriores ao VVR 11.2.	Não	`replace` para o VVR 11.2 e posterior; true para versões anteriores ao VVR 11.2	Modo de gravação quando há uma chave primária definida. VVR 11.2+: `replace`, `upsert` ou `insert`. Versões anteriores: `true` (igual a `replace`) ou `false` (igual a `upsert`). Exige o AnalyticDB for MySQL V3.1.3.5 ou posterior. Só tem efeito quando a DDL define uma chave primária; caso contrário, o sistema sempre usa `INSERT IGNORE INTO`.
`excludeUpdateColumns`	String	Não	(vazio)	Lista separada por vírgulas de colunas a serem excluídas das atualizações quando `replaceMode` é `upsert` (ou `false`). Em caso de chave primária duplicada: apenas as colunas restantes são atualizadas; as colunas excluídas mantêm seus valores existentes. Em caso de chave primária única: todas as colunas são inseridas. Exemplo: `excludeUpdateColumns='column1,column2'`. O valor deve estar em uma única linha, sem quebras de linha.
`connectionMaxActive`	Integer	Não	40	Número máximo de conexões simultâneas com o banco de dados no pool de threads

Parâmetros da tabela de dimensão

Parâmetro	Tipo	Obrigatório	Padrão	Descrição
`cache`	String	Não	ALL	Política de cache: `None`, `LRU` ou `ALL`. Consulte Cache de tabela de dimensão.
`cacheSize`	Integer	Não	100000	Número máximo de linhas em cache. Obrigatório quando `cache` é `LRU`.
`cacheTTLMs`	Integer	Não	Long.MAX_VALUE	Tempo de vida da entrada de cache em milissegundos. Para `LRU`: as entradas expiram após esse período (padrão: nunca expira). Para `ALL`: o cache completo é recarregado após esse período (padrão: nunca recarrega). Não é usado quando `cache` é `None`.
`maxJoinRows`	Integer	Não	1024	Número máximo de linhas da tabela de dimensão correspondentes por registro de entrada. Se cada registro de entrada corresponder a no máximo n linhas da tabela de dimensão, defina este valor como n para otimizar o desempenho do join no Realtime Compute for Apache Flink.

Mapeamentos de tipos de dados

AnalyticDB for MySQL V3.0	Realtime Compute for Apache Flink
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s) or NUMERIC(p, s)	DECIMAL(p, s)
VARCHAR	STRING
BINARY	BYTES
DATE	DATE
TIME	TIME
DATETIME	TIMESTAMP
TIMESTAMP	TIMESTAMP
POINT	STRING

Exemplos

Tabela sink

O exemplo a seguir lê de uma source datagen e grava os dados em uma tabela sink do AnalyticDB for MySQL.

CREATE TEMPORARY TABLE datagen_source (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE adb_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'adb3.0',
  'url'       = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

INSERT INTO adb_sink
SELECT * FROM datagen_source;

Tabela de dimensão

O exemplo a seguir faz o join de uma source datagen com uma tabela de dimensão do AnalyticDB for MySQL usando um join temporal. Os resultados são gravados em uma sink blackhole.

CREATE TEMPORARY TABLE datagen_source (
  `a`       INT,
  `b`       VARCHAR,
  `c`       STRING,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE adb_dim (
  `a` INT,
  `b` VARCHAR,
  `c` VARCHAR
) WITH (
  'connector' = 'adb3.0',
  'url'       = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  `a` INT,
  `b` VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T
JOIN adb_dim FOR SYSTEM_TIME AS OF T.proctime AS H
  ON T.a = H.a;

Referências

Error: multi-statement be found