PolarDB-X CDC (Beta) - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Aprenda a usar el conector de PolarDB-X.

Contexto

PolarDB for Xscale (PolarDB-X) es la base de datos distribuida nativa de la nube y de alto rendimiento de Alibaba Cloud. Ofrece alto rendimiento, gran capacidad de almacenamiento, baja latencia, escalabilidad y disponibilidad extremadamente alta.

Importante

Este conector requiere VVR 11.5 o posterior y PolarDB-X 2.0 o posterior.

El conector PolarDB-X CDC solo admite tablas de origen. Para consultar una instancia de PolarDB-X como tabla de dimensiones o escribir en ella como tabla de destino, use el MySQL connector (vista previa pública).

Categoría	Descripción
Tipos compatibles	Tabla de origen, origen de ingesta de datos
Modo de ejecución	Solo modo streaming
Formato de datos	N/A
Métricas de monitoreo específicas	`currentFetchEventTimeLag`: Tiempo transcurrido entre la generación de los datos y su obtención por parte del operador de origen. Esta métrica solo aplica a la fase de binlog y es 0 durante la fase de instantánea. `currentEmitEventTimeLag`: Tiempo transcurrido entre la generación de los datos y su emisión por parte del operador de origen. Esta métrica solo aplica a la fase de binlog y es 0 durante la fase de instantánea. `sourceIdleTime`: Tiempo transcurrido desde que la tabla de origen produjo datos por última vez.
Tipo de API	SQL, trabajos YAML de ingesta de datos
Admite actualizar o eliminar datos en la tabla de destino	No

Funciones clave

El conector PolarDB-X CDC optimiza el análisis sintáctico de Binlog mediante filtrado en el servidor. Elimina los datos irrelevantes de Binlog directamente en el servidor PolarDB-X, lo que aumenta el rendimiento y conserva el ancho de banda de red.

Suscripción a Binlog bajo demanda

Gracias al filtrado de Binlog en el servidor, el conector envía al cliente únicamente los registros de cambios necesarios. Esto reduce el tráfico de red y mejora el rendimiento del consumo de datos.

Por ejemplo, para suscribirse solo a los cambios de las tablas db.table1 y db.table2 en su servidor PolarDB-X, configure su trabajo de Flink SQL de la siguiente manera:

CREATE TABLE polardbx_table_foo (
  ... -- Define the table schema here
) WITH (
  'connector' = 'polardbx-cdc',
  'database-name' = 'db',
  'table-name' = '.*',
  ..., -- Other parameters
  'polardbx.binlog.include.tables' = 'db.table1,db.table2' -- Captures changes from only the specified tables
);

A diferencia del conector MySQL CDC, que carga todo el Binlog de una instancia para filtrarlo en el cliente, el conector PolarDB-X CDC realiza el filtrado en el servidor. Esta capacidad permite suscripciones bajo demanda desde el cliente, lo que reduce significativamente la sobrecarga de E/S de red.

Límites de uso

El filtrado de Binlog en el servidor y la suscripción a nivel de tabla requieren la versión 2.5.0 o posterior del servidor PolarDB-X y la versión 5.4.20 o posterior del componente Log Service.

SQL

Sintaxis

CREATE TABLE polardbx_customer_table(
  `id` STRING,
  [columnName dataType,]*
  PRIMARY KEY(`id`) NOT ENFORCED
) WITH (
  'connector' = 'polardbx-cdc',
  'hosts' = 'pxc-**************-pub.polarx.rds.aliyuncs.com',
  'username' = 'pdx_user',
  'password' = 'pdx_password',
  'database' = 'full_db',
  'collection' = 'customers'
)

