ApsaraMQ for Kafka:Create a MySQL source connector

How it works

When you create a MySQL source connector through the ApsaraMQ for Kafka console, the system automatically:

1. Activates DataWorks Basic Edition (free of charge).

2. Creates a DataWorks workspace and an exclusive resource group for data integration.

3. Generates destination topics in your Kafka instance -- one topic per source table.

The exclusive resource group (4 vCPUs, 8 GB memory) runs the data synchronization tasks. Resource groups use monthly subscriptions and renew automatically upon expiration.

Topic naming and partitions

DataWorks generates one destination topic per source table using the naming pattern:

<topic-prefix>_<source-table-name>

The underscore is added automatically. For example, if the prefix is mysql and the source tables are table_1, table_2, and table_n, the generated topics are mysql_table_1, mysql_table_2, and mysql_table_n.

Partition count depends on whether the source table has a primary key:

Source table type	Partitions per topic
Has a primary key	6
No primary key	1

Make sure your Kafka instance has enough available topics and partitions. If the instance runs out of either, topic creation fails and the connector cannot be deployed.

Prerequisites

Before you begin, make sure that you have:

• An ApsaraMQ for Kafka instance with the connector feature enabled, deployed in one of these regions: China (Shenzhen), China (Chengdu), China (Beijing), China (Zhangjiakou), China (Hangzhou), China (Shanghai), or Singapore

• An ApsaraDB RDS for MySQL instance with a database, database account, and at least one table created. For SQL reference, see Commonly used SQL statements for MySQL

• Non-overlapping CIDR blocks between the VPC of your RDS MySQL instance and the VPC of your Kafka instance

• DataWorks authorized to access your ENIs through the Cloud Resource Access Authorization page

• Both the data source (ApsaraDB RDS for MySQL instance) and the data destination (ApsaraMQ for Kafka instance) created under the same Alibaba Cloud account

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '<your-username>'@'%';

Create and deploy the connector

1. Log on to the ApsaraMQ for Kafka console.

2. In the Resource Distribution section of the Overview page, select the region of your instance.

3. In the left-side navigation pane, click Connectors.

4. On the Connectors page, select your instance from the Select Instance drop-down list and click Create Connector.

5. Complete the three-step wizard: Click Next. Step 2: Configure source service Select ApsaraDB RDS for MySQL as the source service and configure the following parameters: Click Next. Step 3: Configure destination service Confirm the destination Kafka instance and click Create.

   Step 1: Configure basic information

   Parameter	Description	Example
   Name	A unique connector name within the instance. 1--48 characters; digits, lowercase letters, and hyphens (-) allowed. Cannot start with a hyphen. The system auto-creates a consumer group named connect-<connector-name>.	kafka-source-mysql
   Instance	Displays the name and ID of the selected Kafka instance.	demo alikafka_post-cn-st21p8vj****

   Parameter	Description	Example
   Region of ApsaraDB RDS for MySQL Instance	Region of the source RDS MySQL instance.	China (Shenzhen)
   ApsaraDB RDS for MySQL Instance ID	Instance ID of the source database.	rm-wz91w3vk6owmz****
   Database Name	Name of the database to synchronize.	mysql-to-kafka
   Database Account	Username for the database connection.	mysql_to_kafka
   Password of Database Account	Password for the database connection.	--
   Database Table	One or more table names, separated by commas (,). Each table maps to one topic.	mysql_tbl
   Tables to Automatically Add	A regular expression to auto-detect and synchronize new tables. Use .* for all tables.	.*
   Topic Prefix	Prefix for auto-generated topic names. Must be globally unique.	mysql

6. On the Connectors page, find the connector and click Deploy in the Actions column. When the Status column shows Running, the connector is active and synchronizing data.

Verify data synchronization

1. Insert a test row into a source table: For more SQL examples, see Commonly used SQL statements for MySQL.

      INSERT INTO mysql_tbl
          (mysql_title, mysql_author, submission_date)
          VALUES
          ("mysql2kafka", "tester", NOW());

2. Use the message query feature in ApsaraMQ for Kafka to confirm the data arrived in the corresponding topic. For instructions, see Query messages. A successfully synchronized INSERT event looks like this:

      {
          "schema": {
              "dataColumn": [
                  {
                      "name": "mysql_id",
                      "type": "LONG"
                  },
                  {
                      "name": "mysql_title",
                      "type": "STRING"
                  },
                  {
                      "name": "mysql_author",
                      "type": "STRING"
                  },
                  {
                      "name": "submission_date",
                      "type": "DATE"
                  }
              ],
              "primaryKey": [
                  "mysql_id"
              ],
              "source": {
                  "dbType": "MySQL",
                  "dbName": "mysql_to_kafka",
                  "tableName": "mysql_tbl"
              }
          },
          "payload": {
              "before": null,
              "after": {
                  "dataColumn": {
                      "mysql_title": "mysql2kafka",
                      "mysql_author": "tester",
                      "submission_date": 1614700800000
                  }
              },
              "sequenceId": "1614748790461000000",
              "timestamp": {
                  "eventTime": 1614748870000,
                  "systemTime": 1614748870925,
                  "checkpointTime": 1614748870000
              },
              "op": "INSERT",
              "ddl": null
          },
          "version": "0.0.1"
      }

What's next

• Query messages in your Kafka topics

• Enable the connector feature for additional instances