Conector YAML de MySQL - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

El conector de MySQL se puede usar como origen de datos en los trabajos YAML de ingesta de datos.

Requisitos previos

Antes de usar una tabla de origen CDC de MySQL, complete las operaciones descritas en Configure MySQL.

ApsaraDB RDS for MySQL

Ejecute un network probe para comprobar la conectividad de red con Realtime Compute for Apache Flink.
Versión de MySQL: 5.6, 5.7, 8.0.x o 8.4.
El registro binario debe estar habilitado. Esta opción está habilitada de forma predeterminada.
El formato del registro binario debe ser ROW. Este es el formato predeterminado.
El parámetro binlog_row_image debe tener el valor FULL. Esta es la configuración predeterminada.
La compresión de transacciones del registro binario (Binary Log Transaction Compression) debe estar deshabilitada. Esta función se introdujo en MySQL 8.0.20 y está deshabilitada de forma predeterminada.
Se requiere un usuario de MySQL creado con los permisos SELECT, SHOW DATABASES, REPLICATION SLAVE y REPLICATION CLIENT.
Cree una base de datos y una tabla de MySQL. Para obtener más información, consulte Create a database and an account for an ApsaraDB RDS for MySQL instance. Utilice una cuenta con privilegios para crear la base de datos de MySQL y evitar errores por permisos insuficientes.
Configure una lista blanca de direcciones IP. Para obtener más información, consulte Configure an IP address whitelist for an ApsaraDB RDS for MySQL instance.

PolarDB for MySQL

Ejecute un network probe para comprobar la conectividad de red con Realtime Compute for Apache Flink.
Versión de MySQL: 5.6, 5.7, 8.0.x o 8.4.
El registro binario debe estar habilitado. Esta opción está deshabilitada de forma predeterminada.
El formato del registro binario debe ser ROW. Este es el formato predeterminado.
El parámetro binlog_row_image debe tener el valor FULL. Esta es la configuración predeterminada.
La compresión de transacciones del registro binario (Binary Log Transaction Compression) debe estar deshabilitada. Esta función se introdujo en MySQL 8.0.20 y está deshabilitada de forma predeterminada.
Se requiere un usuario de MySQL creado con los permisos SELECT, SHOW DATABASES, REPLICATION SLAVE y REPLICATION CLIENT.
Cree una base de datos y una tabla de MySQL. Para obtener más información, consulte Create a database and an account for a PolarDB for MySQL cluster. Utilice una cuenta con privilegios para crear la base de datos de MySQL y evitar errores por permisos insuficientes.
Configure una lista blanca de direcciones IP. Para obtener más información, consulte Configure an IP address whitelist for a PolarDB for MySQL cluster.

Self-managed MySQL

Ejecute un network probe para comprobar la conectividad de red con Realtime Compute for Apache Flink.
Versión de MySQL: 5.6, 5.7, 8.0.x o 8.4.
El registro binario debe estar habilitado. Esta opción está deshabilitada de forma predeterminada.
El formato del registro binario debe ser ROW. El formato predeterminado es STATEMENT.
El parámetro binlog_row_image debe tener el valor FULL. Esta es la configuración predeterminada.
La compresión de transacciones del registro binario (Binary Log Transaction Compression) debe estar deshabilitada. Esta función se introdujo en MySQL 8.0.20 y está deshabilitada de forma predeterminada.
Cree un usuario de MySQL y otórguele los permisos SELECT, SHOW DATABASES, REPLICATION SLAVE y REPLICATION CLIENT.
Cree una base de datos y una tabla de MySQL. Para obtener más información, consulte Crear una base de datos y una cuenta para una instancia de MySQL autogestionada. Utilice una cuenta con privilegios para crear la base de datos de MySQL y evitar errores por permisos insuficientes.
Configure una lista blanca de direcciones IP. Para obtener más información, consulte Configure an IP address whitelist for a self-managed MySQL instance.

Límites

Límites generales

El conector CDC de MySQL no admite la función Binary Log Transaction Compression. Por lo tanto, cuando use este conector para consumir datos incrementales, asegúrese de que dicha función esté deshabilitada. De lo contrario, el conector podría fallar al recuperar los datos incrementales.

Límites de ApsaraDB RDS for MySQL

En ApsaraDB RDS for MySQL, no lea datos de una base de datos secundaria ni de una réplica de solo lectura. El periodo de retención predeterminado de los registros binarios en estos recursos es corto. Si los registros binarios caducan y se eliminan, el trabajo podría fallar al consumirlos y generar un error.
ApsaraDB RDS for MySQL habilita de forma predeterminada la sincronización paralela entre instancias principal y secundaria, pero no garantiza un orden de transacciones consistente entre ellas. Esto puede provocar la pérdida de datos durante un cambio de rol entre instancias y la recuperación desde un punto de control. Para evitar este problema, habilite manualmente la opción slave_preserve_commit_order en ApsaraDB RDS for MySQL.

