Todos los productos
Search

Este producto

    Realtime Compute for Apache Flink:Flink CDC: sync entire MySQL database to Kafka

    Centro de documentos

    Realtime Compute for Apache Flink:Flink CDC: sync entire MySQL database to Kafka

    Última actualización: Sep 03, 2026

    Este tema describe cómo sincronizar una base de datos MySQL completa con Kafka. Este enfoque reduce la carga que múltiples trabajos ejercen sobre la base de datos MySQL.

    Contexto

    Una tabla de origen MySQL CDC captura datos de MySQL y sincroniza los cambios en tiempo real de dicha tabla. Esta práctica es común en escenarios de cómputo complejos, como cuando una tabla funciona como tabla de dimensiones en una operación JOIN con otras tablas de datos. Una sola tabla de MySQL puede ser dependencia de varios trabajos. Cuando múltiples trabajos procesan datos de la misma tabla MySQL, la base de datos abre varias conexiones, lo cual ejerce una presión significativa sobre el servidor MySQL y la red.

    Funcionamiento

    Para reducir la presión sobre la base de datos MySQL ascendente, Realtime Compute for Apache Flink permite sincronizar una base de datos MySQL completa con Kafka. Esta solución introduce Kafka como capa intermedia y utiliza un trabajo de ingesta de datos de Flink CDC para sincronizar la información hacia Kafka.

    En un solo trabajo, los datos de una base de datos MySQL ascendente se sincronizan con Kafka en tiempo real. Cada tabla de MySQL se escribe en un topic de Kafka correspondiente en modo upsert. Posteriormente, los trabajos descendentes utilizan un conector upsert de Kafka para leer los datos de los topics en lugar de acceder directamente a las tablas de MySQL. Este método reduce eficazmente la presión que múltiples trabajos ejercen sobre la base de datos MySQL.

    图片 1

    Limitaciones

    • Cada tabla de MySQL que sincronice debe tener una clave primaria.

    • Puede utilizar clústeres de Kafka autogestionados, clústeres EMR Kafka o ApsaraMQ for Kafka. Si utiliza ApsaraMQ for Kafka, solo podrá conectarse mediante un default endpoint.

    • El espacio de almacenamiento del clúster de Kafka debe ser mayor que el de las tablas de origen. De lo contrario, podrían perderse datos por falta de almacenamiento. Los topics creados para la sincronización de bases de datos son topics compactados. En un topic compactado, solo se conserva el mensaje más reciente para cada clave de mensaje, pero los datos nunca caducan. Esto significa que el topic compactado almacena un volumen de datos aproximadamente equivalente al tamaño de la tabla de origen.

    Escenario de ejemplo

    Por ejemplo, en un escenario de análisis de revisión de pedidos en tiempo real, suponga que existen tres tablas: una tabla de usuarios (user), una tabla de pedidos (order) y una tabla de comentarios de usuarios (feedback). Las tablas contienen los datos que se muestran en la siguiente figura.mysql database

    Para mostrar la información de los pedidos y las reseñas de los usuarios, es necesario unir la tabla user para obtener los nombres de usuario del campo name. El siguiente ejemplo de SQL demuestra esta operación.

    -- Join order information with the user table to display the username and product name for each order.
