JDBC connector - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

El conector JDBC lee y escribe en bases de datos relacionales (MySQL, PostgreSQL y Oracle) mediante DDL SQL estándar en trabajos de Flink SQL.

Tipos de tabla compatibles: tabla de origen · tabla de dimensiones · tabla de destino

Modos de ejecución compatibles: modo streaming · modo por lotes · API de SQL

Actualizaciones y eliminaciones en tablas de destino: compatible

Requisitos previos

Antes de empezar, compruebe lo siguiente:

La base de datos y la tabla de destino ya existen.
El archivo JAR del controlador JDBC de su base de datos está disponible para cargarlo.

Limitaciones

Lecturas acotadas: una tabla de origen JDBC es un origen acotado. La tarea finaliza cuando se leen todas las filas. Para capturar datos de cambios en tiempo real, utilice un conector de Change Data Capture (CDC); consulte Create a MySQL CDC source table y Create a PostgreSQL CDC source table (public preview).
Versión de PostgreSQL: la escritura en PostgreSQL requiere la versión 9.5 o posterior, porque la tabla de destino depende de la cláusula ON CONFLICT.
Carga del controlador JDBC: cargue el archivo JAR del controlador JDBC como archivo de dependencia antes de ejecutar su trabajo. Controladores comunes:

Base de datos

Group ID

Artifact ID

MySQL

mysql

mysql-connector-java

Oracle

com.oracle.database.jdbc

ojdbc8

PostgreSQL

org.postgresql

postgresql

Para los controladores que no aparecen en esta lista, compruebe la compatibilidad antes de usarlos.
Comportamiento upsert en MySQL: al escribir en una tabla de destino MySQL con clave primaria, el conector emite instrucciones INSERT INTO ... ON DUPLICATE KEY UPDATE ....

Aviso
Si inserta filas con valores duplicados en un índice único, incluso cuando las claves primarias difieren, se sobrescriben las filas existentes en cualquier tabla física con restricción de índice único. Esto provoca pérdida de datos.

Crear una tabla 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>'
);

Opciones del conector

Generales

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`connector`	STRING	Sí	—	Establezca este valor en `jdbc`.
`url`	STRING	Sí	—	URL JDBC de la base de datos.
`table-name`	STRING	Sí	—	Nombre de la tabla desde la cual se lee o hacia donde se escribe.
`username`	STRING	No	—	Nombre de usuario de la base de datos. Se configura junto con `password`. Ambas opciones admiten una referencia de credencial de KMS con el formato `secret://kms.<secret-name>.<key>`.
`password`	STRING	No	—	Contraseña de la base de datos.
`kms.endpoint`	STRING	No	—	Endpoint de la puerta de enlace compartida o dedicada de KMS; por ejemplo, `kms.cn-hangzhou.aliyuncs.com`. Consulte KMS endpoints. Es obligatoria cuando utiliza credenciales de KMS. Para más detalles, consulte KMS credential management integration.
`kms.akless.assume-role.role-name`	STRING	No	—	Nombre del rol de RAM que se usa para leer las credenciales de KMS. El rol debe referenciarse previamente en Security. Es obligatorio cuando utiliza credenciales de KMS. Para más detalles, consulte KMS credential management integration.
`kms.ca`	STRING	No	—	Ruta del certificado CA de la puerta de enlace dedicada de KMS. Configure esta opción con una ruta absoluta de OSS que tenga el formato `oss://flink-fullymanaged-<workspace ID>/artifacts/namespaces/<namespace name>/<file name>`. Es obligatoria cuando utiliza una puerta de enlace dedicada de KMS. Para más detalles, consulte KMS credential management integration.

Opciones de origen

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`scan.partition.column`	STRING	No	—	Columna que se usa para dividir los datos en particiones. Debe ser de tipo NUMERIC o TIMESTAMP, y la base de datos debe admitir la comparación de ese tipo con valores numéricos. Consulte Escaneo por particiones.
`scan.partition.num`	INTEGER	No	—	Número de particiones.
`scan.partition.lower-bound`	LONG	No	—	Valor mínimo de la primera partición.
`scan.partition.upper-bound`	LONG	No	—	Valor máximo de la última partición.
`scan.fetch-size`	INTEGER	No	`0`	Filas recuperadas por cada viaje de ida y vuelta a la base de datos. Si se establece en `0`, esta opción se ignora.
`scan.auto-commit`	BOOLEAN	No	`true`	Habilita el auto-commit en las transacciones de lectura.