Límites de PolarDB for MySQL

Las tablas de origen CDC de MySQL no admiten la lectura de datos de clústeres con arquitectura Multi-master Cluster en PolarDB for MySQL V1.0.19 y versiones anteriores. Para obtener más información, consulte What is a Multi-master Cluster?. Los registros binarios generados por estos clústeres pueden contener identificadores de tabla duplicados. Esto provoca errores de mapeo de esquemas en la tabla de origen CDC y, en consecuencia, fallos al analizar sintácticamente los datos del registro binario.

Límites de MySQL de código abierto

De forma predeterminada, MySQL mantiene el orden de las transacciones durante la replicación de registros binarios entre la instancia principal y las secundarias. Si una réplica de MySQL tiene habilitada la replicación paralela (slave_parallel_workers > 1), pero no tiene habilitada la opción slave_preserve_commit_order=ON, su orden de confirmación de transacciones podría diferir del de la base de datos principal. Cuando Flink CDC se recupera desde un punto de control, puede omitir datos debido a esta secuencia desordenada. Establezca slave_preserve_commit_order = ON en la réplica de MySQL. Como alternativa, establezca slave_parallel_workers = 1, aunque esto reduce el rendimiento de la replicación.

Notas de uso

Set a server ID to avoid binary log consumption conflicts.
Durante la fase de lectura completa de datos, no es posible guardar un savepoint, agregar una tabla nueva a la tabla de origen, eliminar una tabla existente y luego reiniciar el trabajo desde ese savepoint. Estas acciones provocan que el trabajo falle al leer los datos.

Ingesta de datos

El conector de MySQL se puede utilizar como origen de datos en un trabajo YAML de ingesta de datos.

Sintaxis

source:
   type: mysql
   name: MySQL Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: xxx

Elementos de configuración

