AnalyticDB for MySQL V3.0 connector - Realtime Compute for Apache Flink

El conector de AnalyticDB for MySQL V3.0 permite leer y escribir en clústeres de AnalyticDB for MySQL V3.0 mediante Flink SQL. AnalyticDB for MySQL es un servicio de almacenamiento de datos nativo de la nube que admite escrituras en tiempo real de alto rendimiento, análisis de baja latencia y operaciones complejas de extracción, transformación y carga (ETL).

El conector admite los siguientes tipos de tablas y capacidades:

Elemento	Descripción
Tipos de tabla	Tabla de origen, tabla de dimensiones y tabla sink. Nota Las tablas de origen requieren Ververica Runtime (VVR) 8.0.4 o posterior. Para consultar los parámetros de la tabla de origen, consulte Use Flink to subscribe to binary logs.
Modos de ejecución	Modo streaming y modo por lotes
Formato de datos	N/A
Métricas	N/A
Tipo de API	API de SQL
Actualización y eliminación de datos en tablas sink	Compatible

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un clúster y una tabla de AnalyticDB for MySQL. Consulte Create a cluster y CREATE TABLE
Una lista blanca de direcciones IP configurada para el clúster. Consulte Configure an IP address whitelist

Sintaxis

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

La clave primaria definida en la instrucción DDL de Flink debe coincidir con la clave primaria de la tabla física en AnalyticDB for MySQL; es decir, deben tener los mismos campos, los mismos nombres y estar presentes en ambas. Las claves primarias que no coinciden pueden provocar corrupción de datos.

Comportamientos clave

Clave primaria y modo de escritura

Cuando se define una clave primaria en el DDL, el conector opera en modo upsert, donde las claves primarias duplicadas desencadenan una actualización en lugar de una nueva inserción. El comportamiento exacto de SQL depende del parámetro replaceMode:

replace — utiliza REPLACE INTO. Una clave primaria duplicada sobrescribe toda la fila existente.
upsert — utiliza INSERT INTO ... ON DUPLICATE KEY UPDATE. Solo se actualizan los campos especificados; los demás conservan sus valores actuales.
insert — utiliza INSERT IGNORE INTO. Las claves primarias duplicadas se ignoran silenciosamente y se conserva la fila existente.

Si no se define ninguna clave primaria, el conector siempre utiliza INSERT IGNORE INTO.

replaceMode requiere AnalyticDB for MySQL V3.1.3.5 o posterior y VVR 11.2 o posterior para los valores de cadena ( replace , upsert , insert ). Las versiones anteriores de VVR utilizan true (equivalente a replace ) y false (equivalente a upsert ). VVR 11.2 y posteriores siguen siendo compatibles con true y false .

Búfer de escritura

El conector almacena registros en un búfer en memoria y los vacía por lotes. El vaciado se activa cuando se cumple alguna de las siguientes condiciones:

El número de registros en el búfer alcanza batchSize (valor predeterminado: 1,000) o bufferSize (valor predeterminado: 1,000)
El tiempo transcurrido desde el último vaciado alcanza flushIntervalMs (valor predeterminado: 3,000 ms)

Tanto batchSize como bufferSize solo surten efecto cuando se define una clave primaria.

Caché de tablas de dimensiones

El conector admite tres políticas de caché para tablas de dimensiones. El uso de caché reduce las consultas a la tabla física, pero requiere memoria adicional.

Política de caché	Comportamiento	Recomendado cuando
`ALL` (predeterminado)	Todos los datos de la tabla de dimensiones se cargan en memoria antes de que comience el trabajo. Las búsquedas posteriores solo consultan la caché. Los datos se recargan cuando expira `cacheTTLMs`.	La tabla de dimensiones es pequeña y es común que falten claves
`LRU`	Se almacenan en caché las filas a las que se accede con frecuencia. Si hay un fallo de caché, el conector consulta la tabla física y actualiza la caché. Las entradas expiran después de `cacheTTLMs`.	La tabla de dimensiones es grande, pero el acceso se concentra en un subconjunto de claves
`None`	Sin caché. Cada búsqueda consulta directamente la tabla física.	Hay poco volumen de búsquedas o requisitos estrictos de actualización de datos

Importante

Con la caché ALL, el conector carga de forma asíncrona la tabla de dimensiones completa en la memoria. Debe aumentar la memoria del nodo utilizado para unir tablas (el aumento requerido es al menos el doble del tamaño de la tabla remota) para evitar errores de falta de memoria (OOM). Supervise el uso de memoria del nodo durante la ejecución del trabajo.

