how to use the apsaradb rds for mysql connector - Realtime Compute for Apache Flink

Importante

El conector de ApsaraDB RDS for MySQL dejará de ser compatible en el futuro. Utilice MySQL connector en su lugar.

El conector de ApsaraDB RDS for MySQL permite escribir la salida de Flink SQL en una tabla de destino de ApsaraDB RDS for MySQL o unir un flujo de streaming con una tabla de dimensiones de ApsaraDB RDS for MySQL.

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

Supported running modes: Modo por lotes · Modo de streaming

Tipo de API: SQL

Actualizaciones y eliminaciones de datos en tablas de destino: Compatible

Requisitos previos

Antes de empezar, asegúrese de contar con:

Una base de datos y una tabla de ApsaraDB RDS for MySQL. Consulte Create databases and accounts for an ApsaraDB RDS for MySQL instance
Una lista de permitidos de direcciones IP configurada para la base de datos. Consulte Connect to an ApsaraDB RDS for MySQL instance using a database client or the CLI

Límites

Se requiere Realtime Compute for Apache Flink con Ververica Runtime (VVR) 2.0.0 o posterior. Para obtener el mejor rendimiento y estabilidad, utilice VVR 6.X o posterior.
Solo se admiten bases de datos de ApsaraDB RDS for MySQL.
El conector utiliza semántica de al menos una vez. Si la tabla de destino tiene una clave principal, la idempotencia garantiza la corrección de los datos.

Funcionamiento

Comportamiento de escritura en la tabla de destino

Cada fila de salida se convierte en una sentencia SQL antes de escribirse en la tabla de destino:

Sin clave principal: ejecuta INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...);
Con clave principal: ejecuta INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...) ON DUPLICATE KEY UPDATE col1 = VALUES(col1), col2 = VALUES(col2), ...;

Conflictos de índice único: Si la tabla física tiene una restricción de índice único además de la clave principal, la inserción de dos filas con claves principales diferentes pero el mismo valor de índice único provoca que la fila anterior se sobrescriba, lo que resulta en pérdida de datos.

Claves principales de incremento automático: No declare campos de incremento automático en la DDL de Flink. La base de datos asigna estos valores automáticamente. El conector puede escribir y eliminar filas con campos de incremento automático, pero no puede actualizarlas.

Directivas de caché para tablas de dimensiones

El conector admite tres directivas de caché para las consultas a tablas de dimensiones:

Directiva	Comportamiento	Cuándo utilizarla
`NONE`	Sin almacenamiento en caché: cada consulta accede directamente a la base de datos	Requisitos de baja latencia, conjuntos de datos pequeños
`LRU`	Almacena en caché un número fijo de filas utilizadas recientemente por administrador de tareas	Subconjuntos de acceso frecuente en tablas grandes
`ALL`	Carga toda la tabla en memoria y la recarga periódicamente	Tablas de referencia pequeñas y estáticas

Sintaxis

Tabla de destino

CREATE TABLE rds_sink (
  id  INT,
  num BIGINT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'  = 'rds',
  'tableName'  = '<your-table-name>',
  'userName'   = '<your-user-name>',
  'password'   = '<your-password>',
  'url'        = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>?rewriteBatchedStatements=true'
);

Nota

Agregue ?rewriteBatchedStatements=true al valor de url en las tablas de destino para mejorar el throughput de escritura.

Tabla de dimensiones

CREATE TABLE rds_dim (
  id1 INT,
  id2 VARCHAR
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>',
  'cache'     = 'NONE'
);

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 `rds`
`tableName`	STRING	Sí	—	Nombre de la tabla física en ApsaraDB RDS for MySQL
`userName`	STRING	Sí	—	Nombre de usuario de la base de datos
`password`	STRING	Sí	—	Contraseña de la base de datos
`url`	STRING	Sí	—	Endpoint de la virtual private cloud (VPC) de la base de datos, con el formato `jdbc:mysql://<internal-endpoint>:<port>/<database-name>`. Para tablas de destino, agregue `?rewriteBatchedStatements=true`. Para más detalles sobre endpoints, consulte View and change the internal and public endpoints and port numbers of an ApsaraDB RDS for MySQL instance
`maxRetryTimes`	INTEGER	No	10 (VVR 4.0.7+), 3 (VVR 4.0.6 y anteriores)	Número máximo de reintentos para consultas fallidas a tablas de dimensiones o escrituras en tablas de destino

Parámetros de la tabla de destino