Parámetro	Descripción	Obligatorio	Tipo de dato	Valor predeterminado	Notas
type	El tipo de origen de datos.	Sí	STRING	Ninguna	El valor debe ser mysql.
name	El nombre del origen de datos.	No	STRING	Ninguna	Ninguna.
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 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.
tables	Las tablas de datos de MySQL que se sincronizarán.	Sí	STRING	Ninguna	Este parámetro admite expresiones regulares para leer datos de varias tablas. Utilice comas para separar varias expresiones regulares. Nota No use los caracteres de coincidencia de inicio de cadena `^` ni de fin de cadena `$` en la expresión regular. En VVR 11.2, el punto se usa para dividir la expresión regular y obtener la parte de la base de datos. Los caracteres de inicio y fin inutilizan la expresión regular resultante. Por ejemplo, debe cambiar `^db.user_[0-9]+$` por `db.user_[0-9]+`. El punto separa el nombre de la base de datos del nombre de la tabla. Para usar un punto como comodín de cualquier carácter, escápelo con una barra invertida. Por ejemplo: db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\..
tables.exclude	Las tablas que se excluirán de la sincronización.	No	STRING	Ninguna	Este parámetro admite expresiones regulares para excluir varias tablas. Utilice comas para separar varias expresiones regulares. Nota El punto separa el nombre de la base de datos del nombre de la tabla. Para usar un punto como comodín de cualquier carácter, escápelo con una barra invertida. Por ejemplo: db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\..
port	El número de puerto del servicio de base de datos MySQL.	No	INTEGER	3306	Ninguna.
schema-change.enabled	Indica si se envían eventos de cambio de esquema.	No	BOOLEAN	true	Ninguna.
server-id	Un identificador numérico o un rango para el cliente de base de datos que se usa en la sincronización.	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, como 5400-5408. Nota Cuando se habilita la lectura incremental, se admite la lectura concurrente. En este caso, se recomienda establecer un rango de identificadores para que cada lector concurrente use uno distinto.
jdbc.properties.*	Parámetros de conexión personalizados en la URL de JDBC.	No	STRING	Ninguna	Puede pasar parámetros de conexión personalizados. Por ejemplo, para no usar 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	Ninguna	Puede pasar parámetros personalizados de Debezium. Por ejemplo, use 'debezium.event.deserialization.failure.handling.mode'='ignore' para especificar 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 puede provocar que el conector lea los datos de manera incorrecta. Por ejemplo, no se permite configurar el parámetro debezium.binlog.buffer.size.
scan.incremental.snapshot.chunk.size	El tamaño de cada fragmento, expresado en número de filas.	No	INTEGER	8096	Las tablas de MySQL se dividen 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 reduce la granularidad de la recuperación ante errores, puede provocar errores OOM y disminuir el rendimiento general. Por lo tanto, evalúe el equilibrio adecuado y establezca un tamaño de fragmento razonable.
scan.snapshot.fetch.size	El número máximo de registros que se extraen a la vez al leer todos los datos de una tabla.	No	INTEGER	1024	Ninguna.
scan.startup.mode	El 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 lee los cambios más recientes generados después de su arranque. earliest-offset: el conector no examina los datos históricos. Comienza a leer desde el registro binario disponible más antiguo. specific-offset: el conector no examina los datos históricos. Inicia desde un desplazamiento específico del registro binario. Especifique el desplazamiento mediante la configuración conjunta de scan.startup.specific-offset.file y scan.startup.specific-offset.pos, o bien configure únicamente scan.startup.specific-offset.gtid-set para iniciar desde un conjunto GTID específico. timestamp: el conector no examina los datos históricos. Comienza a leer el registro binario desde una marca de tiempo específica, indicada en milisegundos mediante scan.startup.timestamp-millis. Importante En los modos de inicio earliest-offset, specific-offset y timestamp, si el esquema de la tabla al momento del inicio difiere del esquema en la posición de inicio especificada, el trabajo reportará un error por discrepancia de esquemas. Es decir, al usar estos tres modos, asegúrese de que el esquema de la tabla correspondiente no cambie entre la posición de consumo del registro binario y el momento de inicio del trabajo.
scan.startup.specific-offset.file	El nombre del archivo de registro binario para el desplazamiento inicial cuando se usa el modo de inicio specific-offset.	No	STRING	Ninguna	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset. Formato de ejemplo del nombre de archivo: `mysql-bin.000003`.
scan.startup.specific-offset.pos	El desplazamiento dentro del archivo de registro binario especificado para el punto de inicio cuando se usa el modo specific-offset.	No	INTEGER	Ninguna	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset.
scan.startup.specific-offset.gtid-set	El conjunto GTID para el desplazamiento inicial cuando se usa el modo de inicio specific-offset.	No	STRING	Ninguna	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset. Formato de ejemplo del conjunto GTID: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	La marca de tiempo en milisegundos para el desplazamiento inicial cuando se usa el modo de inicio timestamp.	No	LONG	Ninguna	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 de registro binario correspondiente a la hora indicada. Asegúrese de que el archivo de registro binario asociado a la marca de tiempo especificada no se haya eliminado de la base de datos y pueda leerse.
server-time-zone	La zona horaria de sesión que usa la base de datos.	No	STRING	Si no especifica este parámetro, el sistema usa la zona horaria del entorno de ejecución del trabajo de Flink como zona horaria del servidor de base de datos. Esta corresponde a la zona horaria de la región que seleccionó.	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.
scan.startup.specific-offset.skip-events	El número de eventos del registro binario que se omiten al leer desde un desplazamiento específico.	No	INTEGER	Ninguna	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset.
scan.startup.specific-offset.skip-rows	El número de cambios de fila que se omiten al leer desde un desplazamiento específico. Un solo evento del registro binario puede corresponder a varios cambios de fila.	No	INTEGER	Ninguna	Al usar este parámetro, debe establecer scan.startup.mode en specific-offset.
connect.timeout	El tiempo máximo de espera antes de que expire la conexión al servidor de base de datos MySQL y se reintente.	No	DURATION	30 s	Ninguna.
connect.max-retries	El número máximo de reintentos tras un error de conexión al servicio de base de datos MySQL.	No	INTEGER	3	Ninguna.
connection.pool.size	El tamaño del pool de conexiones de la base de datos.	No	INTEGER	20	El pool de conexiones de la base de datos permite reutilizar conexiones, lo que reduce la cantidad de conexiones activas.
heartbeat.interval	El intervalo con el que el origen avanza el desplazamiento del registro binario mediante eventos de latido.	No	DURATION	30s	Los eventos de latido sirven para avanzar el desplazamiento del registro binario en el origen. Esto resulta muy útil para las tablas de MySQL que se actualizan con poca frecuencia, ya que en ellas el desplazamiento no avanza automáticamente. Los eventos de latido empujan el desplazamiento hacia adelante y evitan problemas derivados de un desplazamiento caducado. Un desplazamiento caducado puede hacer que el trabajo falle sin posibilidad de recuperación, lo que obliga a reiniciarlo sin estado.
rds.region-id	El identificador de región de la instancia de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se usa la función para leer registros archivados desde OSS.	STRING	Ninguna	Para obtener más información sobre los identificadores de región, consulte Regions and zones. Importante Dado que 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, localizar un GTID en un archivo requiere descargar y analizar sintácticamente todos los registros archivados de OSS. Este proceso consume muchos recursos y tiempo, lo que hace inviables las funciones que dependen de desplazamientos GTID. Por ello, 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 admite iniciar desde un GTID determinado ni escenarios con conmutaciones entre nodos primario y secundario en los registros archivados, porque dichas conmutaciones en MySQL dependen de los GTID. Evalúe esta función cuidadosamente antes de usarla.
rds.access-key-id	El AccessKey ID de la cuenta de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se usa la función para leer registros archivados desde OSS.	STRING	Ninguna	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 para especificar el AccessKey ID. Para obtener más información, consulte Manage variables.
rds.access-key-secret	El AccessKey secret de la cuenta de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se usa la función para leer registros archivados desde OSS.	STRING	Ninguna	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 para especificar el AccessKey secret. Para obtener más información, consulte Manage variables.
rds.db-instance-id	El identificador de la instancia de Alibaba Cloud ApsaraDB RDS for MySQL.	Obligatorio cuando se usa la función para leer registros archivados desde OSS.	STRING	Ninguna	Ninguna.
rds.main-db-id	El número de la base de datos principal de la instancia de Alibaba Cloud ApsaraDB RDS for MySQL.	No	STRING	Ninguna	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. Nota Si no se especifica este parámetro, VVR 11.7 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	El tiempo de espera máximo para descargar un solo registro archivado desde OSS.	No	DURATION	60s	Ninguna.
rds.endpoint	El endpoint del servicio para obtener la información de registros binarios de OSS.	No	STRING	Ninguna	Para conocer los valores válidos, consulte Endpoints.
rds.binlog-directory-prefix	El prefijo del directorio donde se almacenan los archivos de registro binario.	No	STRING	rds-binlog-	Ninguna.
rds.use-intranet-link	Indica si se usa la red interna para descargar los archivos de registro binario.	No	BOOLEAN	true	Ninguna.
rds.binlog-directories-parent-path	La ruta absoluta del directorio principal donde se almacenan los archivos de registro binario.	No	STRING	Ninguna	Ninguna.
chunk-meta.group.size	El tamaño de los metadatos del fragmento.	No	INTEGER	1000	Si los metadatos superan este valor, se dividen en varias partes para su transmisión.
chunk-key.even-distribution.factor.lower-bound	El límite inferior del factor de distribución de fragmentos para la fragmentación uniforme.	No	DOUBLE	0.05	Si el factor de distribución es menor que este valor, se aplica una fragmentación no uniforme. Factor de distribución de fragmentos = (MAX(chunk-key) - MIN(chunk-key) + 1) / Número total de filas de datos.
chunk-key.even-distribution.factor.upper-bound	El límite superior del factor de distribución de fragmentos para la fragmentación uniforme.	No	DOUBLE	1000.0	Si el factor de distribución es mayor que este valor, se aplica una fragmentación no uniforme. Factor de distribución de fragmentos = (MAX(chunk-key) - MIN(chunk-key) + 1) / Número total de filas de datos.
scan.incremental.close-idle-reader.enabled	Indica si se cierran los lectores inactivos una vez completada la instantánea.	No	BOOLEAN	false	Para que esta configuración surta efecto, debe establecer `execution.checkpointing.checkpoints-after-tasks-finish.enabled` en true.
scan.only.deserialize.captured.tables.changelog.enabled	En la fase incremental, indica si se deserializan únicamente 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: deserializa solo los datos de cambio de las tablas de destino para acelerar la lectura del registro binario. false (predeterminado): deserializa los datos de cambio de todas las tablas.
scan.parallel-deserialize-changelog.enabled	En la fase incremental, indica si se usan varios hilos para analizar sintácticamente los eventos de cambio.	No	BOOLEAN	false	Valores válidos: true: utiliza varios hilos en la fase de deserialización de eventos de cambio y mantiene el orden de los eventos del registro binario para acelerar la lectura. false (predeterminado): utiliza un solo hilo en la fase de deserialización de eventos. Nota Compatible solo con VVR 8.0.11 y versiones posteriores.
scan.parallel-deserialize-changelog.handler.size	El número de controladores de eventos cuando se usan varios hilos para analizar sintácticamente los eventos de cambio.	No	INTEGER	2	Nota Compatible solo con VVR 8.0.11 y versiones posteriores.
metadata-column.include-list	Las columnas de metadatos que se transmitirán al flujo descendente.	No	STRING	Ninguna	Los metadatos disponibles incluyen `op_ts`, `es_ts`, `query_log`, `file` y `pos`. Utilice comas para separar varias columnas de metadatos. Nota El conector YAML de MySQL CDC no requiere ni admite agregar columnas de metadatos de nombre de base de datos, nombre de tabla y `op_type`. Puede usar directamente `__data_event_type__` en una expresión Transform para obtener el tipo de dato de cambio, o bien emplear `__schema_name__` y `__table_name__` para obtener el nombre de la base de datos y de la tabla. Importante La columna de metadatos `file` representa el archivo de registro binario donde se encuentran los datos. Su valor es "" durante la fase completa y el nombre del archivo de registro binario durante la fase incremental. La columna de metadatos `pos` representa el desplazamiento de los datos dentro del archivo de registro binario. Su valor es "0" durante la fase completa y el desplazamiento real durante la fase incremental. Estas dos columnas de metadatos son compatibles a partir de VVR 11.5. La columna de metadatos `es_ts` representa la hora de inicio de la transacción correspondiente al registro de cambios en MySQL. Solo es compatible con MySQL 8.0.x. No agregue esta columna de metadatos si usa versiones anteriores de MySQL. La marca de tiempo `op_ts` tiene precisión de segundos, mientras que `es_ts` tiene precisión de milisegundos.
scan.newly-added-table.enabled	Al reiniciar desde un checkpoint, indica si se sincronizan las tablas recién agregadas que no coincidieron en el inicio anterior, o si se quitan del estado las tablas que ya no coinciden.	No	BOOLEAN	false	Esta opción surte efecto al reiniciar desde un checkpoint o un savepoint. Importante Durante la fase de lectura completa de datos no es posible guardar un savepoint, agregar una tabla nueva o eliminar una tabla del origen y luego reiniciar el trabajo desde ese savepoint. Hacerlo provocará que el trabajo no pueda leer los datos.
scan.binlog.newly-added-table.enabled	En la fase incremental, indica si se envían los datos de las tablas recién agregadas que coincidan.	No	BOOLEAN	false	No se puede habilitar al mismo tiempo que `scan.newly-added-table.enabled`.
scan.incremental.snapshot.chunk.key-column	Especifica una columna de ciertas tablas para usarla como columna de división durante la fragmentación en la fase de instantánea.	No	STRING	Ninguna	Use dos puntos `:` para unir el nombre de la tabla y el de la columna al definir una regla. El nombre de la tabla puede ser una expresión regular. Defina varias reglas separándolas con punto y coma `;`. Por ejemplo: `db1.user_table_[0-9]+:col1;db[1-2].[app\|web]_order_\\.*:col2`. Es obligatorio para las tablas sin clave primaria, y la columna elegida debe ser de un tipo no nulo (NOT NULL). Para las tablas con clave primaria es opcional, y solo se puede seleccionar una columna de dicha clave.
scan.parse.online.schema.changes.enabled	En la fase incremental, indica si se intenta analizar sintácticamente los eventos DDL de cambios sin bloqueo de RDS.	No	BOOLEAN	false	Valores válidos: true: analiza sintácticamente los eventos DDL de cambios sin bloqueo de RDS. false (predeterminado): no analiza sintácticamente los eventos DDL de cambios 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 solo con VVR 11.0 y versiones posteriores.
scan.incremental.snapshot.backfill.skip	Indica si se omite el rellenado durante la fase de lectura de la instantánea.	No	BOOLEAN	false	Valores válidos: true: omite el rellenado durante la fase de lectura de la instantánea. false (predeterminado): no omite el rellenado durante la fase de lectura de la instantánea. El rellenado solo se aplica 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 ocurran en un fragmento después de haberse leído no se combinan durante la fase de lectura completa, sino que se leen del registro binario al entrar en la fase incremental. Por ejemplo, una actualización en chunk5 que ocurra mientras se toma su instantánea se refleja directamente en ella; si chunk5 se actualiza después de que el lector haya avanzado a chunk80, la actualización se aplicará más tarde desde el registro binario durante la fase incremental. Importante Cuando esta opción está habilitada, los cambios que ocurran durante o después de la exploración de un fragmento seguirán entregándose 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 sink de destino admita escrituras idempotentes por clave primaria. Nota Compatible solo con VVR 11.1 y versiones posteriores.
treat-tinyint1-as-boolean.enabled	Indica si el tipo TINYINT(1) se trata como un tipo booleano.	No	BOOLEAN	true	Valores válidos: true (predeterminado): trata el tipo TINYINT(1) como un tipo booleano. false: no trata el tipo TINYINT(1) como un tipo booleano.
treat-timestamp-as-datetime-enabled	Indica si el tipo TIMESTAMP se trata como un tipo DATETIME.	No	BOOLEAN	false	Valores válidos: true: trata el tipo TIMESTAMP de MySQL como un tipo DATETIME y lo asigna al tipo TIMESTAMP de CDC. false (predeterminado): asigna el tipo TIMESTAMP de MySQL al tipo TIMESTAMP_LTZ de CDC. El tipo TIMESTAMP de MySQL almacena la hora UTC y se ve afectado por la zona horaria. El tipo DATETIME de MySQL almacena la hora literal y no se ve afectado por la zona horaria. Cuando se habilita, este parámetro convierte los datos del tipo TIMESTAMP de MySQL al tipo DATETIME con base en server-time-zone.
include-comments.enabled	Indica si se sincronizan los comentarios de tablas y columnas.	No	BOOELEAN	false	Valores válidos: true: sincroniza los comentarios de tablas y columnas. false (predeterminado): no sincroniza los comentarios de tablas y columnas. Habilitar esta opción incrementa el uso de memoria del trabajo.
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 solo con VVR 11.1 y versiones posteriores.
binlog.session.network.timeout	El tiempo de espera de red para la conexión del registro binario.	No	DURATION	10m	Si se establece en 0s, se usa el tiempo de espera predeterminado del servidor de MySQL. Nota Compatible solo 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	Ninguna	Aplica a escenarios donde es necesario limitar la lectura de datos. Este límite es efectivo tanto en la fase completa como en la incremental. La métrica `numRecordsOutPerSecond` del origen refleja el número de registros que emite todo el flujo de datos por segundo. Ajuste este parámetro con base en dicha métrica. En la fase de lectura completa de datos, generalmente es necesario 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 solo con VVR 11.5 y versiones posteriores.
include-binlog-meta.enable	Indica si se incluye en el mensaje la información original del registro binario de MySQL, como el GTID y el desplazamiento.	No	Boolean	false	Aplica a escenarios de sincronización del registro binario original, como reemplazar un enlace de sincronización de Canal existente. Nota Compatible solo con VVR 11.6 y versiones posteriores.
scan.binlog.tolerate.gtid-holes	Habilitar este parámetro hace que se ignoren los vacíos en la secuencia de GTID, lo que permite al trabajo saltarse eventos discontinuos y continuar ejecutándose.	No	Boolean	false	Antes de habilitar este parámetro, asegúrese de que el desplazamiento inicial del trabajo no haya caducado. Si el trabajo inicia desde un desplazamiento GTID eliminado o caducado, el motor omitirá silenciosamente los registros faltantes, lo que provocará pérdida de datos. Nota Este parámetro es compatible solo con VVR 11.6 y versiones posteriores.
scan.emit.create-table-events.in-batch.enabled	Indica si los esquemas de tabla se envían por lotes durante la fase de inicialización del trabajo.	No	Boolean	false	Esta es una función experimental. Habilítela cuando un solo trabajo sincronice muchas tablas. Nota Este parámetro es compatible solo con VVR 11.4 y versiones posteriores.