Opciones de destino

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`sink.buffer-flush.max-rows`	INTEGER	No	`100`	Número máximo de filas que se almacenan en búfer antes de vaciarlo. Establezca este valor en `0` para vaciar cada fila de inmediato.
`sink.buffer-flush.interval`	DURATION	No	`1000` ms	Tiempo máximo que se retienen las filas en búfer antes de vaciarlo. Establezca este valor en `0` para vaciar cada fila de inmediato.
`sink.max-retries`	INTEGER	No	`3`	Número máximo de reintentos de escritura tras un error.
`sink.ignore-delete`	BOOLEAN	No	`false`	Ignora los mensajes de eliminación en lugar de reenviarlos. Requiere VVR 11.4 o posterior.
`sink.ignore-delete-mode`	STRING	No	`ALL`	Controla qué mensajes de eliminación se ignoran cuando `sink.ignore-delete` es `true`. Requiere VVR 11.4 o posterior. Valores válidos: `ALL` (ignora `-D` y `-U`), `REAL_DELETE` (ignora solo `-D`) y `UPDATE_BEFORE` (ignora solo `-U`).

Nota

Para vaciar las filas almacenadas en búfer de forma asíncrona mediante un temporizador, en lugar de hacerlo por cantidad de filas, establezca sink.buffer-flush.max-rows en 0 y configure sink.buffer-flush.interval con el intervalo de vaciado deseado.

Opciones de tabla de dimensiones

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`lookup.cache.max-rows`	INTEGER	No	—	Número máximo de filas en la caché de búsqueda. Cuando la caché se llena, expira la fila menos utilizada recientemente. El almacenamiento en caché se deshabilita a menos que se configure tanto `lookup.cache.max-rows` como `lookup.cache.ttl`.
`lookup.cache.ttl`	DURATION	No	—	Tiempo máximo durante el cual una fila almacenada en caché es válida antes de expirar.
`lookup.cache.caching-missing-key`	BOOLEAN	No	`true`	Almacena en caché los resultados de búsqueda vacíos para evitar consultas repetidas a la base de datos cuando faltan claves. Establezca este valor en `false` para omitir el almacenamiento en caché de los resultados vacíos.
`lookup.max-retries`	INTEGER	No	`3`	Número máximo de reintentos cuando falla una consulta a la base de datos.

Opciones de PostgreSQL

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`source.extend-type.enabled`	BOOLEAN	No	`false`	Cuando es `true`, las tablas de origen y de dimensiones asignan las columnas `JSONB` y `UUID` de PostgreSQL al tipo `STRING` de Flink. Para hacer lookup joins en columnas UUID, agregue también `stringtype=unspecified` a la URL JDBC. Así, PostgreSQL realiza las consultas por tipo real en lugar de aplicar conversiones.

Comportamientos clave

Escaneo por particiones

Para habilitar lecturas paralelas desde una tabla de origen, configure las opciones scan.partition.column, scan.partition.lower-bound y scan.partition.upper-bound junto con scan.partition.num. La columna de partición debe ser numérica o de tipo TIMESTAMP. Para conocer cómo se calculan las divisiones, consulte Partitioned Scan en la documentación de Apache Flink.

Caché de búsqueda

De forma predeterminada, cada consulta de lookup join accede directamente a la base de datos. Habilite la caché LRU si configura lookup.cache.max-rows y lookup.cache.ttl. Esto reduce la carga de la base de datos, aunque los datos pueden estar ligeramente desactualizados.

Escrituras idempotentes

Cuando la tabla de destino tiene una clave primaria, el conector emite instrucciones upsert específicas de cada base de datos. En MySQL, utiliza INSERT ... ON DUPLICATE KEY UPDATE ....

Ejemplos

Los tres ejemplos usan un conector blackhole o datagen como contraparte ligera, de modo que las instrucciones sean autosuficientes.

Leer desde una base de datos (tabla de origen)

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;

Escribir en una base de datos (tabla 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 un flujo con búsquedas en la base de datos (tabla de dimensiones)

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;

Asignación de tipos de datos

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`

Pasos siguientes

Create a MySQL CDC source table: captura datos de cambios en tiempo real desde MySQL.
Create a PostgreSQL CDC source table (public preview): captura datos de cambios en tiempo real desde PostgreSQL.

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