Introducción a CDC (captura de datos modificados)

La captura de datos modificados (CDC) es una técnica que se utiliza para realizar un seguimiento de los cambios en el nivel de fila en las operaciones de la base de datos (inserciones, actualizaciones, eliminaciones) y notificar a otros sistemas en el orden en que se producen los eventos. En escenarios de recuperación ante desastres, la CDC sincroniza principalmente los datos entre una base de datos principal y una de respaldo, lo que permite la sincronización de datos en tiempo real desde la base de datos principal a la secundaria.

 source ----------> CDC ----------> sink

Centro de control y prevención de enfermedades del túnel marino Apache

SeaTunnel CDC ofrece dos tipos de sincronización de datos:

• Lectura de instantánea : lee datos históricos de una tabla.
• Seguimiento incremental : lee los cambios de registro incrementales de una tabla.

Sincronización de instantáneas sin bloqueo

Se hace hincapié en la fase de sincronización de instantáneas sin bloqueos porque muchas plataformas CDC existentes, como Debezium, pueden bloquear tablas durante la sincronización de datos históricos. La lectura de instantáneas es el proceso de sincronización de los datos históricos de una base de datos. El flujo básico de este proceso es el siguiente:

 storage -------------> splitEnumerator ---------- split ----------> reader ^ | | | \----------------- report -----------/

Particionado dividido

splitEnumerator (distribuidor dividido) divide los datos de la tabla en múltiples divisiones según campos específicos (como ID de tabla o claves únicas) y tamaño de paso definido.

Procesamiento paralelo

Cada división se asigna a un lector diferente para la lectura en paralelo. Un solo lector ocupará una conexión.

Comentarios sobre el evento

Después de completar la operación de lectura de una división, cada lector informa el progreso a splitEnumerator . Los metadatos de la división se proporcionan de la siguiente manera:

 String splitId # Routing ID TableId tableId # Table ID SeatunnelRowType splitKeyType # The type of field used for partitioning Object splitStart # Start point of the partition Object splitEnd # End point of the partition

Una vez que el lector recibe la información de la división, genera las instrucciones SQL correspondientes. Antes de comenzar, registra la posición correspondiente de la división actual en el registro de la base de datos. Después de completar la división actual, el lector informa el progreso al splitEnumerator con los siguientes datos:

 String splitId # Split ID Offset highWatermark # Log position corresponding to the split, for future validation

Sincronización incremental

La fase de sincronización incremental comienza después de la fase de lectura de instantáneas. En esta etapa, todos los cambios que se producen en la base de datos de origen se capturan y se sincronizan con la base de datos de respaldo en tiempo real. Esta fase escucha el registro de la base de datos (por ejemplo, el binlog de MySQL). El seguimiento incremental suele ser de un solo subproceso para evitar extracciones duplicadas del binlog y reducir la carga de la base de datos. Por lo tanto, solo se utiliza un lector, que ocupa una única conexión.

 data log -------------> splitEnumerator ---------- split ----------> reader ^ | | | \----------------- report -----------/

En la fase de sincronización incremental, todas las divisiones y tablas de la fase de instantáneas se combinan en una única división. Los metadatos de la división durante esta fase son los siguientes:

 String splitId Offset startingOffset # The lowest log start position among all splits Offset endingOffset # Log end position, or "continuous" if ongoing, eg, in the incremental phase List<TableId> tableIds Map<TableId, Offset> tableWatermarks # Watermark for all splits List<CompletedSnapshotSplitInfo> completedSnapshotSplitInfos # Snapshot phase split details

Los campos CompletedSnapshotSplitInfo son los siguientes:

 String splitId TableId tableId SeatunnelRowType splitKeyType Object splitStart Object splitEnd Offset watermark # Corresponds to the highWatermark in the report

La división en la fase incremental contiene la marca de agua para todas las divisiones en la fase de instantánea. La marca de agua mínima se selecciona como punto de inicio para la sincronización incremental.

Semántica de "exactamente una vez"

Ya sea en la fase de lectura de instantáneas o de lectura incremental, la base de datos también puede cambiar para la sincronización. ¿Cómo garantizamos exactamente una entrega?

Fase de lectura de instantáneas

En la fase de lectura de instantáneas, por ejemplo, se sincroniza una división mientras se producen cambios, como la inserción de una fila k3 , una actualización de k2 y una eliminación de k1 . Si no se utiliza ninguna identificación de tarea durante el proceso de lectura, las actualizaciones podrían perderse. SeaTunnel maneja esto de la siguiente manera:

• Primero, verifique la posición del binlog (marca de agua baja) antes de leer la división.
• Leyendo los datos en el rango split{start, end} .
• Registrando la marca de agua alta después de la lectura.

Si high = low , los datos de la división no han cambiado durante la lectura. Si (high - low) > 0 , se han producido cambios durante el procesamiento. En tal caso, SeaTunnel hará lo siguiente:

• Almacene en caché los datos divididos en la memoria como una tabla en memoria.
• Aplicar los cambios desde low watermark a high watermark en orden, utilizando claves principales para reproducir operaciones en la tabla en memoria.
• Reportar el punto más alto del agua.

 insert k3 update k2 delete k1 | | | vvv bin log --|---------------------------------------------------|-- log offset low watermark high watermark CDC reads: k1 k3 k4 | Replays v Real data: k2 k3' k4

Fase incremental

Antes de comenzar la fase incremental, SeaTunnel primero valida todas las divisiones del paso anterior. Entre divisiones, los datos pueden actualizarse; por ejemplo, si se insertan nuevos registros entre la división 1 y la división 2, podrían perderse durante la fase de captura de pantalla. Para recuperar estos datos entre divisiones, SeaTunnel sigue este enfoque:

• De todos los informes divididos, busque la marca de agua más pequeña como marca de agua inicial para comenzar a leer el registro.
• Para cada entrada de registro leída, verifique completedSnapshotSplitInfos para ver si los datos se procesaron en alguna división. De lo contrario, se consideran datos entre divisiones y deben corregirse.
• Una vez validadas todas las divisiones, el proceso pasa a la fase incremental completa.

 |------------filter split2-----------------| |----filter split1------| data log -|-----------------------|------------------|----------------------------------|- log offset min watermark split1 watermark split2 watermark max watermark

Punto de control y reanudación

¿Qué sucede con la pausa y reanudación del CDC? SeaTunnel utiliza un algoritmo de captura de instantáneas distribuidas (Chandy-Lamport):

Supongamos que el sistema tiene dos procesos, p1 y p2 , donde p1 tiene tres variables X1 Y1 Z1 y p2 tiene tres variables X2 Y2 Z2 . Los estados iniciales son los siguientes:

 p1 p2 X1:0 X2:4 Y1:0 Y2:2 Z1:0 Z2:3

En este punto, p1 inicia una instantánea global. p1 primero registra su estado de proceso y luego envía un marcador a p2 .

Antes de que el marcador llegue p2 , p2 envía el mensaje M a p1 .

 p1 p2 X1:0 -------marker-------> X2:4 Y1:0 <---------M---------- Y2:2 Z1:0 Z2:3

Al recibir el marcador, p2 registra su estado y p1 recibe el mensaje M Como p1 ya realizó una instantánea local, solo necesita registrar el mensaje M La instantánea final se ve así:

 p1 M p2 X1:0 X2:4 Y1:0 Y2:2 Z1:0 Z2:3

En SeaTunnel CDC, los marcadores se envían a todos los lectores, enumeradores divididos, escritores y otros nodos, cada uno conserva su estado de memoria.