Reutilizar un catálogo existente

A partir de VVR 11.5, puede hacer referencia directa a un catálogo MySQL integrado creado en la página Data Management dentro de un trabajo de ingesta de datos de Flink CDC. Esto reduce el esfuerzo manual para escribir las propiedades de conexión.

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog

Actualmente, los trabajos de ingesta de datos admiten la reutilización automática de los siguientes parámetros del catálogo MySQL:

hostname
port
username
password
catalog.table.metadata-columns
catalog.table.treat-tinyint1-as-boolean

Si necesita sobrescribir alguno de estos parámetros que se reutilizan automáticamente, escriba de forma explícita el parámetro YAML correspondiente. El parámetro escrito de manera explícita tiene mayor prioridad.

Mapeo de tipos

A continuación se muestra una tabla con el mapeo de tipos de datos para la ingesta de datos.

Tipo de campo de MySQL CDC	Tipo de campo de CDC
TINYINT(n)	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
YEAR	INT
INT
MEDIUMINT
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
BIGINT	BIGINT
INT UNSIGNED
INT 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] and p <= 38	DECIMAL(p, s)
DECIMAL(p, s) [UNSIGNED] [ZEROFILL] and p <= 38
FIXED(p, s) [UNSIGNED] [ZEROFILL] and p <= 38
BOOLEAN	BOOLEAN
BIT(1)
TINYINT(1)
DATE	DATE
TIME [(p)]	TIME [(p)]
DATETIME [(p)]	TIMESTAMP [(p)]
TIMESTAMP [(p)]	El mapeo depende del valor del parámetro `treat-timestamp-as-datetime-enabled`: `true`:TIMESTAMP[(p)] `false`:TIMESTAMP_LTZ[(p)]
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
BIT(n)	BINARY(⌈(n + 7) / 8⌉)
BINARY(n)	BINARY(n)
VARBINARY(N)	VARBINARY(N)
NUMERIC(p, s) [UNSIGNED] [ZEROFILL] and 38 < p <= 65	STRING Nota En MySQL, el tipo de dato decimal tiene una precisión máxima de 65, pero en Flink esta se limita a 38. Por lo tanto, si define una columna decimal con una precisión mayor a 38, debe mapearla a una cadena para evitar la pérdida de precisión.
DECIMAL(p, s) [UNSIGNED] [ZEROFILL] and 38 < p <= 65
FIXED(p, s) [UNSIGNED] [ZEROFILL] and 38 < p <= 65
TINYTEXT	STRING
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
JSON	STRING Nota El tipo de dato JSON se convierte en una cadena con formato JSON en Flink.
GEOMETRY	STRING Nota Los tipos de datos espaciales de MySQL se convierten en cadenas con un formato JSON fijo. Para obtener más información, consulte Spatial Data Type Mapping de MySQL.
POINT
LINESTRING
POLYGON
MULTIPOINT
MULTILINESTRING
MULTIPOLYGON
GEOMETRYCOLLECTION
TINYBLOB	BYTES Nota Para el tipo de dato BLOB en MySQL, solo se admiten blobs cuya longitud no supere 2,147,483,647 (2**31-1).
BLOB
MEDIUMBLOB
LONGBLOB

