Conector JDBC - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Pré-requisitos

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

O banco de dados e a tabela de destino já existem
O JAR do driver JDBC do banco de dados está disponível para upload

Limitações

Leituras limitadas: uma tabela de origem JDBC é uma fonte limitada. A tarefa é concluída após a leitura de todas as linhas. Para capturar dados de alteração em tempo real, use um conector CDC (Change Data Capture) — consulte Criar uma tabela de origem MySQL CDC e Criar uma tabela de origem PostgreSQL CDC (preview público).
Versão do PostgreSQL: a gravação no PostgreSQL requer a versão 9.5 ou posterior, pois a tabela de destino depende da cláusula ON CONFLICT.

Upload do driver JDBC: faça o upload do JAR do driver JDBC como arquivo de dependência antes de executar o job. Drivers comuns:

Banco de dados	Group ID	Artifact ID
MySQL	`mysql`	mysql-connector-java
Oracle	`com.oracle.database.jdbc`	ojdbc8
PostgreSQL	`org.postgresql`	postgresql

Para drivers não listados aqui, verifique a compatibilidade antes de usar.

Comportamento de upsert do MySQL: ao gravar em uma tabela de destino MySQL com chave primária, o conector emite instruções INSERT INTO ... ON DUPLICATE KEY UPDATE ....

Aviso
A inserção de linhas com valores de índice exclusivo duplicados — mesmo quando as chaves primárias são diferentes — sobrescreve as linhas existentes em qualquer tabela física com restrição de índice exclusivo, causando perda de dados.

Criar uma tabela JDBC

