Quick start for Flink CDC data ingestion jobs - Realtime Compute for Apache Flink

Realtime Compute for Apache Flink ofrece una potente función de ingesta de datos basada en Flink CDC. Esta guía muestra cómo construir un trabajo de ingesta de datos con Flink CDC para sincronizar una base de datos MySQL completa con una base de datos StarRocks.

Requisitos previos

Se ha creado un espacio de trabajo de Flink. Para más información, consulte Activate Realtime Compute for Apache Flink.
Almacenes de datos de origen y destino
- Se ha creado una instancia de ApsaraDB RDS for MySQL. Para más información, consulte (Deprecated, redirected to "Step 1") Quickly create an ApsaraDB RDS for MySQL instance.
- Se ha creado una instancia de EMR Serverless StarRocks. Para más información, consulte Procedure.
Nota
La instancia de ApsaraDB RDS for MySQL y la instancia de EMR Serverless StarRocks deben estar en la misma Virtual Private Cloud (VPC) que su espacio de trabajo de Flink. Si se encuentran en VPC diferentes, debe establecer una conexión de red y configurar una lista de permitidos para la instancia de ApsaraDB RDS for MySQL. Para más información, consulte How do I access other services across VPCs?, How do I access the Internet? y How do I configure an allowlist?.

Contexto

Suponga que su instancia de ApsaraDB RDS for MySQL tiene una base de datos llamada order_dw_mysql que contiene tres tablas de negocio: orders, orders_pay y product_catalog. Para sincronizar estas tablas y sus datos con la base de datos order_dw_sr en StarRocks, siga estos pasos:

Step 1: Prepare test data in ApsaraDB RDS for MySQL
Step 2: Develop a Flink CDC data ingestion job
Step 3: Start the Flink CDC data ingestion job
Step 4: Verify the synchronization results in StarRocks

Paso 1: Preparar datos de prueba de MySQL

Cree una base de datos y una cuenta.

Cree una base de datos llamada order_dw_mysql y una cuenta estándar con permisos de lectura y escritura sobre ella. Para más información, consulte (Deprecated, redirected to "Step 1") Create a database and an account y Manage databases.
Inicie sesión en la instancia de ApsaraDB RDS for MySQL mediante Data Management (DMS).

Para más información, consulte (Deprecated, redirected to "Step 2") Log on to an ApsaraDB RDS for MySQL instance by using DMS.

En la ventana SQL Console, introduzca los siguientes comandos y haga clic en Execute para crear tres tablas de negocio e insertar datos.