Establecer un ID de servidor para evitar conflictos en el consumo de registros binarios

Cuando un trabajo de ingesta de datos lee registros binarios, el origen se registra en MySQL como cliente de replicación mediante un server-id. Si varios trabajos u otros clientes de replicación usan el mismo ID de servidor, se generan conflictos en el consumo de registros binarios y los trabajos fallan. Tenga en cuenta lo siguiente al configurar los ID de servidor:

De forma predeterminada, un server-id es un valor único aleatorio entre 5400 y 6400. Si varios trabajos usan los valores predeterminados, pueden surgir conflictos. Se recomienda configurar de forma explícita ID que no se traslapen entre los trabajos.
Si el paralelismo del origen es mayor que 1, debe configurar un rango de ID de servidor. La cantidad de ID disponibles en el rango no debe ser menor que el paralelismo, ya que cada lector paralelo usa un ID distinto.

Por ejemplo, si el paralelismo del origen es 4, configure un rango que contenga cuatro ID.

source:
  type: mysql
  name: MySQL Source
  hostname: <hostname>
  port: 3306
  username: <username>
  password: <password>
  tables: app_db.\.*
  server-id: 5400-5403

sink:
  type: hologres

Si varios trabajos leen desde la misma instancia MySQL, asigne un rango sin traslapes a cada uno. Por ejemplo, el trabajo A utiliza 5400-5403 y el trabajo B emplea 5404-5407.