Parámetro	Tipo	Obligatorio	Valor predeterminado	Descripción
`batchSize`	INTEGER	No	4096 (VVR 4.0.7+), 5000 (VVR 4.0.0–4.0.6), 100 (VVR 3.x y anteriores)	Número de filas escritas por lote
`bufferSize`	INTEGER	No	10000	Número máximo de filas almacenadas en caché en memoria antes de desencadenar una escritura. Compatible con VVR 4.0.7 y posteriores. Solo surte efecto cuando se define una clave principal
`flushIntervalMs`	INTEGER	No	2000 (VVR 4.0.7+), 0 (VVR 4.0.0–4.0.6), 1000 (VVR 3.x y anteriores)	Intervalo en milisegundos en el que se vacía el búfer hacia la tabla de destino, independientemente de si se alcanzan los umbrales de `batchSize` o `bufferSize`. Si se establece en `0` (valor predeterminado para VVR 4.0.0–4.0.6), es posible que pequeñas cantidades de datos en el búfer nunca se escriban; actualice a una versión posterior de VVR para evitar esto
`ignoreDelete`	BOOLEAN	No	false	Establezca este valor en `true` para omitir operaciones de eliminación. Útil cuando varios operadores actualizan diferentes campos de la misma fila; sin esta configuración, una eliminación en un operador seguida de una actualización parcial en otro deja los campos no actualizados como `null` o con sus valores predeterminados
`connectionMaxActive`	INTEGER	No	40	Tamaño del pool de conexiones. Compatible con VVR 4.0.7 y posteriores. Aumente este valor si ocurren tiempos de espera agotados en el pool de conexiones; disminúyalo si la base de datos limita el número de conexiones simultáneas

Parámetros de la tabla de dimensiones

Parámetro	Tipo	Obligatorio	Valor predeterminado	Descripción
`cache`	STRING	No	NONE (VVR anterior a 4.0.6), ALL (VVR 4.0.6+)	Directiva de caché. Valores válidos: `NONE`, `LRU`, `ALL`. Consulte Cache policies
`cacheSize`	INTEGER	No	100000	Número máximo de filas para almacenar en caché. Obligatorio cuando `cache` se establece en `LRU`; se ignora para `NONE` y `ALL`
`cacheTTLMs`	LONG	No	Sin caducidad para `NONE` y `LRU`; sin recarga para `ALL`	Tiempo de vida de la caché en milisegundos. Para `LRU`, las filas caducan después de este periodo. Para `ALL`, toda la caché se recarga en este intervalo
`maxJoinRows`	INTEGER	No	1024	Número máximo de filas de la tabla de dimensiones que coinciden por cada fila de entrada. Configure este valor con el número máximo de filas de dimensiones esperado por cada fila de la tabla principal para evitar escaneos innecesarios

Métricas

La tabla de destino expone las siguientes métricas. Las tablas de dimensiones no tienen métricas.

Métrica	Descripción
`numRecordsOut`	Total de filas escritas
`numRecordsOutPerSecond`	Filas escritas por segundo
`numBytesOut`	Total de bytes escritos
`numBytesOutPerSecond`	Bytes escritos por segundo
`currentSendTime`	Latencia de escritura actual
`numRecordsOutErrors`	Total de errores de escritura

Para consultar las definiciones de las métricas, véase Métricas.

Asignaciones de tipos de datos

Tipo de Flink	Tipo de ApsaraDB RDS for MySQL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
TINYINT(1) (solo tablas de dimensiones)	BOOLEAN
SMALLINT	SMALLINT
SMALLINT	TINYINT UNSIGNED
INT	INT
INT	SMALLINT UNSIGNED
BIGINT	BIGINT
BIGINT	INT UNSIGNED
DECIMAL(20, 0)	BIGINT UNSIGNED
FLOAT	FLOAT
DECIMAL	DECIMAL
DOUBLE	DOUBLE
DATE	DATE
TIME	TIME
TIMESTAMP	TIMESTAMP
VARCHAR	VARCHAR
VARBINARY	VARBINARY

Ejemplos

Ejemplo de tabla de destino

El siguiente ejemplo lee desde un origen DataGen y escribe en una tabla de destino de ApsaraDB RDS for MySQL.

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

CREATE TEMPORARY TABLE rds_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>?rewriteBatchedStatements=true'
);

INSERT INTO rds_sink
SELECT * FROM datagen_source;

Ejemplo de tabla de dimensiones

El siguiente ejemplo une un flujo de streaming con una tabla de dimensiones de ApsaraDB RDS for MySQL mediante una unión temporal.

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

CREATE TEMPORARY TABLE rds_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>'
);

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

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

Preguntas frecuentes

Pasos siguientes

MySQL connector: el reemplazo recomendado para este conector
ApsaraDB RDS for MySQL: descripción general del producto y documentación de funciones
Métricas: definiciones de todas las métricas del conector