Parámetros en la cláusula WITH

Parámetros comunes

Parámetro	Tipo	Obligatorio	Valor predeterminado	Descripción
`connector`	String	Sí	—	Establezca este valor en `adb3.0`
`url`	String	Sí	—	URL de Java Database Connectivity (JDBC) de la base de datos, con el formato `jdbc:mysql://<endpoint>:<port>/<databaseName>`. Obtenga el endpoint y el puerto en la sección Network Information de la página de su clúster en AnalyticDB for MySQL console.
`userName`	String	Sí	—	Nombre de usuario para acceder a la base de datos
`password`	String	Sí	—	Contraseña para acceder a la base de datos
`tableName`	String	Sí	—	Nombre de la tabla de destino en la base de datos
`maxRetryTimes`	Integer	No	10	Número máximo de reintentos tras un fallo de lectura o escritura

Parámetros de tabla sink

Parámetro	Tipo	Obligatorio	Valor predeterminado	Descripción
`batchSize`	Integer	No	1000	Número de registros escritos por lote. Solo surte efecto cuando se define una clave primaria.
`bufferSize`	Integer	No	1000	Número máximo de registros almacenados en el búfer de memoria antes de activar un vaciado. Solo surte efecto cuando se define una clave primaria.
`flushIntervalMs`	Integer	No	3000	Tiempo máximo (en milisegundos) entre vaciados. Cuando transcurre este intervalo, todos los registros en el búfer se escriben independientemente del tamaño del búfer.
`ignoreDelete`	Boolean	No	false	Indica si se deben ignorar las operaciones de eliminación. Establezca este valor en `true` para omitir eliminaciones o en `false` para aplicarlas.
`replaceMode`	Boolean	No	`true`	Modo de escritura cuando se define una clave primaria. VVR 11.2+: `replace`, `upsert` o `insert`. Versiones anteriores: `true` (equivalente a `replace`) o `false` (equivalente a `upsert`). Requiere AnalyticDB for MySQL V3.1.3.5 o posterior. Solo surte efecto cuando se define una clave primaria en el DDL; de lo contrario, siempre se utiliza `INSERT IGNORE INTO`.
`excludeUpdateColumns`	String	No	(vacío)	Lista separada por comas de columnas que se excluirán de las actualizaciones cuando `replaceMode` sea `upsert` (o `false`). Si hay una clave primaria duplicada, solo se actualizan las columnas restantes y las excluidas conservan sus valores existentes. Si la clave primaria es única, se insertan todas las columnas. Ejemplo: `excludeUpdateColumns='column1,column2'`. El valor debe estar en una sola línea sin saltos de línea.
`connectionMaxActive`	Integer	No	40	Número máximo de conexiones simultáneas a la base de datos en el pool de hilos

Parámetros de tabla de dimensiones

Parámetro	Tipo	Obligatorio	Valor predeterminado	Descripción
`cache`	String	No	ALL	Política de caché: `None`, `LRU` o `ALL`. Consulte Caché de tablas de dimensiones.
`cacheSize`	Integer	No	100000	Número máximo de filas en caché. Es obligatorio cuando `cache` es `LRU`.
`cacheTTLMs`	Integer	No	Long.MAX_VALUE	Tiempo de vida de las entradas de caché en milisegundos. Para `LRU`: las entradas expiran después de este periodo (valor predeterminado: nunca expiran). Para `ALL`: la caché completa se recarga después de este periodo (valor predeterminado: nunca se recarga). No se utiliza cuando `cache` es `None`.
`maxJoinRows`	Integer	No	1024	Número máximo de filas de la tabla de dimensiones que coinciden por cada registro de entrada. Si cada registro de entrada coincide con un máximo de n filas de la tabla de dimensiones, establezca este valor en n para optimizar el rendimiento de la unión en Realtime Compute for Apache Flink.

Asignaciones de tipos de datos

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) o NUMERIC(p, s)	DECIMAL(p, s)
VARCHAR	STRING
BINARY	BYTES
DATE	DATE
TIME	TIME
DATETIME	TIMESTAMP
TIMESTAMP	TIMESTAMP
POINT	STRING

Ejemplos

Tabla sink

En el siguiente ejemplo se lee desde un origen datagen y se escriben los datos en una tabla sink de 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;

Tabla de dimensiones

En el siguiente ejemplo se une un origen datagen con una tabla de dimensiones de AnalyticDB for MySQL mediante una unión temporal. Los resultados se escriben en un 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;

Referencias

Error: multi-statement be found