Acelerar la lectura de registros binarios

Cuando utiliza el conector MySQL como origen de datos para la ingesta, este analiza los archivos de registro binario para generar diversos mensajes de cambio durante la fase incremental. Dichos archivos registran todas las modificaciones de las tablas en formato binario. Puede acelerar el análisis de los archivos de registro binario de las siguientes maneras.

Habilite el análisis en paralelo y los filtros de análisis. Esta función requiere Realtime Compute for Apache Flink con Ververica Runtime (VVR) 8.0.7 o posterior, y no está disponible en la edición comunitaria del conector MySQL CDC.
- Habilite la opción scan.only.deserialize.captured.tables.changelog.enabled para analizar eventos de cambio únicamente en las tablas especificadas.
- Habilite la opción scan.parallel-deserialize-changelog.enabled para usar varios hilos al analizar el archivo de registro binario y entregar los eventos a la cola del consumidor en orden. Al habilitar esta opción, por lo general también necesita aumentar el TaskManager CPU.
Optimice los parámetros de Debezium.
```
debezium.max.queue.size: 162580
debezium.max.batch.size: 40960
debezium.poll.interval.ms: 50
```
- debezium.max.queue.size: número máximo de registros que admite la cola de bloqueo. Cuando Debezium lee un flujo de eventos de la base de datos, coloca los eventos en una cola de bloqueo antes de escribirlos en el destino. El valor predeterminado es 8192.
- debezium.max.batch.size: cantidad máxima de eventos que procesa el conector en cada iteración. El valor predeterminado es 2048.
- debezium.poll.interval.ms: tiempo en milisegundos que espera el conector antes de solicitar nuevos eventos de cambio. El valor predeterminado es 1000 milisegundos, es decir, 1 segundo.