CREATE TABLE `orders` (
  order_id bigint not null primary key,
  user_id varchar(50) not null,
  shop_id bigint not null,
  product_id bigint not null,
  buy_fee numeric(20,2) not null,   
  create_time timestamp not null,
  update_time timestamp not null default now(),
  state int not null 
);
CREATE TABLE `orders_pay` (
  pay_id bigint not null primary key,
  order_id bigint not null,
  pay_platform int not null, 
  create_time timestamp not null
);
CREATE TABLE `product_catalog` (
  product_id bigint not null primary key,
  catalog_name varchar(50) not null
);
-- Prepare data
INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
INSERT INTO orders VALUES
(100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
(100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
(100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
(100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
(100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
(100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
(100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
INSERT INTO orders_pay VALUES
(2001, 100001, 1, '2023-02-15 17:40:56'),
(2002, 100002, 1, '2023-02-15 17:40:56'),
(2003, 100003, 0, '2023-02-15 17:40:56'),
(2004, 100004, 0, '2023-02-15 17:40:56'),
(2005, 100005, 0, '2023-02-15 18:40:56'),
(2006, 100006, 0, '2023-02-15 18:40:56'),
(2007, 100007, 0, '2023-02-15 18:40:56');

Paso 2: Desarrollar un trabajo de Flink CDC

Inicie sesión en la consola de administración de Realtime Compute for Apache Flink.
Haga clic en Console para ir al espacio de trabajo del proyecto.
En el panel de navegación izquierdo, seleccione Development > Data Ingestion.
Haga clic en el icono , haga clic en New Draft with Template, seleccione MySQL to StarRocks data synchronization y luego haga clic en Next.
Introduzca un Job Name y una Location, seleccione una Engine Version y luego haga clic en OK.

Configure el código YAML del trabajo.

El siguiente código proporciona un ejemplo para sincronizar todas las tablas de la base de datos order_dw_mysql en MySQL con la base de datos order_dw_sr en StarRocks.

source:
  type: mysql
  hostname: rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: order_dw_mysql.\.*
  server-id: 8601-8604
  # (Optional) Synchronize data from tables that are newly created during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded chunks to prevent potential TaskManager OutOfMemory errors.
  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: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030
  load-url: fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030
  username: ${secret_values.starrocksusername}
  password: ${secret_values.starrockspassword}
  table.create.properties.replication_num: 1
  sink.buffer-flush.interval-ms: 5000 # Flush data every 5 seconds.
route:
  - source-table: order_dw_mysql.\.*
    sink-table: order_dw_sr.<>
    replace-symbol: <>
    description: route all tables in source_db to sink_db
pipeline:
  name: Sync MySQL Database to StarRocks

La siguiente tabla describe los parámetros de configuración obligatorios para este ejemplo. Para más información sobre los parámetros de ingesta de datos, consulte MySQL y StarRocks.

Nota

Los trabajos YAML solo admiten variables de proyecto. Puede usar variables para evitar que información como contraseñas se muestre en texto sin formato. Para más información, consulte Variable Management.

Categoría	Parámetro	Descripción	Valor de ejemplo
source	hostname	La dirección IP o nombre de host de la base de datos MySQL. Se recomienda utilizar el endpoint interno.	`rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com`
	port	El número de puerto del servicio de base de datos MySQL.	3306
	username	El nombre de usuario y la contraseña para la base de datos MySQL. Utilice las credenciales de la cuenta creadas en Step 1: Prepare test data in ApsaraDB RDS for MySQL.	`${secret_values.mysqlusername}`
	password		`${secret_values.mysqlpassword}`
	tables	Los nombres de las tablas MySQL. Puede usar expresiones regulares para leer datos de varias tablas. En este tema, se sincronizan todas las tablas y datos de la base de datos order_dw_mysql.	order_dw_mysql.\.*
	server-id	Un ID numérico único para la conexión del cliente a la base de datos.	8601-8604
sink	jdbc-url	La URL de conexión JDBC. Especifique la dirección IP y el puerto de consulta del Frontend (FE) con el formato `jdbc:mysql://ip:port`. En la pestaña Instance Details de la consola E-MapReduce, puede consultar el internal endpoint y el query port del FE para la instancia de destino.	`jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030`
	load-url	La URL del servicio HTTP utilizada para conectarse al nodo FE. En la pestaña Instance Details de la consola E-MapReduce, puede consultar el internal endpoint y el HTTP port del FE para la instancia de destino.	`fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030`
	username	Las credenciales para conectarse a StarRocks. Utilice las credenciales que configuró al crear la instancia de StarRocks. Nota Este ejemplo utiliza variables para evitar exponer las credenciales en texto sin formato. Para más información, consulte Manage variables.	`${secret_values.starrocksusername}`
	password		`${secret_values.starrockspassword}`
	sink.buffer-flush.interval-ms	El intervalo de vaciado para el búfer interno. Se utiliza un intervalo corto (5 segundos) porque este ejemplo contiene una pequeña cantidad de datos, lo que permite ver los resultados rápidamente.	5000
route	source-table	La tabla o tablas de origen que se van a enrutar. Puede usar una expresión regular para coincidir con varias tablas. Por ejemplo, `order_dw_mysql.\.*` enruta todas las tablas de la base de datos `order_dw_mysql`.	order_dw_mysql.\.*
	sink-table	El patrón de tabla de destino para los datos enrutados. Puede usar el símbolo definido en el parámetro `replace-symbol` como marcador de posición para cada nombre de tabla de origen y lograr un enrutamiento de muchos a muchos. Para más información sobre las reglas de enrutamiento, consulte Route module.	order_dw_sr.<>
	replace-symbol	El marcador de posición para el nombre de la tabla de origen utilizado en la coincidencia de patrones.	<>

Haga clic en Deploy.

Paso 3: Iniciar el trabajo de Flink CDC

En la página Data Ingestion, haga clic en Deploy y luego haga clic en OK en el cuadro de diálogo que aparece.
En la página O&M > Deployments, busque el trabajo YAML de destino y haga clic en Start en la columna Actions.
Haga clic en Start.

En este ejemplo, seleccione Initial Mode. Para más información sobre los parámetros, consulte Start a job. Una vez iniciado el trabajo, puede supervisar su estado en la página Deployments.

Paso 4: Verificar resultados en StarRocks

Después de que el trabajo entre en estado RUNNING, puede verificar los datos en StarRocks.

Connect to an EMR Serverless StarRocks instance by using EMR StarRocks Manager.
En el panel de navegación izquierdo, haga clic en SQL Editor. En la pestaña Database, haga clic en el icono de actualización .

Aparece una base de datos llamada order_dw_sr bajo default_catalog.

En la pestaña Query List, haga clic en + File para crear un elemento en Scripts. Introduzca las siguientes sentencias SQL y haga clic en Run.

SELECT * FROM default_catalog.order_dw_sr.orders order by order_id;
SELECT * FROM default_catalog.order_dw_sr.orders_pay order by pay_id;
SELECT * FROM default_catalog.order_dw_sr.product_catalog order by product_id;

Consulte los resultados debajo de los comandos.

Los resultados muestran que las tablas y los datos de la base de datos MySQL ahora existen en StarRocks.

Las tablas sincronizadas incluyen default_catalog.order_dw_sr.orders, default_catalog.order_dw_sr.orders_pay y default_catalog.order_dw_sr.product_catalog. Puede ejecutar sentencias SELECT para consultar cada tabla y verificar la integridad de los datos.

Documentación relacionada

Para obtener pasos detallados sobre cómo desarrollar un trabajo de ingesta de datos con Flink CDC, consulte Develop a Flink CDC data ingestion job.
Para más información sobre los módulos source, sink, transform y route para trabajos de ingesta de datos con Flink CDC, consulte Flink CDC data ingestion job development reference.