Realtime Compute for Apache Flink:Conector SQL de MySQL

El conector de MySQL admite lo siguiente:

• Procesamiento unificado por lotes y en streaming que permite leer tanto datos completos como incrementales. Esto elimina la necesidad de mantener dos procesos separados.

• Lectura simultánea de datos completos para escalar horizontalmente el rendimiento.

• Cambio fluido entre la lectura de datos completos y la de datos incrementales, con reducción automática de escala para ahorrar recursos de cómputo.

• Transferencia de datos reanudable durante la fase de lectura completa, lo que mejora la estabilidad.

• Lectura de datos completos sin bloqueos, que no afecta los servicios en línea.

• Compatibilidad con la lectura de registros de copia de seguridad de ApsaraDB RDS for MySQL.

• Análisis sintáctico paralelo de archivos de registro binario para reducir la latencia de lectura.

SQL

Puede usar el conector de MySQL en trabajos SQL como tabla de origen, tabla de dimensiones o tabla de destino.

Sintaxis

CREATE TEMPORARY TABLE mysqlcdc_source (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

Parámetros WITH

• General

  Parámetro	Descripción	Obligatorio	Tipo de dato	Valor predeterminado	Notas
  connector	El tipo de tabla.	Sí	STRING	Ninguna	Cuando se usa como tabla de origen, establezca este parámetro en mysql-cdc o mysql. Ambos valores son equivalentes. Si la utiliza como tabla de dimensiones o tabla de destino, el valor debe ser mysql.
  hostname	La dirección IP o el nombre de host de la base de datos MySQL.	Sí	STRING	Ninguna	Se recomienda especificar una dirección de red privada virtual (VPC). Nota Si la base de datos MySQL y Realtime Compute for Apache Flink no están en la misma VPC, debe establecer una conexión de red entre distintas VPC o usar un endpoint público para acceder a la base de datos. Para obtener más información, consulte Manage and operate workspaces y How can a fully managed Flink cluster access the Internet?.
  username	El nombre de usuario del servicio de base de datos MySQL.	Sí	STRING	Ninguna	Ninguna.
  password	La contraseña del servicio de base de datos MySQL.	Sí	STRING	Ninguna	Ninguna.
  database-name	El nombre de la base de datos MySQL.	Sí	STRING	Ninguna	Cuando se utiliza una base de datos como tabla de origen, puede usar una expresión regular en el nombre de la base de datos para leer datos de varias bases de datos. Al usar expresiones regulares, no utilice los símbolos ^ ni $ para coincidir con el inicio y el final de la cadena. Para obtener más información, consulte las notas del parámetro table-name.
  table-name	El nombre de la tabla MySQL.	Sí	STRING	Ninguna	Puede usar una expresión regular en el nombre de la tabla de origen para leer datos de varias tablas. Al usar expresiones regulares, no utilice los símbolos ^ ni $ para coincidir con el inicio y el final de la cadena. Para obtener más información, consulte la siguiente nota. Nota Cuando una tabla de origen MySQL CDC coincide con nombres de tablas mediante una expresión regular, concatena los valores de database-name y table-name que usted especifique con la cadena \\. para formar una expresión regular de ruta completa. Antes de VVR 8.0.1, se usaba el carácter .. Después, el conector utiliza esta expresión regular para coincidir con los nombres completos de las tablas en la base de datos MySQL. Por ejemplo, si establece 'database-name'='db_.' y 'table-name'='tb_.+', el conector usa la expresión regular db_.\\.tb_.+ para coincidir con los nombres completos de las tablas y determinar cuáles leer. Antes de VVR 8.0.1, la expresión regular era db_.*.tb_.+.
  port	El número de puerto del servicio de base de datos MySQL.	No	INTEGER	3306	Ninguna.

• Solo tabla de origen

  Parámetro	Descripción	Obligatorio	Tipo de dato	Valor predeterminado	Notas
  server-id	Identificador numérico del cliente de la base de datos.	No	STRING	Se genera un valor aleatorio entre 5400 y 6400.	Este identificador debe ser único a nivel global dentro del clúster de MySQL. Establezca un identificador distinto para cada trabajo que se conecte a la misma base de datos. Este parámetro también admite un formato de rango de identificadores, por ejemplo, 5400-5408. Cuando se habilita la lectura incremental, se admite la lectura concurrente. En este caso, establezca un rango de identificadores para que cada lector concurrente utilice uno distinto. Para obtener más información, consulte Use Server ID.
  scan.incremental.snapshot.enabled	Indica si se habilitan las instantáneas incrementales.	No	BOOLEAN	true	Las instantáneas incrementales están habilitadas de forma predeterminada. Este es un mecanismo nuevo para leer instantáneas de datos completos. En comparación con el método anterior de lectura de instantáneas, ofrece varias ventajas, entre ellas: El origen puede leer los datos completos en paralelo. El origen admite puntos de control a nivel de fragmento durante la lectura de datos completos. El origen no necesita adquirir un bloqueo de lectura global (FLUSH TABLES WITH read lock) al leer los datos completos. Si desea que el origen admita lectura concurrente, cada lector requiere un identificador de servidor único. Por lo tanto, server-id debe definirse como un rango, por ejemplo, 5400-6400, y su tamaño debe ser mayor o igual que el nivel de concurrencia. Nota Este elemento de configuración se eliminó en Ververica Runtime (VVR) 11.1 y versiones posteriores.
  scan.incremental.snapshot.chunk.size	Tamaño de cada fragmento, expresado en número de filas.	No	INTEGER	8096	Cuando se habilita la lectura de instantáneas incrementales, la tabla se divide en varios fragmentos para su lectura. Los datos de un fragmento se almacenan en caché en memoria hasta que se leen por completo. Cuantas menos filas contenga cada fragmento, mayor será el número total de fragmentos en la tabla. Aunque esto mejora la granularidad de la recuperación ante errores, puede provocar errores de falta de memoria (OOM) y reducir el rendimiento general. Por lo tanto, evalúe ambos factores y establezca un tamaño de fragmento adecuado.
  scan.snapshot.fetch.size	Número máximo de registros que se extraen a la vez al leer los datos completos de una tabla.	No	INTEGER	1024	Ninguna.
  scan.startup.mode	Modo de inicio para el consumo de datos.	No	STRING	initial	Valores válidos: initial (predeterminado): en el primer inicio o en un inicio sin estado, el conector examina todos los datos históricos y luego lee los datos más recientes del registro binario. latest-offset: en el primer inicio o en un inicio sin estado, el conector no examina los datos históricos. Comienza a leer desde el final del registro binario, por lo que solo procesa los cambios más recientes generados después de su arranque. earliest-offset: el conector no examina los datos históricos. Inicia la lectura desde el registro binario disponible más antiguo. specific-offset: el conector no examina los datos históricos. Comienza desde un desplazamiento específico del registro binario. Puede especificar el desplazamiento mediante la configuración conjunta de scan.startup.specific-offset.file y scan.startup.specific-offset.pos, o bien configurar únicamente scan.startup.specific-offset.gtid-set para iniciar desde un conjunto GTID específico. timestamp: el conector no examina los datos históricos. Inicia la lectura del registro binario desde una marca de tiempo específica, definida en milisegundos mediante scan.startup.timestamp-millis. Importante Cuando utilice los modos de inicio earliest-offset, specific-offset o timestamp, asegúrese de que el esquema de la tabla correspondiente no cambie entre la posición de consumo del registro binario indicada y el momento de inicio del trabajo. Esto evita errores por discrepancias de esquema.
  scan.startup.specific-offset.file	Nombre del archivo de registro binario que define el desplazamiento inicial cuando se utiliza el modo de inicio specific-offset.	No	STRING	Ninguno	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset. Ejemplo de formato de nombre de archivo: mysql-bin.000003.
  scan.startup.specific-offset.pos	Desplazamiento dentro del archivo de registro binario indicado que define la posición inicial cuando se utiliza el modo de inicio specific-offset.	No	INTEGER	Ninguno	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset.
  scan.startup.specific-offset.gtid-set	Conjunto GTID que define el desplazamiento inicial cuando se utiliza el modo de inicio specific-offset.	No	STRING	Ninguno	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset. Ejemplo de formato de conjunto GTID: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	Marca de tiempo en milisegundos que define el desplazamiento inicial cuando se utiliza el modo de inicio timestamp.	No	LONG	Ninguno	Al usar este parámetro, debe establecer scan.startup.mode en timestamp. La unidad de la marca de tiempo es milisegundos. Importante Cuando se especifica una hora, MySQL CDC intenta leer el evento inicial de cada archivo de registro binario para determinar su marca de tiempo. Después, localiza el archivo correspondiente a la hora indicada. Asegúrese de que el archivo de registro binario asociado a dicha marca de tiempo no se haya eliminado de la base de datos y esté disponible para su lectura.
  server-time-zone	Zona horaria de sesión que utiliza la base de datos.	No	STRING	Si no especifica este parámetro, el sistema utiliza la zona horaria del entorno de ejecución del trabajo de Flink como zona horaria del servidor de la base de datos. Esta corresponde a la zona horaria de la región seleccionada.	Ejemplo: Asia/Shanghai. Este parámetro controla cómo se convierte el tipo TIMESTAMP de MySQL al tipo STRING. Para obtener más información, consulte Debezium temporal values.
  debezium.min.row.count.to.stream.results	Cuando el número de filas de una tabla supera este valor, se utiliza el modo de lectura por lotes.	No	INTEGER	1000	Flink lee los datos de una tabla de origen de MySQL de una de las siguientes maneras: Lectura completa: carga directamente en memoria todos los datos de la tabla. Este método es rápido, pero consume una cantidad proporcional de memoria. Si la tabla de origen es muy grande, existe riesgo de errores OOM. Lectura por lotes: procesa los datos en varios lotes, con una cantidad determinada de filas por lote, hasta completar la lectura. Evita los riesgos de OOM en tablas grandes, aunque resulta más lento.
  connect.timeout	Tiempo máximo de espera antes de que expire la conexión con el servidor de la base de datos MySQL y se intente de nuevo.	No	DURATION	30s	Ninguna.
  connect.max-retries	Número máximo de reintentos tras un error de conexión con el servicio de base de datos MySQL.	No	INTEGER	3	Ninguna.
  connection.pool.size	Tamaño del pool de conexiones de la base de datos.	No	INTEGER	20	El pool de conexiones permite reutilizarlas, lo que reduce la cantidad total de conexiones a la base de datos.
  jdbc.properties.*	Parámetros de conexión personalizados en la URL de JDBC.	No	STRING	Ninguno	Puede pasar parámetros de conexión personalizados. Por ejemplo, para no utilizar el protocolo SSL, configure 'jdbc.properties.useSSL' = 'false'. Para obtener más información sobre los parámetros de conexión compatibles, consulte MySQL Configuration Properties.
  debezium.*	Parámetros personalizados para que Debezium lea los registros binarios.	No	STRING	Ninguno	Puede pasar parámetros personalizados de Debezium. Por ejemplo, utilice 'debezium.event.deserialization.failure.handling.mode'='ignore' para definir la lógica de manejo de errores de análisis sintáctico. Aviso No modifique los parámetros de Debezium de forma arbitraria, ya que esto podría provocar que el conector lea los datos de manera incorrecta. Por ejemplo, no se permite configurar el parámetro debezium.binlog.buffer.size.
  heartbeat.interval	Intervalo con el que el origen avanza el desplazamiento del registro binario mediante eventos de latido.	No	DURATION	30s	Los eventos de latido permiten avanzar el desplazamiento del registro binario en el origen. Resultan especialmente útiles para las tablas de MySQL que se actualizan con poca frecuencia, ya que en esos casos el desplazamiento no avanza de forma automática. Estos eventos empujan el desplazamiento hacia adelante y evitan problemas derivados de su expiración. Un desplazamiento expirado puede hacer que el trabajo falle de manera irrecuperable y requiera un reinicio sin estado.
  scan.incremental.snapshot.chunk.key-column	Define la columna que se utilizará para dividir los datos en fragmentos durante la fase de instantánea.	Consulte la columna Notas.	STRING	Ninguno	Obligatorio para tablas sin clave primaria. La columna seleccionada debe ser de un tipo no nulo (NOT NULL). Opcional para tablas con clave primaria. Solo se puede seleccionar una columna de dicha clave.
  rds.region-id	Identificador de la región de la instancia de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se utiliza la función de lectura de registros archivados desde OSS.	STRING	Ninguno	Para obtener más información sobre los identificadores de región, consulte Regions and zones. Importante La cadena GTID de MySQL CDC se genera de forma aleatoria y no aumenta de manera monótona como los desplazamientos de los archivos de registro binario. Por ello, localizar un GTID en un archivo exige descargar y analizar sintácticamente todos los registros archivados de OSS. Este proceso consume muchos recursos y tiempo, lo que hace inviables las funciones basadas en desplazamientos GTID. En consecuencia, la función de registros archivados de OSS solo admite iniciar desde una marca de tiempo específica o desde un desplazamiento concreto del archivo de registro binario. No permite comenzar desde un GTID determinado ni admite escenarios con conmutaciones entre nodos primario y secundario en los registros archivados, ya que dichas conmutaciones en MySQL dependen de los GTID. Evalúe detenidamente esta función antes de utilizarla.
  rds.access-key-id	AccessKey ID de la cuenta de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se utiliza la función de lectura de registros archivados desde OSS.	STRING	Ninguno	Para obtener más información, consulte How do I view the AccessKey ID and AccessKey secret?. Importante Para evitar que se filtre su información de AccessKey, utilice la función de gestión de secretos al especificar el AccessKey ID. Para obtener más información, consulte Manage variables.
  rds.access-key-secret	AccessKey secret de la cuenta de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se utiliza la función de lectura de registros archivados desde OSS.	STRING	Ninguno	Para obtener más información, consulte How do I view the AccessKey ID and AccessKey secret? Importante Para evitar que se filtre su información de AccessKey, utilice la función de gestión de secretos al especificar el AccessKey secret. Para obtener más información, consulte Manage variables.
  rds.db-instance-id	Identificador de la instancia de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se utiliza la función de lectura de registros archivados desde OSS.	STRING	Ninguno	Ninguna.
  rds.main-db-id	Número de la base de datos principal de la instancia de Alibaba Cloud ApsaraDB RDS for MySQL.	No	STRING	Ninguno	Para obtener más información sobre cómo consultar el número de la base de datos principal, consulte ApsaraDB RDS for MySQL log backup. Compatible únicamente con VVR 8.0.7 y versiones posteriores. Nota Si no se especifica este parámetro, VVR 11.1 y versiones posteriores consultan automáticamente el número de la base de datos principal a partir de la información de conexión de ApsaraDB RDS for MySQL.
  rds.download.timeout	Tiempo de espera máximo para descargar un registro archivado individual desde OSS.	No	DURATION	60s	Ninguna.
  rds.endpoint	Endpoint del servicio para obtener la información de registros binarios de OSS.	No	STRING	Ninguno	Para obtener más información sobre los valores válidos, consulte Endpoints. Compatible únicamente con VVR 8.0.8 y versiones posteriores.
  scan.incremental.close-idle-reader.enabled	Indica si se cierran los lectores inactivos una vez completada la instantánea.	No	BOOLEAN	false	Compatible únicamente con VVR 8.0.1 y versiones posteriores. Para que esta configuración surta efecto, debe establecer execution.checkpointing.checkpoints-after-tasks-finish.enabled en true.
  scan.read-changelog-as-append-only.enabled	Indica si se convierte el flujo de datos de registro de cambios en un flujo de solo adición.	No	BOOLEAN	false	Valores válidos: true: todos los tipos de mensajes, incluidos INSERT, DELETE, UPDATE_BEFORE y UPDATE_AFTER, se convierten en mensajes INSERT. Habilite esta opción únicamente en escenarios especiales, por ejemplo, cuando necesite conservar los mensajes de eliminación de la tabla ascendente. false (predeterminado): todos los tipos de mensajes se envían al destino descendente tal como se reciben. Nota Compatible únicamente con VVR 8.0.8 y versiones posteriores.
  scan.only.deserialize.captured.tables.changelog.enabled	En la fase incremental, indica si se analizan sintácticamente solo los eventos de cambio de las tablas especificadas.	No	BOOLEAN	El valor predeterminado es false en las versiones VVR 8.x. El valor predeterminado es true en VVR 11.1 y versiones posteriores.	Valores válidos: true: analiza sintácticamente solo los datos de cambio de las tablas de destino para acelerar la lectura del registro binario. false (predeterminado): analiza sintácticamente los datos de cambio de todas las tablas. Nota Compatible únicamente con VVR 8.0.7 y versiones posteriores. Si utiliza VVR 8.0.8 o una versión anterior, debe cambiar el nombre del parámetro a debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	En la fase incremental, indica si se intenta analizar sintácticamente los eventos DDL de cambio sin bloqueo de RDS.	No	BOOLEAN	false	Valores válidos: true: analiza sintácticamente los eventos DDL de cambio sin bloqueo de RDS. false (predeterminado): no analiza sintácticamente los eventos DDL de cambio sin bloqueo de RDS. Esta es una función experimental. Antes de aplicar un cambio sin bloqueo en línea, tome una instantánea del trabajo de Flink para poder recuperarlo. Nota Compatible únicamente con VVR 11.1 y versiones posteriores.
  scan.incremental.snapshot.backfill.skip	Indica si se omite el rellenado retrospectivo durante la fase de lectura de la instantánea.	No	BOOLEAN	false	Valores válidos: true: omite el rellenado retrospectivo durante la fase de lectura de la instantánea. false (predeterminado): no omite el rellenado retrospectivo durante la fase de lectura de la instantánea. El rellenado retrospectivo se aplica únicamente durante la consulta de instantánea de un fragmento individual y no abarca toda la fase de lectura completa. Cuando se omite, la consulta de instantánea de cada fragmento lee los datos más recientes de la tabla en ese instante. Las actualizaciones que ocurren en un fragmento después de haberse leído no se combinan durante la fase de lectura completa; se leen desde el registro binario una vez iniciada la fase incremental. Por ejemplo, una actualización en chunk5 que ocurra mientras se toma su instantánea se refleja directamente en ella. Sin embargo, si chunk5 se actualiza después de que el lector haya avanzado a chunk80, el cambio se aplicará más tarde desde el registro binario durante la fase incremental. Importante Cuando esta opción está habilitada, los cambios que ocurren durante o después del examen de un fragmento se entregan igualmente desde el registro binario en la fase incremental y podrían duplicarse. Solo se garantiza la semántica at-least-once. Habilítela únicamente cuando el destino descendente admita escrituras idempotentes por clave primaria. Nota Compatible únicamente con VVR 11.1 y versiones posteriores.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Indica si se asignan primero los fragmentos ilimitados durante la fase de lectura de la instantánea.	No	BOOELEAN	false	Valores válidos: true: asigna primero los fragmentos ilimitados durante la fase de lectura de la instantánea. false (predeterminado): no asigna primero los fragmentos ilimitados durante la fase de lectura de la instantánea. Esta es una función experimental. Habilitarla reduce el riesgo de errores OOM en TaskManager al sincronizar el último fragmento durante la fase de instantánea. Agregue este parámetro antes del primer inicio del trabajo. Nota Compatible únicamente con VVR 11.1 y versiones posteriores.
  binlog.session.network.timeout	Tiempo de espera de lectura y escritura de red para la conexión del registro binario.	No	DURATION	10m	Si se establece en 0s, se utiliza el tiempo de espera predeterminado del servidor de MySQL. Nota Compatible únicamente con VVR 11.5 y versiones posteriores.
  scan.rate-limit.records-per-second	Limita el número máximo de registros que el origen envía por segundo.	No	LONG	Ninguno	Útil para escenarios donde se necesita limitar la lectura de datos. Este límite se aplica tanto en la fase completa como en la incremental. La métrica numRecordsOutPerSecond del origen refleja la cantidad de registros que emite todo el flujo de datos por segundo. Puede ajustar este parámetro tomando como referencia dicha métrica. Durante la fase de lectura de datos completos, normalmente conviene reducir el número de filas leídas en cada lote. Para ello, disminuya el valor del parámetro scan.incremental.snapshot.chunk.size. Nota Compatible únicamente con VVR 11.5 y versiones posteriores.
  scan.binlog.tolerate.gtid-holes	Habilitar este parámetro permite ignorar los vacíos en la secuencia GTID, de modo que el trabajo omita los eventos discontinuos y continúe ejecutándose.	No	BOOLEAN	false	Antes de habilitar este parámetro, asegúrese de que el desplazamiento inicial del trabajo no haya expirado. Si el trabajo comienza desde un desplazamiento GTID eliminado o expirado, el motor omitirá silenciosamente los registros faltantes, lo que provocará pérdida de datos. Nota Este parámetro es compatible únicamente con VVR 11.6 y versiones posteriores.
  scan.filter-unchanged-row.enabled	Indica si se filtran los eventos de cambio de los campos que no se utilizan en la instrucción SQL.	No	BOOLEAN	false	Si establece este parámetro en true, una tabla de origen CDC comparará únicamente las columnas referenciadas en la instrucción SQL cuando reciba un evento UPDATE. Si los valores de dichas columnas no cambian, el evento no se enviará al destino descendente. Este parámetro resulta útil en escenarios JOIN de múltiples flujos. Cuando una tabla ascendente actualiza un campo no referenciado en la instrucción SQL, no se generan eventos DELETE e INSERT innecesarios. Esto reduce la amplificación de datos y la fluctuación de resultados en los operadores descendentes. Nota Este parámetro es compatible únicamente con VVR 11.8 y versiones posteriores.
  scan.projection-pushdown.enabled	Indica si se habilita la inserción de proyecciones (projection pushdown).	No	BOOLEAN	false	Si establece este parámetro en true, una tabla de origen CDC analizará y transmitirá únicamente las columnas referenciadas en la instrucción SQL. Las columnas no referenciadas no se serializan ni se envían por la red, lo que reduce la sobrecarga de E/S y de cómputo. Este parámetro funciona mejor si se combina con scan.filter-unchanged-row.enabled. La inserción de proyecciones reduce primero el volumen de datos y, después, se filtran las filas sin cambios para aligerar aún más la carga de procesamiento en los operadores descendentes. Nota Este parámetro es compatible únicamente con VVR 11.8 y versiones posteriores.

• Parámetros específicos de tablas de dimensiones

  Parámetro	Descripción	Obligatorio	Tipo de datos	Valor predeterminado	Notas
  url	La URL JDBC de MySQL.	No	STRING	Ninguna	El formato de la URL es: jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	Número máximo de reintentos después de una lectura de datos con errores.	No	INTEGER	3	Compatible solo con VVR 6.0.7 y versiones posteriores.
  lookup.cache.strategy	Política de caché.	No	STRING	Ninguna	Las políticas de caché compatibles son None, LRU y ALL. Para obtener más información sobre los valores, consulte Dimension table JOIN statements. Nota Cuando utilice la política de caché LRU, también debe configurar el parámetro lookup.cache.max-rows.
  lookup.cache.max-rows	Número máximo de filas almacenadas en la caché.	No	INTEGER	100000	Si selecciona la política de caché LRU, debe establecer el tamaño de la caché. Si selecciona la política de caché ALL, no necesita establecer el tamaño de la caché.
  lookup.cache.ttl	Tiempo de vida (TTL) de la caché.	No	DURATION	10 s	La configuración de lookup.cache.ttl depende de lookup.cache.strategy: Si lookup.cache.strategy se establece en None, no necesita configurar lookup.cache.ttl. Esto significa que la caché no caduca. Si lookup.cache.strategy se establece en LRU, lookup.cache.ttl representa el TTL de la caché. De forma predeterminada, la caché no caduca. Si lookup.cache.strategy se establece en ALL, lookup.cache.ttl indica el tiempo de carga de la caché. De forma predeterminada, la caché no se vuelve a cargar. Utilice un formato de tiempo, por ejemplo, 1min o 10s.
  lookup.max-join-rows	Número máximo de resultados devueltos cuando un registro de la tabla principal coincide con registros de la tabla de dimensiones.	No	INTEGER	1024	Ninguna.
  lookup.filter-push-down.enabled	Especifica si se habilita el pushdown de filtros para la tabla de dimensiones.	No	BOOLEAN	false	Valores válidos: true: habilita el pushdown de filtros para la tabla de dimensiones. Al cargar datos desde la tabla de la base de datos MySQL, la tabla de dimensiones filtra los datos previamente según las condiciones establecidas en el trabajo SQL. false (predeterminado): deshabilita el pushdown de filtros para la tabla de dimensiones. Al cargar datos desde la tabla de la base de datos MySQL, la tabla de dimensiones carga todos los datos. Nota Compatible solo con VVR 8.0.7 y versiones posteriores. Importante El pushdown de la tabla de dimensiones solo debe habilitarse cuando una tabla de Flink se utiliza como tabla de dimensiones. Las tablas de origen de MySQL no admiten habilitar el pushdown de filtros. Si una tabla de Flink se usa tanto como tabla de origen como de dimensiones, y se habilita el pushdown de filtros para la tabla de dimensiones, debe establecer explícitamente este elemento de configuración en false para la tabla de origen mediante SQL Hints. De lo contrario, el trabajo podría ejecutarse de forma anómala.
  lookup.cache.cache-empty	Especifica si se almacenan en la caché los resultados de consulta vacíos.	No	BOOLEAN	true	Surte efecto únicamente cuando la política de caché es LRU. Valores válidos: true (predeterminado): si una consulta no devuelve datos, se almacena un resultado vacío en la caché y se devuelve directamente para las consultas posteriores. false: si una consulta no devuelve datos, no se almacena ningún resultado en la caché y la siguiente consulta vuelve a buscar los datos. Nota Este parámetro es compatible solo con VVR 11.9.0 y versiones posteriores.

• Solo para tablas sink

  Parámetro	Descripción	Obligatorio	Tipo de datos	Valor predeterminado	Notas
  url	La URL JDBC de MySQL.	No	STRING	Ninguna	El formato de la URL es: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	Número máximo de reintentos después de una escritura de datos con errores.	No	INTEGER	3	Ninguna.
  sink.buffer-flush.batch-size	Número de filas en una sola escritura por lotes.	No	INTEGER	4096	Ninguna.
  sink.buffer-flush.max-rows	Número de filas de datos almacenadas en la memoria.	No	INTEGER	10000	Este parámetro surte efecto únicamente después de especificar una clave principal.
  sink.buffer-flush.interval	Intervalo para vaciar la caché. Si los datos almacenados no cumplen las condiciones de salida tras el tiempo de espera especificado, el sistema envía automáticamente todos los datos de la caché.	No	DURATION	1s	Ninguna.
  sink.ignore-delete	Especifica si se ignoran las operaciones DELETE de datos.	No	BOOLEAN	false	Cuando el flujo generado por Flink SQL incluye registros de tipo delete o update-before, pueden producirse inconsistencias de datos si varias tareas de salida actualizan simultáneamente distintos campos de la misma tabla. Por ejemplo, si después de eliminar un registro otra tarea actualiza solo algunos campos, los campos no actualizados quedarán como null o con sus valores predeterminados, lo que provocará errores en los datos. Al establecer sink.ignore-delete en true, puede ignorar las operaciones DELETE y UPDATE_BEFORE del flujo de entrada para evitar estos problemas. Nota UPDATE_BEFORE forma parte del mecanismo de retracción de Flink y se utiliza para "retractar" el valor anterior durante una operación de actualización. Cuando ignoreDelete = true, se omiten todos los registros de tipo DELETE y UPDATE_BEFORE. Solo se procesan los registros INSERT y UPDATE_AFTER.
  sink.ignore-delete-mode	Estrategia para gestionar los registros de tipo delete después de ignorar las operaciones DELETE.	No	STRING	ALL	Valores válidos: ALL: ignora tanto los registros -D como los -U. REAL_DELETE: ignora únicamente los registros -D. UPDATE_BEFORE: ignora únicamente los registros -U. Nota Esta opción es compatible solo con el motor Realtime Compute VVR 11.8 y versiones posteriores. Surte efecto únicamente cuando sink.ignore-delete=true. Configurarla de forma aislada genera un error.
  sink.ignore-null-when-update	Al actualizar datos, especifica si el campo correspondiente se establece en null o si se omite su actualización cuando el valor del campo entrante es null.	No	BOOLEAN	false	Valores válidos: true: no actualiza el campo. Este parámetro solo puede establecerse en true cuando la tabla de Flink tiene definida una clave principal. Cuando se establece en true: En VVR 8.0.6 y versiones anteriores, la tabla sink no admite escritura por lotes. En VVR 8.0.7 y versiones posteriores, la tabla sink admite escritura por lotes. La escritura por lotes mejora significativamente la eficiencia de escritura y el rendimiento general, pero introduce latencia en los datos y riesgo de errores OOM. Por lo tanto, debe evaluar esta compensación según su escenario de negocio. false: actualiza el campo a null. Nota Este parámetro es compatible solo con VVR 8.0.5 y versiones posteriores.
  sink.force-batch-on-non-primary-table	Especifica si se almacenan en búfer los registros de forma obligatoria para insertarlos por lotes mediante JDBC Batch al escribir datos en una tabla sink de MySQL sin clave principal.	No	BOOLEAN	false	Este parámetro surte efecto únicamente en tablas sin clave principal. Si lo establece en true, los eventos DELETE y UPDATE_BEFORE se descartan silenciosamente, mientras que los eventos INSERT y UPDATE_AFTER se escriben por lotes como instrucciones INSERT. Se recomienda habilitar este parámetro solo si los datos son de tipo append-only o si la semántica descrita le resulta aceptable. Nota Este parámetro es compatible solo con VVR 11.8 y versiones posteriores. Habilite este parámetro únicamente si los datos son de tipo append-only o si la semántica descrita le resulta aceptable. De lo contrario, los datos podrían volverse inconsistentes.

Mapeo de tipos

• Tablas de origen CDC

  Tipo de campo de MySQL CDC	Tipo de campo de Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  TINYINT UNSIGNED ZEROFILL
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  SMALLINT UNSIGNED ZEROFILL
  BIGINT	BIGINT
  INT UNSIGNED
  INT UNSIGNED ZEROFILL
  MEDIUMINT UNSIGNED
  MEDIUMINT UNSIGNED ZEROFILL
  BIGINT UNSIGNED	DECIMAL(20, 0)
  BIGINT UNSIGNED ZEROFILL
  SERIAL
  FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
  DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
  DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
  REAL [UNSIGNED] [ZEROFILL]
  NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
  DECIMAL(p, s) [UNSIGNED] [ZEROFILL]
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]	TIMESTAMP [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  Importante

  No utilice el tipo TINYINT(1) en MySQL para almacenar valores distintos de 0 y 1. Cuando property-version=0, la tabla de origen de MySQL CDC asigna TINYINT(1) al tipo BOOLEAN en Flink de forma predeterminada. Esto puede provocar imprecisiones en los datos. Para usar el tipo TINYINT(1) con valores distintos de 0 y 1, consulte el parámetro de configuración catalog.table.treat-tinyint1-as-boolean.

• Tablas de dimensiones y tablas sink

  Tipo de campo de MySQL	Tipo de campo de Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) Nota donde p <= 38.
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  CHAR(n)	CHAR(n)
  VARCHAR(n)	VARCHAR(n)
  BIT(n)	BINARY(⌈n/8⌉)
  BINARY(n)	BINARY(n)
  VARBINARY(N)	VARBINARY(N)
  TINYTEXT	STRING
  TEXT
  MEDIUMTEXT
  LONGTEXT
  TINYBLOB	BYTES Importante Flink solo admite registros del tipo BLOB de MySQL cuyo tamaño sea menor o igual a 2,147,483,647 (2^31 - 1) bytes.
  BLOB
  MEDIUMBLOB
  LONGBLOB

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium configuration
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Enable parsing filter
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Parse only the change events of specified tables.
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium configuration
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Enable parsing filter
  scan.only.deserialize.captured.tables.changelog.enabled: true