Ejemplo de uso:

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

La capacidad de consumo de registros binarios de MySQL CDC Enterprise Edition es de 85 MB/s, aproximadamente el doble que la versión comunitaria de código abierto. Cuando la velocidad de generación de archivos de registro binario supera los 85 MB/s (es decir, un archivo de 512 MB cada 6 segundos), la latencia del trabajo de Flink aumenta de forma continua. Dicha latencia disminuye gradualmente una vez que baja la velocidad de generación de los archivos. Si un archivo de registro binario contiene una transacción grande, la latencia de procesamiento puede incrementarse temporalmente y se reduce después de leer el registro de esa transacción.

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:

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, currentEmitEventTimeLag indica la latencia al leer los datos de las tablas relevantes para el trabajo desde dicho registro.

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

La contrapresión puede reducir la velocidad a la que el origen envía datos a los operadores posteriores. Es posible que observe aumentos periódicos en sourceIdleTime, así como un crecimiento continuo de currentFetchEventTimeLag y currentEmitEventTimeLag. Para resolverlo, incremente el paralelismo del nodo donde se origina la contrapresión.
Revise la métrica TM CPU Usage en la página CPU y TM GC Time en la página JVM para determinar si faltan recursos de CPU o memoria. Puede aumentar los recursos del trabajo para mejorar el rendimiento de lectura.