Parámetros WITH

Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
connector	Nombre del conector.	STRING	Sí	Ninguna	El valor debe ser `polardbx-cdc`.
hostname	Dirección IP o nombre de host de la base de datos PolarDB-X.	STRING	Sí	Ninguna	Use la dirección del clúster que aparece en la información de conexión de la instancia.
port	Número de puerto de la base de datos PolarDB-X.	INTEGER	No	3306	Ninguna
username	Nombre de usuario para la base de datos PolarDB-X.	STRING	Sí	Ninguna	Ninguna
password	Contraseña para la base de datos PolarDB-X.	STRING	Sí	Ninguna	Ninguna
database-name	Nombre de la base de datos PolarDB-X.	STRING	Sí	Ninguna	Puede usar una expresión regular para leer datos de varias bases de datos. Nota Al usar una expresión regular, no use los caracteres ^ y $ para coincidir con el inicio y el final de la cadena.
table-name	Nombre de la tabla PolarDB-X.	STRING	Sí	Ninguna	Puede usar una expresión regular para leer datos de varias tablas. Nota Al usar una expresión regular, no use los caracteres ^ y $ para coincidir con el inicio y el final de la cadena.
server-time-zone	Zona horaria de la sesión para la conexión a la base de datos.	STRING	No	Zona horaria del entorno de ejecución del trabajo.	Especifique un identificador de zona horaria IANA, como `Asia/Shanghai`. Este parámetro controla cómo se convierten los tipos `TIMESTAMP` de la tabla de origen a `STRING`.
scan.incremental.snapshot.chunk.size	Número de filas en cada fragmento para las lecturas de instantáneas incrementales.	INTEGER	No	8096	Durante la fase de instantánea, el conector divide una tabla en fragmentos y los almacena en caché en memoria. Un tamaño de fragmento menor aumenta el número total de fragmentos. Si bien esto mejora la granularidad de la tolerancia a fallos, también incrementa el riesgo de errores de falta de memoria (OOM) y reduce el rendimiento. Establezca un valor adecuado para equilibrar el rendimiento, la tolerancia a fallos y el consumo de memoria.
scan.snapshot.fetch.size	Número máximo de filas que se obtienen a la vez al leer una instantánea de la tabla.	INTEGER	No	1024	Ninguna
connect.timeout	Tiempo de espera para establecer una conexión con la base de datos PolarDB-X.	DURATION	No	30s	Ninguna
connection.pool.size	Tamaño del pool de conexiones de la base de datos.	INTEGER	No	20	Un pool de conexiones de base de datos reutiliza las conexiones para reducir la sobrecarga asociada.
connect.max-retries	Número máximo de reintentos tras un intento fallido de conexión a la base de datos PolarDB-X.	INTEGER	No	3	Ninguna
scan.startup.mode	Modo de inicio para el consumo de datos.	STRING	No	initial	Valores válidos: initial (predeterminado): En su primer inicio, el conector toma una instantánea completa de las tablas y luego lee los cambios incrementales desde el Binlog. latest-offset: No realiza ninguna instantánea. En el primer inicio, el conector comienza a leer desde el final del Binlog y captura solo los cambios realizados después de que se inicia el trabajo. earliest-offset: Nunca realiza una instantánea. Comienza a leer desde el desplazamiento de Binlog más antiguo disponible. specific-offset: Nunca realiza una instantánea. Comienza a leer desde un desplazamiento específico. Especifique el desplazamiento mediante scan.startup.specific-offset.file y scan.startup.specific-offset.pos, o bien mediante scan.startup.specific-offset.gtid-set. timestamp: Nunca realiza una instantánea. Comienza a leer el Binlog desde una marca de tiempo específica. La marca de tiempo se especifica en milisegundos mediante scan.startup.timestamp-millis. Importante Para los modos de inicio earliest-offset, specific-offset y timestamp, el esquema de la tabla al momento del inicio debe coincidir con el esquema en el desplazamiento especificado. Una discrepancia de esquema provocará que el trabajo falle. Asegúrese de que el esquema de la tabla no cambie entre el desplazamiento de Binlog especificado y el inicio del trabajo.
scan.startup.specific-offset.file	Nombre del archivo Binlog desde el cual comenzar cuando `scan.startup.mode` es `specific-offset`.	STRING	No	Ninguna	Este parámetro solo se usa cuando scan.startup.mode está establecido en specific-offset. Ejemplo: `mysql-bin.000003`.
scan.startup.specific-offset.pos	Desplazamiento dentro del archivo Binlog especificado cuando `scan.startup.mode` está establecido en `specific-offset`.	INTEGER	No	Ninguna	Este parámetro solo se usa cuando scan.startup.mode está establecido en specific-offset.
scan.startup.specific-offset.gtid-set	Conjunto GTID desde el cual comenzar cuando `scan.startup.mode` está establecido en `specific-offset`.	STRING	No	Ninguna	Este parámetro solo se usa cuando scan.startup.mode está establecido en specific-offset. Ejemplo: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	Marca de tiempo de inicio en milisegundos cuando `scan.startup.mode` está establecido en `timestamp`.	LONG	No	Ninguna	Este parámetro solo se usa cuando scan.startup.mode está establecido en timestamp. La unidad es milisegundos.
scan.startup.specific-offset.skip-events	Número de eventos de Binlog que se omiten al leer desde un desplazamiento específico.	INTEGER	No	Ninguna	Este parámetro solo se usa cuando scan.startup.mode está establecido en specific-offset.
scan.startup.specific-offset.skip-rows	Número de cambios de fila que se omiten al leer desde un desplazamiento específico. Un solo evento de Binlog puede contener varios cambios de fila.	INTEGER	No	Ninguna	Este parámetro solo se usa cuando scan.startup.mode está establecido en specific-offset.
heartbeat.interval	Intervalo en el que el origen envía un evento de latido para avanzar el desplazamiento de Binlog.	DURATION	No	Ninguna	Un evento de latido avanza el desplazamiento de Binlog del origen. Esto evita que el Binlog se purgue en un origen inactivo. Si el Binlog se purga, el trabajo falla y requiere un reinicio sin estado para recuperarse.
chunk-meta.group.size	Tamaño de los metadatos del fragmento.	INTEGER	No	1000	Si el tamaño de los metadatos supera este valor, se divide en varias partes para su transmisión.
chunk-key.even-distribution.factor.upper-bound	Límite superior del factor de distribución de fragmentos para la partición uniforme.	DOUBLE	No	1000.0	Si el factor de distribución es mayor que este valor, se usa una fragmentación no uniforme. Factor de distribución de fragmentos = (MAX(chunk-key) - MIN(chunk-key) + 1) / número total de filas.
chunk-key.even-distribution.factor.lower-bound	Límite inferior del factor de distribución de fragmentos para la partición uniforme.	DOUBLE	No	0.05	Si el factor de distribución es menor que este valor, se usa una fragmentación no uniforme. Factor de distribución de fragmentos = (MAX(chunk-key) - MIN(chunk-key) + 1) / número total de filas.
scan.newly-added-table.enabled	Indica si se deben buscar tablas recién agregadas que coincidan con los criterios de captura al reiniciar desde un punto de control.	BOOLEAN	No	false	Cuando se habilita, un trabajo reiniciado desde un Checkpoint o Savepoint busca y sincroniza las tablas recién agregadas que coinciden con el patrón 'table-name'. También deja de rastrear las tablas que ya no coinciden.
scan.incremental.snapshot.chunk.key-column	Especifica la columna usada para la fragmentación de datos durante la fase de instantánea.	STRING	Consulte Observaciones	Ninguna	Obligatorio para tablas sin clave primaria. La columna especificada no debe admitir valores nulos. Opcional para tablas con clave primaria. Solo se admite una columna de la clave primaria.
scan.incremental.close-idle-reader.enabled	Indica si se deben cerrar los lectores inactivos una vez completada la fase de instantánea.	BOOLEAN	No	false	Para que esta configuración surta efecto, `execution.checkpointing.checkpoints-after-tasks-finish.enabled` también debe establecerse en `true`.
scan.incremental.snapshot.backfill.skip	Indica si se debe omitir el proceso de relleno durante la fase de lectura de instantáneas.	BOOLEAN	No	false	Valores válidos: true: Omite el proceso de relleno durante la fase de lectura de instantáneas. false (predeterminado): No omite el proceso de relleno durante la fase de lectura de instantáneas. El relleno solo aplica durante la consulta de instantánea de un solo fragmento y no cubre toda la fase de lectura completa. Cuando se omite el relleno, 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 haber sido leído no se fusionan durante la fase de lectura completa y se leen desde el Binlog tras entrar en la fase incremental. Por ejemplo, una actualización en chunk5 que ocurre mientras se toma la instantánea de chunk5 se refleja directamente en dicha instantánea; si chunk5 se actualiza después de que el lector haya avanzado a chunk80, la actualización se aplica posteriormente desde el Binlog durante la fase incremental. Importante Cuando se habilita, los cambios que ocurren durante o después del escaneo de un fragmento aún se entregan desde el Binlog en la fase incremental y pueden duplicarse. Solo se garantiza la semántica de al menos una vez. Habilite esta opción únicamente cuando el destino descendente admita escrituras idempotentes por clave primaria.
scan.parse.online.schema.changes.enabled	Indica si se debe intentar analizar sintácticamente eventos DDL para cambios sin bloqueo de RDS durante la fase incremental.	BOOLEAN	No	false	Valores válidos: true: Analiza sintácticamente eventos DDL para cambios sin bloqueo de RDS. false (predeterminado): No analiza sintácticamente eventos DDL para cambios sin bloqueo de RDS. Esta es una función experimental. Antes de realizar cambios de esquema sin bloqueo en producción, cree un Savepoint para el trabajo de Flink a fin de garantizar que pueda recuperarse.
scan.only.deserialize.captured.tables.changelog.enabled	Indica si se deben deserializar eventos de cambio solo para las tablas capturadas especificadas durante la fase incremental.	BOOLEAN	No	true	Valores válidos: true (predeterminado): Deserializa datos de cambio solo para las tablas objetivo, lo que puede acelerar la lectura de Binlog. false: Deserializa datos de cambio para todas las tablas.
scan.read-changelog-as-append-only.enabled	Indica si se debe convertir el flujo de registro de cambios en un flujo de solo anexado.	BOOLEAN	No	false	Valores válidos: true: Convierte todos los tipos de mensajes (incluidos `INSERT`, `DELETE`, `UPDATE_BEFORE` y `UPDATE_AFTER`) en mensajes `INSERT`. Use esta opción solo en casos específicos, por ejemplo, para conservar un registro de las filas eliminadas del origen. false (predeterminado): Todos los tipos de mensajes se envían al destino tal como están.
scan.parallel-deserialize-changelog.enabled	Indica si se deben usar varios hilos para deserializar eventos de cambio durante la fase incremental.	BOOLEAN	No	false	Valores válidos: true: Usa varios hilos para la deserialización a fin de acelerar la lectura, conservando al mismo tiempo el orden de los eventos de Binlog. false (predeterminado): Usa un solo hilo para la deserialización.
scan.parallel-deserialize-changelog.handler.size	Número de controladores de eventos que se usan al deserializar eventos de cambio en paralelo.	INTEGER	No	2	Ninguna
scan.incremental.snapshot.unbounded-chunk-first.enabled	Indica si se deben distribuir primero los fragmentos no acotados durante la fase de lectura de instantáneas.	BOOLEAN	No	false	Valores válidos: true: Distribuye primero los fragmentos no acotados durante la fase de lectura de instantáneas. false (predeterminado): No distribuye primero los fragmentos no acotados durante la fase de lectura de instantáneas. Esta es una función experimental. Habilitarla puede reducir el riesgo de un error de falta de memoria (OOM) en el TaskManager al sincronizar el último fragmento de una instantánea. Esta configuración debe establecerse antes del primer inicio del trabajo.
polardbx.binlog.ignore.archive-events.enabled	Indica si se deben ignorar los eventos de archivo (principalmente eventos `DELETE`) en el Binlog de PolarDB-X.	BOOLEAN	No	false
polardbx.binlog.ignore.query-events.enabled	Indica si se deben ignorar los eventos `ROWS_QUERY_LOG_EVENT` en el Binlog de PolarDB-X.	BOOLEAN	No	false
polardbx.binlog.include.tables	Lee eventos de Binlog solo de las tablas especificadas. Use comas (`,`) para separar varios nombres de tabla.	STRING	No	Ninguna
polardbx.binlog.exclude.tables	Ignora eventos de Binlog de las tablas especificadas. Use comas (`,`) para separar varios nombres de tabla.	STRING	No	Ninguna

