Realtime Compute for Apache Flink permite crear trabajos de ingestión de datos con Flink CDC mediante un archivo YAML para sincronizar datos desde un origen hacia un destino. Este tema describe los pasos para desarrollar un trabajo de ingestión de datos con Flink CDC.
Contexto
Mediante configuraciones YAML, es posible definir fácilmente pipelines ETL complejos que se convierten automáticamente en lógica de ejecución de Flink. La ingestión de datos con Flink CDC ofrece una solución potente de integración de datos basada en Flink CDC. Esta solución admite de manera eficiente la sincronización completa de bases de datos, la sincronización de tablas individuales, la sincronización de bases de datos y tablas fragmentadas, la detección automática de nuevas tablas, el manejo de cambios de esquema y las columnas calculadas personalizadas. También permite el procesamiento ETL, el filtrado mediante cláusulas WHERE y la eliminación de columnas. Este enfoque declarativo simplifica significativamente el proceso de integración de datos, además de mejorar la eficiencia y la confiabilidad.
Ventajas de Flink CDC
En Realtime Compute for Apache Flink, puede desarrollar un trabajo de ingestión de datos con Flink CDC, un trabajo SQL o un trabajo DataStream para realizar la sincronización de datos. Las siguientes secciones describen las ventajas de utilizar un trabajo de ingestión de datos con Flink CDC frente a las otras dos opciones.
Flink CDC vs. Flink SQL
Los trabajos de ingestión de datos con Flink CDC y los trabajos SQL utilizan distintos tipos de datos para la transmisión:
Los trabajos SQL transmiten los datos como
RowData. Cada objetoRowDatatiene su propio tipo de cambio, el cual incluye cuatro tipos principales: inserción (+I), actualización previa (-U), actualización posterior (+U) y eliminación (-D).Flink CDC utiliza
SchemaChangeEventpara comunicar información sobre cambios de esquema, como la creación de una tabla, la adición de una columna o el truncamiento de una tabla. UtilizaDataChangeEventpara comunicar cambios en los datos, como inserciones, actualizaciones y eliminaciones. Un mensaje de actualización contiene tanto el contenido previo como el posterior, lo que permite escribir los datos de cambio originales en el destino.
La siguiente tabla enumera las ventajas de los trabajos de ingestión de datos con Flink CDC frente a los trabajos SQL.
|
Ingestión con Flink CDC |
Flink SQL |
|
Detección automática de esquemas y sincronización completa de bases de datos |
Requiere instrucciones |
|
Admite múltiples directivas para cambios de esquema |
No admite cambios de esquema |
|
Conserva el registro de cambios original |
Altera la estructura original del registro de cambios |
|
Permite leer y escribir en varias tablas |
Lee y escribe en una sola tabla |
En comparación con las instrucciones CTAS/CDAS, los trabajos de Flink CDC ofrecen funciones más potentes, entre ellas:
Sincronización inmediata de cambios de esquema en el origen, sin esperar a que nuevas escrituras de datos activen la sincronización.
Conservación del registro de cambios original, lo que garantiza que los mensajes de actualización no se dividan.
Sincronización de más tipos de cambios de esquema, como
TRUNCATE TABLEyDROP TABLE.Mapeo flexible de tablas y definición de nombres de tablas de destino.
Comportamientos de evolución de esquema flexibles y configurables.
Filtrado de datos mediante cláusulas
WHERE.Soporte para la eliminación de columnas.
Flink CDC vs. Flink DataStream
La siguiente tabla enumera las ventajas de los trabajos de ingestión de datos con Flink CDC frente a los trabajos DataStream.
|
Ingestión con Flink CDC |
Flink DataStream |
|
Accesible para usuarios de todos los niveles de habilidad. |
Requiere experiencia en Java y sistemas distribuidos. |
|
Oculta la complejidad subyacente y simplifica el desarrollo. |
Requiere conocimiento del framework Flink. |
|
Formato YAML fácil de aprender. |
Requiere conocimiento de herramientas como Maven para la gestión de dependencias. |
|
Alta reutilización de trabajos existentes. |
Es difícil reutilizar código existente. |
Limitaciones
Utilice Ververica Runtime (VVR) 11.1 o posterior para desarrollar trabajos de ingestión de datos con Flink CDC. Si necesita usar VVR 8.x, utilice VVR 8.0.11.
Cada trabajo admite solo un origen y un destino. Para leer desde varios orígenes o escribir en varios destinos, debe crear múltiples trabajos de Flink CDC.
Los trabajos de Flink CDC no se pueden desplegar en un clúster de sesión.
Los trabajos de ingestión de datos con Flink CDC no admiten el ajuste automático.
Conectores de ingestión de datos con Flink CDC
Para obtener detalles sobre los conectores de origen y destino compatibles con la ingestión de datos con Flink CDC, consulte Supported connectors.
Crear un trabajo de ingestión de datos con Flink CDC
A partir de una plantilla
Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, seleccione .
Haga clic en
y luego en New Draft with Template.-
Seleccione una plantilla de sincronización de datos.
Actualmente, solo están disponibles las plantillas de MySQL a StarRocks, de MySQL a Paimon y de MySQL a Hologres.
Introduzca la información del trabajo, como OK, location y engine version, y luego haga clic en OK.
-
Configure la información del origen y del destino para el trabajo de Flink CDC.
Para obtener detalles sobre la configuración de parámetros, consulte la documentación de los conectores correspondientes.
A partir de un trabajo CTAS/CDAS
Si un trabajo contiene varias instrucciones CXAS, Flink detecta y convierte únicamente la primera.
Debido a las diferencias en la compatibilidad de funciones integradas entre Flink SQL y Flink CDC, es posible que las reglas
transformgeneradas no funcionen tal cual. Debe revisarlas y ajustarlas según sea necesario.Si el origen es MySQL y un trabajo CTAS/CDAS original sigue en ejecución, debe ajustar el
server-iddel origen en el trabajo de ingestión de datos con Flink CDC para evitar conflictos.
Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, seleccione .
-
Haga clic en
y luego en New Draft from CTAS/CDAS. Seleccione el trabajo CTAS o CDAS deseado y luego haga clic en OK.En la página de selección, el sistema muestra únicamente trabajos CTAS y CDAS válidos. Los trabajos ETL regulares y los borradores con errores de sintaxis no se muestran.
Introduzca la información del trabajo, como OK, location y engine version, y luego haga clic en OK.
A partir de Flink CDC de código abierto
Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, seleccione .
Haga clic en
y luego seleccione New Draft. Introduzca el name y la engine version, y luego haga clic en Create.Pegue el código YAML de su trabajo de Flink CDC de código abierto en el editor.
-
(Opcional) Haga clic en Validate.
Esta opción comprueba errores de sintaxis, problemas de conectividad de red y problemas de permisos.
Desde cero
Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, seleccione .
Haga clic en
y luego seleccione New Draft. Introduzca el name y la engine version, y luego haga clic en Create.-
Configure el trabajo de Flink CDC mediante YAML. Ejemplo:
# Required source: # The type of the data source. type: <Replace with your source connector type> # The configuration for the data source. For details about configuration items, see the documentation for the corresponding connector. ... # Required sink: # The type of the sink. type: <Replace with your sink connector type> # The configuration for the sink. For details about configuration items, see the documentation for the corresponding connector. ... # Optional transform: # A transform rule for the flink_test.customers table. - source-table: flink_test.customers # Projection configuration. Specifies the columns to synchronize and performs data transformations. projection: id, username, UPPER(username) as username1, age, (age + 1) as age1, test_col1, __schema_name__ || '.' || __table_name__ identifier_name # Filter condition. Synchronizes only data where id is greater than 10. filter: id > 10 # Description of the transform rule. description: append calculated columns based on source table # Optional route: # A route rule that specifies the mapping between the source table and the sink table. - source-table: flink_test.customers sink-table: db.customers_o # Description of the route rule. description: sync customers table - source-table: flink_test.customers_suffix sink-table: db.customers_s # Description of the route rule. description: sync customers_suffix table # Optional pipeline: # The name of the job. name: MySQL to Hologres PipelineNotaEn un trabajo de Flink CDC, separe una clave y su valor con dos puntos y un espacio. El formato es
Key: Value.La siguiente tabla describe los bloques de código.
Obligatorio
Módulo
Descripción
Sí
source
El punto de inicio del pipeline de datos. Flink CDC captura los datos de cambio desde el origen.
Nota-
Actualmente, MySQL es el único origen compatible. Para obtener detalles específicos de configuración, consulte MySQL connector.
-
Puede utilizar variables para gestionar información confidencial. Para obtener más información, consulte Variable Management.
sink
El endpoint del pipeline de datos. Flink CDC transmite los cambios de datos capturados al sistema de destino.
Nota-
Para obtener información sobre los destinos compatibles, consulte Flink CDC data ingestion connectors. Para obtener detalles de configuración, consulte la documentación del conector específico.
-
Puede utilizar variables para gestionar información confidencial. Para obtener más información, consulte Variable Management.
No
pipeline
(pipeline de datos)
Define configuraciones básicas para todo el trabajo del pipeline de datos, como el nombre del pipeline.
transform
Especifica reglas de transformación de datos para operar sobre los datos a medida que fluyen por el pipeline de Flink. Admite procesamiento ETL, filtrado mediante cláusulas
WHERE, eliminación de columnas y columnas calculadas.Si necesita transformar los datos de cambio sin procesar capturados por Flink CDC para adaptarlos a sistemas posteriores específicos, utilice el bloque
transform.route
Si este módulo no se configura, el trabajo realiza la sincronización completa de la base de datos o de la tabla de destino de forma predeterminada.
En algunos casos, es posible que necesite enviar los datos de cambio capturados a diferentes destinos según reglas específicas. El módulo
routele permite especificar de manera flexible la relación de mapeo entre el origen y el destino, enviando datos a diferentes objetivos.Para obtener más información sobre la sintaxis y la configuración de cada módulo, consulte Reference for Flink CDC data ingestion jobs.
El siguiente código proporciona un ejemplo de sincronización de todas las tablas de la base de datos app_db en MySQL hacia una base de datos en Hologres.
source: type: mysql hostname: <hostname> port: 3306 username: ${secret_values.mysqlusername} password: ${secret_values.mysqlpassword} tables: app_db.\.* server-id: 5400-5404 # (Optional) Synchronize data from tables newly created in the incremental phase. scan.binlog.newly-added-table.enabled: true # (Optional) Synchronize table and field comments. include-comments.enabled: true # (Optional) Prioritize dispatching unbounded chunks to prevent potential TaskManager OutOfMemory issues. scan.incremental.snapshot.unbounded-chunk-first.enabled: true # (Optional) Enable parsing filters to accelerate reading. scan.only.deserialize.captured.tables.changelog.enabled: true 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 -
-
(Opcional) Haga clic en Validate.
Esta opción comprueba errores de sintaxis, problemas de conectividad de red y problemas de permisos.
Temas relacionados
Una vez que termine de desarrollar un trabajo de Flink CDC, deberá desplegarlo. Para obtener más información, consulte Deploy a job.
Para crear rápidamente un trabajo de Flink CDC que sincronice datos desde una base de datos MySQL hacia StarRocks, consulte Tutorial: Create a Flink CDC data ingestion job.