Leer registros binarios archivados desde OSS

Cuando utiliza una instancia ApsaraDB RDS for MySQL como origen de datos, puede leer las copias de seguridad de registros almacenadas en OSS. Si el archivo correspondiente a la marca de tiempo o posición del registro binario especificado se encuentra en OSS, Flink extrae automáticamente el archivo de registro desde OSS hacia el clúster. Si el archivo reside de forma local en la base de datos, Flink cambia automáticamente a la lectura mediante una conexión de base de datos. Esta función solo está disponible en Realtime Compute for Apache Flink y no es compatible con la edición comunitaria del conector MySQL CDC.

Para habilitar la lectura desde las copias de seguridad de registros en OSS, debe configurar los parámetros de conexión de ApsaraDB RDS for MySQL. Por ejemplo:

source:
  type: mysql
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: <yourPassword>
  tables: <yourTables>
  # RDS connection parameters for reading archived binlogs from OSS
  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 # Database instance ID.
  rds.main-db-id: 12345678 # Primary database ID.
  rds.endpoint: rds.aliyuncs.com

Ingesta de datos para la sincronización de bases de datos y esquemas

Para los trabajos que solo implican lógica de sincronización de datos, se recomienda ejecutarlos como trabajos de ingesta de datos, ya que están profundamente optimizados para escenarios de integración de datos. Consulte Get Started with Flink CDC Data Ingestion y Develop a Flink CDC Data Ingestion Job (Public Preview) para ver las instrucciones de uso.

A continuación se muestra un ejemplo de trabajo de ingesta de datos que sincroniza toda la base de datos app_db de MySQL a Hologres, incluidos los cambios de esquema posteriores provenientes de la base de datos de origen:

source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: Sync MySQL Database to Hologres

Detección de tablas nuevas en la ingesta de datos

El conector de ingesta de datos de MySQL ofrece opciones de configuración para detectar tablas nuevas en dos escenarios distintos.

Parámetro	Descripción	Notas
`scan.newly-added-table.enabled`	Cuando un trabajo se reinicia desde un punto de control, esta opción sincroniza las tablas que no se detectaron durante el inicio anterior. Lee tanto los datos de instantánea como los incrementales de estas tablas nuevas.	Esta opción solo es compatible cuando `scan.startup.mode` se establece en `initial`. No tiene efecto en otros modos de inicio.
`scan.binlog.newly-added-table.enabled`	Durante la fase incremental, esta opción sincroniza automáticamente los datos de las tablas recién detectadas.	Se recomienda habilitar esta opción al iniciar el trabajo por primera vez. Así, el trabajo analiza automáticamente las sentencias DDL `CREATE TABLE` y sincroniza los datos hacia el destino. Si la habilita y reinicia el trabajo después de haber creado la tabla en la base de datos, podría generar datos incompletos. En el modo de inicio `initial`, las operaciones DDL no se sincronizan hacia el destino hasta que finaliza la fase de instantánea. Las tablas creadas durante dicha fase no se sincronizan automáticamente, incluso si `scan.binlog.newly-added-table.enabled` está habilitado.

Importante

Durante la fase de lectura completa, no se admite guardar un savepoint y luego agregar o eliminar tablas de origen antes de reiniciar desde ese savepoint. Hacerlo impide que el trabajo lea los datos correctamente.
No habilite scan.newly-added-table.enabled y scan.binlog.newly-added-table.enabled al mismo tiempo, ya que habilitar ambas opciones puede provocar duplicación de datos.