Diagnosticar la latencia de datos para optimizar el rendimiento del trabajo

Si experimenta latencia de datos durante la fase incremental, analice el problema con los siguientes pasos:

1. Compruebe las métricas currentFetchEventTimeLag y currentEmitEventTimeLag en la página Overview. La métrica currentFetchEventTimeLag representa la latencia al leer datos del registro binario. Por su parte, la métrica currentEmitEventTimeLag representa la latencia al leer del registro binario los datos de las tablas relevantes para el trabajo.

   Escenario	Descripción
   currentFetchEventTimeLag es bajo, mientras que currentEmitEventTimeLag es alto y rara vez se actualiza.	Un valor bajo de currentFetchEventTimeLag indica que la extracción del registro binario de la base de datos es eficiente. Sin embargo, dicho registro contiene pocos datos de las tablas que necesita leer el trabajo, por lo que currentEmitEventTimeLag rara vez se actualiza. Este comportamiento es normal.
   Tanto currentFetchEventTimeLag como currentEmitEventTimeLag son altos.	Indica que la tabla de origen tiene un rendimiento de lectura deficiente. Continúe con los pasos posteriores de esta sección para optimizarlo.

2. La contrapresión puede reducir la velocidad a la que el origen envía datos a los operadores posteriores. Es posible que observe que sourceIdleTime aumenta periódicamente y que tanto currentFetchEventTimeLag como currentEmitEventTimeLag crecen de forma continua. Para resolverlo, aumente el paralelismo del nodo donde se origina la contrapresión.

