MaxCompute:Open-source Flink CDC for near-real-time writes to Delta Table

Prepare environments

Submit the YAML file by using the Flink CDC CLI

1. Download the required JAR packages:

   • flink-cdc package

     Go to the flink-cdc page to download the binary compressed package flink-cdc-3.1.1-bin.tar.gz and decompress the package to obtain the flink-cdc-3.1.1 directory. The flink-cdc-3.1.1 directory contains the bin, lib, log, and conf directories. Then, move the files in these directories to the related directories for flink-1.18.0.

   • Connector packages

     Download the following connector packages and move the packages to the flink-1.18.0/lib directory.

     Note

     You can click the download links to download only the released connector versions. If you want to reference a SNAPSHOT version, you must compile the source code of the version based on master or release branches on your on-premises machine.

     • MySQL pipeline connector 3.1.1

     • MaxCompute pipeline connector 3.1.1

   • Driver package

     Download the MySQL Connector Java package and pass the package to the Flink CDC CLI by using the --jar parameter, or place the package in the $flink-1.18.0/lib directory and restart the Flink cluster. This is because CDC connectors no longer contain these drivers.

2. Write a YAML file for task configuration. The following sample code provides an example of the file mysql-to-maxcompute.yaml for database synchronization:

   ################################################################################
   # Description: Sync MySQL all tables to MaxCompute
   ################################################################################
   source:
     type: mysql
     hostname: localhost
     port: 3306
     username: root
     password: 123456
     tables: app_db.\.*
     server-id: 5400-5404
     server-time-zone: UTC
   
   # Configure the accessId, accessKey, endpoint, and project parameters.
   sink:
      type: maxcompute
      name: MaxComputeSink
      accessId: ${your_accessId}
      accessKey: ${your_accessKey}
      endpoint: ${your_maxcompute_endpoint}
      project: ${your_project}
      bucketsNum: 8
   
   pipeline:
     name: Sync MySQL Database to MaxCompute
     parallelism: 1
   

   Parameters:

   • For more information about the parameters in the source section, see MySQL Connector.

   • For more information about the parameters in the sink section, see Configuration items of the connector.

3. Run the following command to submit the task to the Flink cluster that is deployed in standalone mode:

   ./bin/flink-cdc.sh mysql-to-maxcompute.yaml

   After the task is submitted, the following result is returned:

   Pipeline has been submitted to cluster.
   Job ID: f9f9689866946e25bf151ecc179ef46f
   Job Description: Sync MySQL Database to MaxCompute

   In the Flink web UI, a task named Sync MySQL Database to MaxCompute is running.

4. Execute the following SQL statements in MaxCompute to check whether the orders, shipments, and products tables are created and data can be written to the tables.

   -- Query the orders table.
   read orders;
   
   -- The following result is returned:
   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+
   
   -- Query the shipments table.
   read shipments;
   
   -- The following result is returned:
   +------------+------------+
   | id         | city       |
   +------------+------------+
   | 1          | beijing    |
   | 2          | xian       |
   +------------+------------+
   
   -- Query the products table.
   read products;
   
   -- The following result is returned:
   +------------+------------+
   | id         | product    |
   +------------+------------+
   | 3          | Peanut     |
   | 1          | Beer       |
   | 2          | Cap        |
   +------------+------------+

Synchronize changes in real time

In this example, the orders table is used. When data in the source table in the MySQL database is modified, the data in the destination MaxCompute table is also modified in real time.

1. Run the following command to access the MySQL container:

   docker-compose exec mysql mysql -uroot -p123456

2. Insert a data record into the orders table of MySQL.

   INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

   Run the read orders; command in MaxCompute to query data in the orders table. The following result is returned:

   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 3          | 100        |
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+

3. Add a field to the orders table of MySQL.

   ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

   Run the read orders; command in MaxCompute to query data in the orders table. The following result is returned:

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 4          | NULL       |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

4. Update a data record in the orders table of MySQL.

   UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

   Run the read orders; command in MaxCompute to query data in the orders table. The following result is returned:

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

5. Delete a data record from the orders table of MySQL.

   DELETE FROM app_db.orders WHERE id=2;

   Run the read orders; command in MaxCompute to query data in the orders table. The following result is returned:

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   +------------+------------+------------+

Each time you perform an operation in MySQL, a data preview is performed in MaxCompute. The data displayed in the orders table in MaxCompute is updated in real time.

Route the changes

Flink CDC provides the feature to route the schema or data of a source table to other table names. You can use this feature to perform operations such as table or database name replacement, and database synchronization. The following sample code provides an example.

################################################################################
# Description: Sync MySQL all tables to MaxCompute
################################################################################
source:
   type: mysql
   hostname: localhost
   port: 3306
   username: root
   password: 123456
   tables: app_db.\.*
   server-id: 5400-5404
   server-time-zone: UTC

# Configure the accessId, accessKey, endpoint, and project parameters.
sink:
   type: maxcompute
   name: MaxComputeSink
   accessId: ${your_accessId}
   accessKey: ${your_accessKey}
   endpoint: ${your_maxcompute_endpoint}
   project: ${your_project}
   bucketsNum: 8

route:
   - source-table: app_db.orders
     sink-table: ods_db.ods_orders
   - source-table: app_db.shipments
     sink-table: ods_db.ods_shipments
   - source-table: app_db.products
     sink-table: ods_db.ods_products

pipeline:
   name: Sync MySQL Database to MaxCompute
   parallelism: 1

For more information about the parameters in the route section, see Route.

The preceding configuration in the route section is used to synchronize the schema and data of the app_db.orders table to the ods_db.ods_orders table. This way, database migration is implemented. You can use regular expressions to match multiple tables in source-table to synchronize data from multiple tables in a sharded database. Sample code:

route:
  - source-table: app_db.order\.*
    sink-table: ods_db.ods_orders

In this case, data of tables, such as app_db.order01, app_db.order02, and app_db.order03, can be aggregated into the ods_db.ods_orders table.

Clear the environments

After you perform the preceding operations, you must clear the environments.

1. Run the following command in the directory in which the docker-compose.yml file is stored to stop all containers:

   docker-compose down

2. Run the following command in the flink-1.18.0 directory in which Flink resides to stop the Flink cluster:

   ./bin/stop-cluster.sh

Configuration items of the connector

Table location mappings

When the Flink CDC connector automatically creates a table, the location information of the source table is mapped to the MaxCompute table based on the mappings described in the following table.

Data type mappings