Mapeo de tipos

Tipo de PolarDB-X	Tipo 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]	DECIMAL(p, s)
BOOLEAN	BOOLEAN
TINYINT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
TIMESTAMP [(p)]	TIMESTAMP_LTZ [(p)]
TIMESTAMP [(p)]	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
CHAR(n)	STRING
VARCHAR(n)
TEXT
BINARY	BYTES
VARBINARY
BLOB

Ingesta de datos

Realtime Compute for Apache Flink 11.6 ahora admite el conector PolarDB-X como origen de datos para trabajos YAML de ingesta de datos.

Sintaxis

source:
   type: polardbx
   name: PolarDB-X Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: pdb.order_table
   # Ignore archive events in the binlog
   polardbx.binlog.ignore.archive-events.enabled: true
   # Ignore query events in the binlog
   polardbx.binlog.ignore.query-events.enabled: true
   # Subscribe only to the binlog for pdb.order_table to save bandwidth.
   polardbx.binlog.include.tables: pdb.order_table

sink:
  type: values

Parámetro

Parámetro	Descripción	Obligatorio	Tipo	Predeterminado	Observaciones
type	Tipo del origen de datos.	Sí	STRING	N/A	El valor debe ser `polardbx`.
name	Nombre del origen de datos.	No	STRING	N/A	N/A
hostname	Dirección IP o nombre de host de la instancia PolarDB-X.	Sí	STRING	N/A	Se recomienda la dirección VPC. Nota Si su instancia de PolarDB-X y el espacio de trabajo de Realtime Compute for Apache Flink no están en la misma VPC, debe establecer conectividad entre VPC o acceder a la instancia a través de la red pública. Para obtener más detalles, consulte Workspace Management and Operations y How can a fully-managed Flink cluster access the public network?.
username	Nombre de usuario para conectarse a la base de datos PolarDB-X.	Sí	STRING	N/A	N/A
password	Contraseña para el nombre de usuario especificado.	Sí	STRING	N/A	N/A
tables	Tablas de PolarDB-X que se van a sincronizar.	Sí	STRING	N/A	Admite expresiones regulares para leer datos de varias tablas. Puede separar varias expresiones regulares con una coma (,). Nota No use los anclajes de inicio y fin `^` y `$` en las expresiones regulares. Un punto (.) separa el nombre de la base de datos y el nombre de la tabla. Para coincidir con un punto literal, debe escaparlo con una barra invertida. Por ejemplo: `db0.\.`, `db1.user_table_[0-9]+` o `db[1-2].[app\|web]order_\.`.
tables.exclude	Tablas que se excluirán de la sincronización.	No	STRING	N/A	Admite expresiones regulares para excluir varias tablas. Puede separar varias expresiones regulares con una coma (,). Nota Un punto (.) separa el nombre de la base de datos y el nombre de la tabla. Para coincidir con un punto literal, debe escaparlo con una barra invertida. Por ejemplo: `db0.\.`, `db1.user_table_[0-9]+` o `db[1-2].[app\|web]order_\.`.
port	Puerto de la instancia PolarDB-X.	No	INTEGER	3306	N/A
schema-change.enabled	Indica si se deben emitir eventos de cambio de esquema.	No	BOOLEAN	true	N/A
jdbc.properties.*	Parámetros de conexión personalizados para la URL JDBC.	No	STRING	N/A	Puede pasar parámetros de conexión personalizados. Por ejemplo, para deshabilitar SSL, establezca 'jdbc.properties.useSSL' = 'false'.
debezium.*	Parámetros personalizados de Debezium para leer el binlog.	No	STRING	N/A	Puede pasar parámetros personalizados de Debezium. Por ejemplo, establezca 'debezium.event.deserialization.failure.handling.mode' = 'ignore' para definir cómo maneja el conector los errores de deserialización.
scan.incremental.snapshot.chunk.size	Tamaño de cada fragmento, en número de filas.	No	INTEGER	8096	Una tabla de PolarDB-X se divide en varios fragmentos para su lectura. Los datos de cada fragmento se almacenan en caché en memoria hasta que se leen por completo. Un tamaño de fragmento menor aumenta el número total de fragmentos, lo que proporciona una recuperación ante fallos más granular, pero puede incrementar el riesgo de OOM y reducir el rendimiento general. Debe equilibrar estas contrapartidas y configurar un tamaño de fragmento razonable.
scan.snapshot.fetch.size	Número máximo de registros obtenidos a la vez durante una instantánea completa de la tabla.	No	INTEGER	1024	N/A
scan.startup.mode	Modo de inicio para consumir datos.	No	STRING	initial	Valores válidos: initial (predeterminado): Toma una instantánea inicial de la tabla y luego lee los cambios más recientes desde el binlog. latest-offset: Omite la fase de instantánea y lee el binlog desde la posición más reciente, capturando solo los cambios realizados después de que se inicia el conector. earliest-offset: Omite la fase de instantánea y comienza a leer desde la posición de binlog más antigua disponible. specific-offset: Omite la fase de instantánea y comienza a leer desde una posición específica. Especifique la posición mediante `scan.startup.specific-offset.file` y `scan.startup.specific-offset.pos`, o bien mediante `scan.startup.specific-offset.gtid-set`. timestamp: Omite la fase de instantánea y comienza a leer el binlog desde una marca de tiempo específica, la cual se indica en milisegundos mediante el parámetro scan.startup.timestamp-millis. Importante Para los modos de inicio earliest-offset, specific-offset y timestamp, el trabajo falla con un error si el esquema de la tabla al momento del inicio difiere del esquema de la tabla en la posición de inicio especificada. En otras palabras, cuando use estos tres modos de inicio, debe asegurarse de que el esquema de la tabla correspondiente no cambie entre la posición de consumo de Binlog especificada y el momento de inicio del trabajo.
scan.startup.specific-offset.file	Nombre del archivo binlog para la posición de inicio.	No	STRING	N/A	Obligatorio cuando scan.startup.mode está establecido en specific-offset. Formato de ejemplo: `mysql-bin.000003`.
scan.startup.specific-offset.pos	Desplazamiento dentro del archivo binlog especificado desde el cual comenzar.	No	INTEGER	N/A	Obligatorio cuando scan.startup.mode está establecido en specific-offset.
scan.startup.specific-offset.gtid-set	Conjunto GTID para la posición de inicio.	No	STRING	N/A	Se usa cuando scan.startup.mode está establecido en specific-offset. Formato de ejemplo de conjunto GTID: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	Marca de tiempo en milisegundos para la posición de inicio.	No	LONG	N/A	Obligatorio cuando scan.startup.mode está establecido en timestamp. Importante Cuando especifique una marca de tiempo, el conector localiza el archivo de inicio correcto leyendo el primer evento de cada archivo binlog para comprobar su marca de tiempo. Asegúrese de que el archivo binlog de destino para la marca de tiempo especificada no se haya purgado de la base de datos y siga siendo legible.
server-time-zone	Zona horaria de sesión usada por la base de datos.	No	STRING	Zona horaria del entorno de ejecución del trabajo de Flink (la zona de disponibilidad de su espacio de trabajo).	Ejemplo: Asia/Shanghai. Este parámetro controla cómo se convierten los tipos `TIMESTAMP` en PolarDB-X a `STRING`. Para obtener más información, consulte la documentación de Debezium sobre tipos de tiempo.
scan.startup.specific-offset.skip-events	Número de eventos de binlog que se omiten al leer desde una posición específica.	No	INTEGER	N/A	Se usa cuando scan.startup.mode está establecido en specific-offset.
scan.startup.specific-offset.skip-rows	Número de cambios de fila que se omiten al leer desde una posición específica. Un solo evento de binlog puede contener varios cambios de fila.	No	INTEGER	N/A	Se usa cuando scan.startup.mode está establecido en specific-offset.
connect.timeout	Tiempo máximo de espera para una conexión a la base de datos antes de reintentar.	No	DURATION	30s	N/A
connect.max-retries	Número máximo de reintentos para una conexión fallida a la base de datos.	No	INTEGER	3	N/A
connection.pool.size	Tamaño del pool de conexiones de la base de datos.	No	INTEGER	20	El pool de conexiones reutiliza las conexiones para reducir la sobrecarga asociada.
heartbeat.interval	Intervalo en el que el origen envía un evento de latido para avanzar la posición del binlog.	No	DURATION	30s	Los eventos de latido avanzan la posición del binlog para las tablas que se actualizan con poca frecuencia. Sin ellos, la posición del binlog podría no avanzar, lo que puede provocar su caducidad. Un binlog caducado provoca que el trabajo falle y requiera un reinicio sin estado para recuperarse.
scan.incremental.snapshot.chunk.key-column	Especifica una columna para usar como clave de división de fragmentos durante la fase de instantánea.	No	STRING	N/A	Solo puede seleccionar una columna de la clave primaria.
chunk-meta.group.size	Tamaño de los metadatos del fragmento.	No	INTEGER	1000	Si el tamaño de los metadatos supera este valor, se divide en varias partes para su transmisión.
chunk-key.even-distribution.factor.lower-bound	Límite inferior del factor de distribución de fragmentos para determinar si se usa fragmentación uniforme.	No	DOUBLE	0.05	Si el factor de distribución es menor que este valor, se usa 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	Límite superior del factor de distribución de fragmentos para determinar si se usa fragmentación uniforme.	No	DOUBLE	1000.0	Si el factor de distribución es mayor que este valor, se usa 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 deben cerrar los lectores inactivos una vez completada la fase de 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	Indica si se deben deserializar eventos de cambio solo para las tablas especificadas durante la fase incremental.	No	BOOLEAN	true	Valores válidos: true: Deserializa datos de cambio solo para las tablas objetivo, lo que acelera la lectura de binlog. false: Deserializa datos de cambio para todas las tablas.
scan.parallel-deserialize-changelog.enabled	Indica si se deben usar varios hilos para analizar sintácticamente eventos de cambio durante la fase incremental.	No	BOOLEAN	false	Valores válidos: true: Usa varios hilos para la deserialización de eventos de cambio conservando el orden de los eventos de binlog, lo que acelera la lectura. false (predeterminado): Usa un solo hilo para la deserialización de eventos.
scan.parallel-deserialize-changelog.handler.size	Número de controladores de eventos que se usan al analizar sintácticamente eventos de cambio en paralelo.	No	INTEGER	2	N/A
metadata-column.include-list	Columnas de metadatos que se pasan al destino descendente.	No	STRING	N/A	Los metadatos disponibles incluyen `op_ts`, `es_ts`, `query_log`, `file` y `pos`. Puede usar comas para separar varias columnas de metadatos. Nota El conector YAML de PolarDB-X CDC no requiere ni admite agregar columnas de metadatos para el nombre de la base de datos, el nombre de la tabla o `op_type`. Puede usar `__data_event_type__` en una expresión de transformación para obtener el tipo de datos de cambio, o usar `__schema_name__` y `__table_name__` para obtener los nombres de la base de datos y la tabla. Importante La columna `file` contiene el nombre del archivo binlog (fase incremental) o una cadena vacía (fase de instantánea). La columna `pos` contiene el desplazamiento del binlog (fase incremental) o `0` (fase de instantánea).
scan.newly-added-table.enabled	Al reiniciar desde un punto de control o un punto de guardado, indica si se deben sincronizar las tablas recién agregadas que coinciden con el patrón de tablas y dejar de rastrear las tablas que ya no coinciden.	No	BOOLEAN	false	Esta configuración solo surte efecto al reiniciar desde un punto de control o un punto de guardado.
scan.binlog.newly-added-table.enabled	Durante la fase incremental, indica si se deben enviar datos para las tablas recién coincidentes.	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 clave de división para la fragmentación durante la fase de instantánea para tablas específicas.	No	STRING	N/A	Use dos puntos (`:`) para conectar un nombre de tabla y un nombre de columna a fin de definir una regla. Los nombres de tabla pueden ser expresiones regulares. Puede definir varias reglas separadas por punto y coma (`;`). Por ejemplo: `db1.user_table_[0-9]+:col1;db[1-2].[app\|web]_order_\\.*:col2`. Este parámetro es obligatorio para tablas sin clave primaria, y la columna seleccionada debe ser de un tipo que no admita valores nulos (`NOT NULL`). Para tablas con clave primaria, este parámetro es opcional y solo puede seleccionar una columna de la clave primaria.
scan.parse.online.schema.changes.enabled	Durante la fase incremental, indica si se deben analizar sintácticamente eventos DDL para cambios de esquema sin bloqueo de ApsaraDB RDS.	No	BOOLEAN	false	Valores válidos: true: Analiza sintácticamente eventos DDL para cambios de esquema sin bloqueo de ApsaraDB RDS. false (predeterminado): No analiza sintácticamente eventos DDL para cambios de esquema sin bloqueo de ApsaraDB RDS. Esta es una función experimental. Antes de realizar un cambio sin bloqueo en línea, le recomendamos crear un punto de guardado para el trabajo de Flink a fin de garantizar que pueda restaurarse si es necesario.
scan.incremental.snapshot.backfill.skip	Indica si se debe omitir el relleno durante la fase de lectura de instantáneas.	No	BOOLEAN	false	Valores válidos: true: Omite el relleno durante la fase de lectura de instantáneas. false (predeterminado): No omite el relleno durante la fase de lectura de instantáneas. El relleno solo aplica durante la consulta de instantánea de un solo fragmento y no cubre toda la fase de lectura completa. Cuando se omite el relleno, 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 haber sido leído no se fusionan durante la fase de lectura completa y se leen desde el Binlog tras entrar en la fase incremental. Por ejemplo, una actualización en chunk5 que ocurre mientras se toma la instantánea de chunk5 se refleja directamente en dicha instantánea; si chunk5 se actualiza después de que el lector haya avanzado a chunk80, la actualización se aplica posteriormente desde el Binlog durante la fase incremental. Importante Cuando se habilita, los cambios que ocurren durante o después del escaneo de un fragmento aún se entregan desde el Binlog en la fase incremental y pueden duplicarse. Solo se garantiza la semántica de al menos una vez. Habilite esta opción únicamente cuando el destino descendente admita escrituras idempotentes por clave primaria.
treat-tinyint1-as-boolean.enabled	Indica si se debe tratar el tipo de datos `TINYINT(1)` como `BOOLEAN`.	No	BOOLEAN	true	Valores válidos: true (predeterminado): Trata el tipo `TINYINT(1)` como `BOOLEAN`. false: No trata el tipo `TINYINT(1)` como `BOOLEAN`.
treat-timestamp-as-datetime-enabled	Indica si se debe tratar el tipo de datos `TIMESTAMP` como `DATETIME`.	No	BOOLEAN	false	Valores válidos: true: Trata el tipo `TIMESTAMP` de PolarDB-X como `DATETIME` y lo mapea al tipo `TIMESTAMP` de CDC. false (predeterminado): Mapea el tipo `TIMESTAMP` de PolarDB-X al tipo `TIMESTAMP_LTZ` de CDC. El tipo `TIMESTAMP` de PolarDB-X almacena la hora UTC y se ve afectado por la zona horaria. El tipo `DATETIME` de PolarDB-X almacena la hora literal y no se ve afectado por la zona horaria. Cuando se habilita, este parámetro convierte los datos `TIMESTAMP` de PolarDB-X al tipo `DATETIME` basándose en `server-time-zone`.
include-comments.enabled	Indica si se deben sincronizar los comentarios de tablas y columnas.	No	BOOLEAN	false	Valores válidos: true: Sincroniza los comentarios de tablas y columnas. false (predeterminado): No sincroniza los comentarios de tablas y columnas. Habilitar este parámetro aumenta el uso de memoria del trabajo.
scan.incremental.snapshot.unbounded-chunk-first.enabled	Indica si se deben distribuir primero los fragmentos no acotados durante la fase de lectura de instantáneas.	No	BOOLEAN	false	Valores válidos: true: Distribuye primero los fragmentos no acotados durante la fase de lectura de instantáneas. false (predeterminado): No distribuye primero los fragmentos no acotados durante la fase de lectura de instantáneas. Esta es una función experimental. Habilitarla puede reducir el riesgo de un error OOM en el TaskManager cuando se sincroniza el último fragmento. Le recomendamos agregar este parámetro antes del primer inicio del trabajo.
binlog.session.network.timeout	Tiempo de espera de red para la conexión de binlog.	No	DURATION	10m	Si se establece en `0s`, se usa el tiempo de espera predeterminado del servidor.
scan.rate-limit.records-per-second	Número máximo de registros que el origen puede emitir por segundo.	No	LONG	N/A	Use este parámetro para limitar el rendimiento de lectura de datos. Este límite aplica tanto a la fase de instantánea como a la incremental. La métrica `numRecordsOutPerSecond` del origen muestra el número de registros emitidos por segundo para todo el flujo de datos. Puede usar esta métrica para ajustar este parámetro. Durante la fase de lectura de instantáneas, también puede ser necesario reducir el número de filas leídas por lote disminuyendo el valor del parámetro `scan.incremental.snapshot.chunk.size`.
include-binlog-meta.enable	Indica si se deben incluir metadatos de binlog sin procesar de PolarDB-X, como GTID y posición de binlog, en los registros emitidos.	No	BOOLEAN	false	Aplicable a escenarios de sincronización de binlog sin procesar, como reemplazar una canalización de sincronización existente basada en Canal.
polardbx.binlog.ignore.archive-events.enabled	Indica si se deben ignorar los eventos de archivo (principalmente eventos `DELETE`) en el binlog de PolarDB-X.	No	BOOLEAN	false
polardbx.binlog.ignore.query-events.enabled	Indica si se deben ignorar los eventos `ROWS_QUERY_LOG_EVENT` en el binlog de PolarDB-X.	No	BOOLEAN	false
polardbx.binlog.include.tables	Lista separada por comas de tablas a las que suscribirse para eventos de binlog.	No	STRING	N/A	Nota Este parámetro solo afecta a la fase incremental (binlog) y no aplica a la fase de instantánea.
polardbx.binlog.exclude.tables	Lista separada por comas de tablas cuyos eventos de binlog deben ignorarse.	No	STRING	N/A	Nota Este parámetro solo afecta a la fase incremental (binlog) y no aplica a la fase de instantánea.