3. Revise la métrica TM CPU Usage en la página CPU y la métrica TM GC Time en la página JVM para determinar si faltan recursos de CPU o memoria. Puede aumentar los recursos del trabajo para optimizar el rendimiento de lectura o habilitar los parámetros de minilotes para mejorar el rendimiento. Para obtener más información, consulte High-Performance Flink SQL Optimization Techniques.

4. Si el trabajo incluye un operador SinkUpsertMaterializer con un estado grande, el rendimiento de lectura se puede ver afectado. Considere aumentar el paralelismo del trabajo o evitar el operador SinkUpsertMaterializer. Para obtener más información, consulte Avoid using SinkUpsertMaterializer. Quitar este operador de un trabajo existente requiere un reinicio sin estado, ya que cambia la topología del trabajo e iniciarlo desde el estado actual podría provocar errores o pérdida de datos.

Optimización de los parámetros de fragmentos

Los datos y los metadatos de los fragmentos se almacenan en memoria, lo que a veces provoca errores de falta de memoria (OOM). Ajuste los parámetros según el componente que presente el error:

• JobManager

  El JobManager almacena los metadatos de todos los fragmentos. Un número excesivo de fragmentos puede causar un error OOM. Para resolverlo, aumente el valor de scan.incremental.snapshot.chunk.size y reduzca así la cantidad de fragmentos. También puede aumentar la memoria de montón del JobManager si establece jobmanager.memory.heap.size en la configuración de tiempo de ejecución. Para obtener más información, consulte Configuración de parámetros de Flink.