CREATE TABLE jdbc_table (
  `id`   BIGINT,
  `name` VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:<db-type>://<host>:<port>/<database>',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

Opções do conector

Geral

Opção	Tipo	Obrigatória	Padrão	Descrição
`connector`	STRING	Sim	—	Defina como `jdbc`.
`url`	STRING	Sim	—	URL JDBC do banco de dados.
`table-name`	STRING	Sim	—	Nome da tabela de leitura ou gravação.
`username`	STRING	Não	—	Nome de usuário do banco de dados. Configure junto com `password`. Ambas as opções aceitam uma referência de credencial KMS no formato `secret://kms.<secret-name>.<key>`.
`password`	STRING	Não	—	Senha do banco de dados.
`kms.endpoint`	STRING	Não	—	O endpoint do gateway compartilhado ou gateway dedicado do KMS, por exemplo, `kms.cn-hangzhou.aliyuncs.com`. Consulte Endpoints do KMS. Obrigatório ao usar credenciais KMS. Para mais detalhes, consulte Integração com gerenciamento de credenciais KMS.
`kms.akless.assume-role.role-name`	STRING	Não	—	O nome da função RAM usada para ler credenciais KMS. A função deve ser referenciada previamente em Segurança. Obrigatório ao usar credenciais KMS. Para mais detalhes, consulte Integração com gerenciamento de credenciais KMS.
`kms.ca`	STRING	Não	—	O caminho do certificado CA do gateway dedicado do KMS. Defina esta opção como um caminho absoluto do OSS no formato `oss://flink-fullymanaged-<workspace ID>/artifacts/namespaces/<namespace name>/<file name>`. Obrigatório ao usar um gateway dedicado do KMS. Para mais detalhes, consulte Integração com gerenciamento de credenciais KMS.

Opções de origem

Opção	Tipo	Obrigatória	Padrão	Descrição
`scan.partition.column`	STRING	Não	—	Coluna usada para dividir os dados em partições. A coluna deve ser NUMERIC ou TIMESTAMP, e o banco de dados deve suportar comparação desse tipo com valores numéricos. Consulte Leitura particionada.
`scan.partition.num`	INTEGER	Não	—	Número de partições.
`scan.partition.lower-bound`	LONG	Não	—	Menor valor da primeira partição.
`scan.partition.upper-bound`	LONG	Não	—	Maior valor da última partição.
`scan.fetch-size`	INTEGER	Não	`0`	Linhas buscadas por ida e volta ao banco de dados. Se definida como `0`, esta opção é ignorada.
`scan.auto-commit`	BOOLEAN	Não	`true`	Ativa o auto-commit para transações de leitura.

Opções de destino

Opção	Tipo	Obrigatória	Padrão	Descrição
`sink.buffer-flush.max-rows`	INTEGER	Não	`100`	Máximo de linhas no buffer antes do flush. Defina como `0` para fazer flush imediato de cada linha.
`sink.buffer-flush.interval`	DURATION	Não	`1000` ms	Tempo máximo de retenção das linhas no buffer antes do flush. Defina como `0` para fazer flush imediato de cada linha.
`sink.max-retries`	INTEGER	Não	`3`	Máximo de tentativas de gravação em caso de falha.
`sink.ignore-delete`	BOOLEAN	Não	`false`	Ignora mensagens de exclusão em vez de encaminhá-las. Requer VVR 11.4+.
`sink.ignore-delete-mode`	STRING	Não	`ALL`	Controla quais mensagens de exclusão são ignoradas quando `sink.ignore-delete` é `true`. Requer VVR 11.4+. Valores válidos: `ALL` (ignora `-D` e `-U`), `REAL_DELETE` (ignora apenas `-D`), `UPDATE_BEFORE` (ignora apenas `-U`).

Nota

Para liberar as linhas do buffer de forma assíncrona por temporizador em vez de por contagem de linhas, defina sink.buffer-flush.max-rows como 0 e configure sink.buffer-flush.interval com o intervalo de flush desejado.

Opções de tabela de dimensão

Opção	Tipo	Obrigatória	Padrão	Descrição
`lookup.cache.max-rows`	INTEGER	Não	—	Máximo de linhas no cache de consulta. Quando o cache está cheio, a linha menos recentemente usada expira. O cache fica desativado a menos que `lookup.cache.max-rows` e `lookup.cache.ttl` estejam definidos.
`lookup.cache.ttl`	DURATION	Não	—	Tempo máximo de validade de uma linha em cache antes de expirar.
`lookup.cache.caching-missing-key`	BOOLEAN	Não	`true`	Armazena em cache os resultados de consulta vazios para evitar consultas repetidas ao banco de dados para chaves inexistentes. Defina como `false` para não armazenar resultados vazios em cache.
`lookup.max-retries`	INTEGER	Não	`3`	Máximo de tentativas quando uma consulta ao banco de dados falha.

Opções do PostgreSQL

Opção	Tipo	Obrigatória	Padrão	Descrição
`source.extend-type.enabled`	BOOLEAN	Não	`false`	Quando `true`, as tabelas de origem e as tabelas de dimensão mapeiam colunas `JSONB` e `UUID` do PostgreSQL para `STRING` do Flink. Para joins de consulta em colunas UUID, adicione também `stringtype=unspecified` à URL JDBC para que o PostgreSQL consulte pelo tipo real em vez de fazer conversão.

Comportamentos principais

Leitura particionada

Para ativar leituras paralelas de uma tabela de origem, configure as opções scan.partition.column, scan.partition.lower-bound e scan.partition.upper-bound junto com scan.partition.num. A coluna de partição deve ser numérica ou TIMESTAMP. Para detalhes sobre como as divisões são calculadas, consulte Leitura particionada na documentação do Apache Flink.

Cache de consulta

Por padrão, cada consulta de lookup join acessa o banco de dados diretamente. Ative o cache LRU definindo lookup.cache.max-rows e lookup.cache.ttl para reduzir a carga no banco de dados ao custo de dados levemente desatualizados.

Gravações idempotentes

Quando a tabela de destino tem uma chave primária, o conector emite instruções de upsert específicas do banco de dados. Para MySQL, o conector usa INSERT ... ON DUPLICATE KEY UPDATE ....

Exemplos

Todos os três exemplos usam um conector blackhole ou datagen como contraparte leve para que as instruções sejam autônomas.

Ler de um banco de dados (tabela de origem)

CREATE TEMPORARY TABLE jdbc_source (
  `id`   INT,
  `name` VARCHAR
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:mysql://localhost:3306/mydb',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

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

INSERT INTO blackhole_sink SELECT * FROM jdbc_source;

Gravar em um banco de dados (tabela de destino)

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

CREATE TEMPORARY TABLE jdbc_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:mysql://localhost:3306/mydb',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

INSERT INTO jdbc_sink SELECT * FROM datagen_source;

Enriquecer um stream com consultas ao banco de dados (tabela de dimensão)

CREATE TEMPORARY TABLE datagen_source (
  `id`       INT,
  `data`     BIGINT,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE jdbc_dim (
  `id`   INT,
  `name` VARCHAR
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:mysql://localhost:3306/mydb',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  `id`   INT,
  `data` BIGINT,
  `name` VARCHAR
) WITH (
  'connector' = 'blackhole'
);

-- Look up the matching name for each stream record at processing time
INSERT INTO blackhole_sink
SELECT T.`id`, T.`data`, H.`name`
FROM datagen_source AS T
JOIN jdbc_dim FOR SYSTEM_TIME AS OF T.proctime AS H
  ON T.id = H.id;

Mapeamento de tipos de dados

MySQL	Oracle	PostgreSQL	Flink SQL
`TINYINT`	—	—	`TINYINT`
`SMALLINT`, `TINYINT UNSIGNED`	—	`SMALLINT`, `INT2`, `SMALLSERIAL`, `SERIAL2`	`SMALLINT`
`INT`, `MEDIUMINT`, `SMALLINT UNSIGNED`	—	`INTEGER`, `SERIAL`	`INT`
`BIGINT`, `INT UNSIGNED`	—	`BIGINT`, `BIGSERIAL`	`BIGINT`
`BIGINT UNSIGNED`	—	—	`DECIMAL(20, 0)`
`FLOAT`	`BINARY_FLOAT`	`REAL`, `FLOAT4`	`FLOAT`
`DOUBLE`, `DOUBLE PRECISION`	`BINARY_DOUBLE`	`FLOAT8`, `DOUBLE PRECISION`	`DOUBLE`
`NUMERIC(p, s)`, `DECIMAL(p, s)`	`SMALLINT`, `FLOAT(s)`, `DOUBLE PRECISION`, `REAL`, `NUMBER(p, s)`	`NUMERIC(p, s)`, `DECIMAL(p, s)`	`DECIMAL(p, s)`
`BOOLEAN`, `TINYINT(1)`	—	`BOOLEAN`	`BOOLEAN`
`DATE`	`DATE`	`DATE`	`DATE`
`TIME [(p)]`	`DATE`	`TIME [(p)] [WITHOUT TIMEZONE]`	`TIME [(p)] [WITHOUT TIMEZONE]`
`DATETIME [(p)]`	`TIMESTAMP [(p)] [WITHOUT TIMEZONE]`	`TIMESTAMP [(p)] [WITHOUT TIMEZONE]`	`TIMESTAMP [(p)] [WITHOUT TIMEZONE]`
`CHAR(n)`, `VARCHAR(n)`, `TEXT`	`CHAR(n)`, `VARCHAR(n)`, `CLOB`	`CHAR(n)`, `CHARACTER(n)`, `VARCHAR(n)`, `CHARACTER VARYING(n)`, `TEXT`, `JSONB`, `UUID`	`STRING`
`BINARY`, `VARBINARY`, `BLOB`	`RAW(s)`, `BLOB`	`BYTEA`	`BYTES`
—	—	`ARRAY`	`ARRAY`

Próximos passos

Criar uma tabela de origem MySQL CDC — capture dados de alteração em tempo real do MySQL
Criar uma tabela de origem PostgreSQL CDC (preview público) — capture dados de alteração em tempo real do PostgreSQL