SELECT order.id as order_id, product, user.name as user_name
FROM order LEFT JOIN user
ON order.user_id = user.id;
-- Join reviews with the user table to display the content of each review and the corresponding username.
SELECT feedback.id as feedback_id, comment, user.name as user_name
FROM feedback LEFT JOIN user
ON feedback.user_id = user.id;

    En los dos trabajos SQL anteriores, ambos utilizan la tabla user. Durante la ejecución, ambos trabajos leen los datos completos e incrementales de MySQL. Una lectura completa requiere crear una conexión a MySQL, mientras que una lectura incremental requiere crear un cliente de Binlog. A medida que aumenta el número de trabajos, también crece la demanda de recursos de conexión a MySQL y de clientes de Binlog, lo que ejerce una presión significativa sobre la base de datos ascendente. Para aliviar esta presión, puede utilizar un trabajo de ingesta de datos de Flink CDC para sincronizar los datos de la base de datos MySQL ascendente con Kafka en tiempo real, permitiendo así su consumo por parte de múltiples trabajos descendentes.

    Requisitos previos

    Preparación

    Preparar el origen de datos de MySQL

    1. Cree una base de datos ApsaraDB RDS for MySQL. Para obtener más información, consulte Create a database.

      Cree una base de datos llamada order_dw para la instancia de destino.

    2. Prepare el origen de datos MySQL CDC.

      1. En la página de detalles de la instancia, haga clic en Log on to Database en la parte superior de la página.

      2. En el cuadro de diálogo de inicio de sesión de DMS que aparece, introduzca el nombre de usuario y la contraseña de la cuenta de base de datos que creó, y luego haga clic en Login.

      3. Después de iniciar sesión, haga doble clic en la base de datos order_dw en el panel izquierdo para cambiar de base de datos.

      4. En la consola SQL, introduzca las sentencias DDL para crear las tres tablas de negocio y las sentencias para insertar datos.

        CREATE TABLE `user` (
  id bigint not null primary key,
  name varchar(50) not null
);
CREATE TABLE `order` (
  id bigint not null primary key,
  product varchar(50) not null,
  user_id bigint not null
);
CREATE TABLE `feedback` (
  id bigint not null primary key,
  user_id bigint not null,
  comment varchar(50) not null
);
-- Prepare data
INSERT INTO `user` VALUES(1, 'Tom'),(2, 'Jerry');
INSERT INTO `order` VALUES
(1, 'Football', 2),
(2, 'Basket', 1);
INSERT INTO `feedback` VALUES
(1, 1, 'Good.'),
(2, 2, 'Very good');

    3. Haga clic en Execute y, a continuación, haga clic en Execute.

    Procedimiento

    1. Cree e inicie un trabajo de ingesta de datos de Flink CDC para sincronizar los datos de la base de datos MySQL ascendente con Kafka en tiempo real, de modo que puedan ser consumidos por múltiples trabajos descendentes. El trabajo de sincronización de base de datos crea automáticamente los topics. Puede definir los nombres de los topics mediante el módulo route. Los topics utilizan la configuración predeterminada del clúster de Kafka para el número de particiones y réplicas, y cleanup.policy se establece en compact.

      Nombres de topics predeterminados

      De forma predeterminada, los topics de Kafka creados por el trabajo de sincronización de base de datos utilizan el formato de nomenclatura {database_name}.{table_name}. El siguiente trabajo crea tres topics: order_dw.user, order_dw.order y order_dw.feedback.

      1. En la página Development > Data Ingestion, cree un trabajo de ingesta de datos de Flink CDC y copie el siguiente código en el editor YAML.

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{usernmae}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (Optional) Synchronize data from newly created tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize dispatching unbounded splits to avoid potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # The following parameters are required for ApsaraMQ for Kafka.
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. En la esquina superior derecha, haga clic en Deploy para desplegar el trabajo.

      3. En la barra de navegación izquierda, haga clic en O&M > Deployments. En la columna Actions del trabajo objetivo, haga clic en Start, seleccione Initial Mode y, a continuación, haga clic en Start.

      Nombres de topics por tabla

      Puede utilizar el módulo route para especificar un nombre de topic para cada tabla. El siguiente trabajo crea tres topics: user1, order2 y feedback3.

      1. En la página Development > Data Ingestion, cree un trabajo de ingesta de datos de Flink CDC y copie el siguiente código en el editor YAML.

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{usernmae}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (Optional) Synchronize data from newly created tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize dispatching unbounded splits to avoid potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  - source-table: order_dw.user
    sink-table: user1
  - source-table: order_dw.order
    sink-table: order2
  - source-table: order_dw.feedback
    sink-table: feedback3
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # The following parameters are required for ApsaraMQ for Kafka.
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. En la esquina superior derecha, haga clic en Deploy para desplegar el trabajo.

      3. En la barra de navegación izquierda, seleccione O&M > Deployments, haga clic en Start en la columna Actions del trabajo objetivo, seleccione Initial Mode y, a continuación, haga clic en Start.

      Nombres de topics por lotes

      Puede utilizar el módulo route para especificar un patrón para los nombres de los topics generados. El siguiente trabajo crea tres topics: topic_user, topic_order y topic_feedback.

      1. En la página Development > Data Ingestion, cree un trabajo de ingesta de datos de Flink CDC y copie el siguiente código en el editor YAML.

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{usernmae}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (Optional) Synchronize data from newly created tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize dispatching unbounded splits to avoid potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  - source-table: order_dw.\.*
    sink-table: topic_<>
    replace-symbol: <>
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # The following parameters are required for ApsaraMQ for Kafka.
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. En la esquina superior derecha, haga clic en Deploy para desplegar el trabajo.

      3. En la barra de navegación izquierda, haga clic en O&M > Deployments. Haga clic en Start en la columna Actions del trabajo objetivo, seleccione Initial Mode y, a continuación, haga clic en Start.

    1. Consuma datos de Kafka en tiempo real.

      El trabajo de ingesta de datos escribe la información de la base de datos MySQL ascendente en Kafka en formato JSON. Posteriormente, múltiples trabajos descendentes pueden consumir datos de un solo topic para recuperar el estado más reciente de las tablas de la base de datos. Puede consumir los datos de las tablas sincronizadas con Kafka de una de las siguientes maneras:

      Mediante catálogo

      Lea datos de un topic de Kafka utilizándolo como tabla de origen.

      1. En la página Development > ETL, cree un trabajo SQL de streaming y copie el siguiente código en el editor SQL.

        CREATE TEMPORARY TABLE print_user_proudct(
  order_id BIGINT,
  product STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
CREATE TEMPORARY TABLE print_user_feedback(
  feedback_id BIGINT,
  `comment` STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
BEGIN STATEMENT SET;      -- Required when writing to multiple sinks.
-- Join order information with the user table in the Kafka JSON Catalog to display the username and product name for each order.
INSERT INTO print_user_proudct
SELECT `order`.key_id as order_id, value_product as product, `user`.value_name as user_name
FROM `kafka-catalog`.`kafka`.`order`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/  as `order` -- Specify the group and startup mode.
LEFT JOIN `kafka-catalog`.`kafka`.`user`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/ as `user` -- Specify the group and startup mode.
ON `order`.value_user_id = `user`.key_id;
-- Join reviews with the user table to display the content of each review and the corresponding username.
INSERT INTO print_user_feedback
SELECT feedback.key_id as feedback_id, value_comment as `comment`, `user`.value_name as user_name
FROM `kafka-catalog`.`kafka`.feedback/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/  as feedback  -- Specify the group and startup mode.
LEFT JOIN `kafka-catalog`.`kafka`.`user`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/ as `user` -- Specify the group and startup mode.
ON feedback.value_user_id = `user`.key_id;
END;      -- Required when writing to multiple sinks.

        Este ejemplo utiliza el conector Print para imprimir los resultados directamente. También puede enviar los resultados a una tabla de resultados que utilice otro conector para un análisis posterior. Para obtener más información sobre la sintaxis para escribir en múltiples destinos, consulte INSERT INTO statement.

        Nota

        Cuando utiliza este método directamente, pueden producirse cambios en el esquema. Como resultado, el esquema analizado por el catálogo JSON de Kafka puede diferir del esquema de la tabla MySQL correspondiente. Por ejemplo, los campos eliminados pueden seguir apareciendo y algunos campos pueden tener valores nulos.

        El esquema leído del catálogo consta de campos provenientes de los datos consumidos. Si se elimina un campo pero sus mensajes no han caducado, el campo puede seguir apareciendo con un valor nulo. No se requiere ningún manejo especial para este caso.

      2. En la esquina superior derecha, haga clic en Deploy para desplegar el trabajo.

      3. En la barra de navegación izquierda, haga clic en O&M > Deployments, haga clic en Start en la columna Actions del trabajo objetivo, seleccione Initial Mode y, a continuación, haga clic en Start.

      Mediante tabla temporal

      Defina un esquema personalizado y lea datos desde una tabla temporal.

      1. En la página Development > ETL, cree un trabajo SQL de streaming y copie el siguiente código en el editor SQL.

        CREATE TEMPORARY TABLE user_source (
  key_id BIGINT,
  value_name STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE order_source (
  key_id  BIGINT,
  value_product STRING,
  value_user_id BIGINT  
) WITH (
  'connector' = 'kafka',
  'topic' = 'order',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE feedback_source (
  key_id  BIGINT,
  value_user_id BIGINT,
  value_comment STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'feedback',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE print_user_proudct(
  order_id BIGINT,
  product STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
CREATE TEMPORARY TABLE print_user_feedback(
  feedback_id BIGINT,
  `comment` STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
BEGIN STATEMENT SET;      -- Required when writing to multiple sinks.
-- Join order information with the user table from the Kafka JSON Catalog to display the username and product name for each order.
INSERT INTO print_user_proudct
SELECT order_source.key_id as order_id, value_product as product, user_source.value_name as user_name
FROM order_source LEFT JOIN user_source
ON order_source.value_user_id = user_source.key_id;
-- Join reviews with the user table to display the content of each review and the corresponding username.
INSERT INTO print_user_feedback
SELECT feedback_source.key_id as feedback_id, value_comment as `comment`, user_source.value_name as user_name
FROM feedback_source  LEFT JOIN user_source
ON feedback_source.value_user_id = user_source.key_id;
END;      -- Required when writing to multiple sinks.

        Este ejemplo utiliza el conector Print para imprimir los resultados directamente. También puede enviar los resultados a una tabla de resultados de otro conector para un análisis posterior. Para obtener más información sobre la sintaxis para escribir en múltiples destinos, consulte INSERT INTO statement.

        La siguiente tabla describe los parámetros de configuración de la tabla temporal.

        Parámetro

        Descripción

        Notas

        connector

        El tipo de conector.

        Establezca el valor en kafka.

        topic

        El nombre del topic correspondiente.

        Debe coincidir con la descripción del catálogo JSON de Kafka.

        properties.bootstrap.servers

        Las direcciones de los brokers de Kafka.

        El formato es host:port,host:port,host:port, separado por comas (,).

        scan.startup.mode

        La posición inicial para leer datos de Kafka.

        Valores válidos:

        • earliest-offset: Comienza a leer desde el offset disponible más antiguo.

        • latest-offset: Comienza a leer desde el offset más reciente.

        • group-offsets (predeterminado): Lee desde el offset confirmado para el grupo especificado por properties.group.id.

        • timestamp: Lee desde la marca de tiempo especificada por scan.startup.timestamp-millis.

        • specific-offsets: Comienza a leer desde los offsets especificados en scan.startup.specific-offsets.

        Nota

        Este parámetro surte efecto cuando el trabajo se inicia sin un estado guardado. Cuando el trabajo se reinicia o se recupera de un punto de control, prioriza la lectura desde el estado guardado.

        key.format

        El formato que utiliza el conector Flink Kafka para serializar o deserializar la clave del mensaje de Kafka.

        Establezca el valor en json.

        key.fields

        Los campos de la tabla de origen o de resultados que corresponden a la clave del mensaje de Kafka.

        Utilice punto y coma (;) para separar varios nombres de campo. Por ejemplo, field1;field2.

        key.fields-prefix

        Un prefijo personalizado para todos los campos de clave de mensaje de Kafka, con el fin de evitar conflictos de nomenclatura con los campos de valor del mensaje o los campos de metadatos.

        Este valor debe coincidir con el valor del parámetro key.fields-prefix del catálogo JSON de Kafka.

        value.format

        El formato que utiliza el conector Flink Kafka para serializar o deserializar el valor del mensaje de Kafka.

        Establezca el valor en json.

        value.fields-prefix

        Un prefijo personalizado para todos los campos de valor de mensaje de Kafka, con el fin de evitar conflictos de nomenclatura con los campos de clave del mensaje o los campos de metadatos.

        Debe coincidir con el valor del parámetro value.fields-prefix del catálogo JSON de Kafka.

        value.fields-include

        La directiva para manejar los campos de clave de mensaje dentro del valor del mensaje.

        Establezca el valor en EXCEPT_KEY. Esto indica que el valor del mensaje no incluye los campos de la clave del mensaje.

        value.json.infer-schema.flatten-nested-columns.enable

        Especifica si se deben expandir recursivamente las columnas JSON anidadas en el valor del mensaje de Kafka.

        El valor del parámetro infer-schema.flatten-nested-columns.enable del catálogo correspondiente.

        value.json.infer-schema.primitive-as-string

        Especifica si se deben inferir todos los tipos primitivos como String en el valor del mensaje de Kafka.

        El valor del parámetro infer-schema.primitive-as-string del catálogo correspondiente.

      2. En la esquina superior derecha, haga clic en Deploy para desplegar el trabajo.

      3. En la barra de navegación izquierda, haga clic en O&M > Deployments, haga clic en Start en la columna Actions del trabajo objetivo, seleccione Initial Mode y, a continuación, haga clic en Start.

    2. Consulte los resultados del trabajo.

      1. En la barra de navegación izquierda, haga clic en O&M > Deployments y, a continuación, haga clic en el trabajo objetivo.

      2. En la pestaña Logs, dentro de la pestaña Running Task Managers, haga clic en la tarea que tenga la Path, ID que desea consultar.

      3. Haga clic en Logs y busque información de registro relacionada con PrintSinkOutputWriter.

        Busque en los registros la salida de PrintSinkOutputWriter. La salida contiene cuatro registros de datos unidos: +I[1, Good., Tom], +I[2, Very good, Jerry], +I[2, Basket, Tom] y +I[1, Football, Jerry]. Esto indica que las uniones entre la tabla de usuarios y las tablas de pedidos y comentarios se realizaron correctamente.

    Documentos relacionados