• TaskManager

  • El TaskManager lee los datos de cada fragmento. Si un fragmento contiene demasiadas filas, puede producirse un error OOM. Para solucionarlo, disminuya el valor de scan.incremental.snapshot.chunk.size y reduzca así el número de filas por fragmento. Otra opción es aumentar la memoria de montón del TaskManager si incrementa el valor de TaskManager Memory en la configuración de tiempo de ejecución.

  • En VVR 8.0.8 y versiones anteriores, el último fragmento puede contener una gran cantidad de datos y provocar un error OOM en el TaskManager. Se recomienda actualizar a VVR 8.0.9 o posterior para evitar este problema.

  • En una tabla de origen de MySQL CDC con clave principal compuesta, los fragmentos se dividen de forma predeterminada según la primera columna de dicha clave. Si los datos están muy sesgados y muchas filas comparten el mismo valor en esa columna, el fragmento correspondiente puede volverse demasiado grande y causar un error OOM en el TaskManager. Configure scan.incremental.snapshot.chunk.key-column para especificar otra columna de la clave principal como criterio de división.

SET 'table.optimizer.source-merge.enabled' = 'true';

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    'rds.region-id' = 'cn-beijing',
    'rds.access-key-id' = 'your_access_key_id',
    'rds.access-key-secret' = 'your_access_key_secret',
    'rds.db-instance-id' = 'rm-xxxxxxxx',  // The ID of the database instance.
    'rds.main-db-id' = '12345678', // The ID of the primary database.
    'rds.endpoint' = 'rds.aliyuncs.com'
    ...
)