ApsaraMQ for Kafka：Create a MySQL source connector

Precautions

• Regions
  • Your data source and destination Message Queue for Apache Kafka instance may not be deployed in the same region. In this case, make sure that you have a Cloud Enterprise Network (CEN) instance within the Alibaba Cloud account and that the VPCs of your data source and Message Queue for Apache Kafka instance are attached to the CEN instance. In addition, make sure that the bandwidth for cross-region connections is configured to ensure end-to-end connectivity between the data source and the data destination.

    Otherwise, a CEN instance may be automatically created, and the VPCs of your destination Message Queue for Apache Kafka instance and the Elastic Compute Service (ECS) instance where your exclusive resource group resides are attached to the CEN instance to ensure end-to-end connectivity. The bandwidth of the automatically created CEN instance is extremely low because the bandwidth is not manually configured. This may cause a connectivity error when you create a MySQL source connector or when the connector is running.

  • Your data source and destination Message Queue for Apache Kafka instance may be in the same region. In this case, when you create a MySQL source connector, an ENI is automatically created in the VPC of the data source or Message Queue for Apache Kafka instance. The ENI is also automatically bound to the ECS instance where your exclusive resource group resides. This ensures end-to-end connectivity between the data source and the data destination.
• Exclusive resource groups in DataWorks
  • DataWorks allows you to use each exclusive resource group to run up to three MySQL source connectors. If DataWorks finds that an existing resource group has been used to run less than three MySQL source connectors when you create a MySQL source connector, DataWorks uses this resource group to run the newly created MySQL source connector.
  • Each exclusive resource group in DataWorks can be associated with the ENIs of up to two VPCs. DataWorks may fail to create a MySQL source connector by using a resource group if the ENI to be associated with the resource group and ENIs that are already associated with the resource group have overlapping CIDR blocks or due to other technical limits. In this case, even if less than three connectors are running on this resource group, DataWorks still creates a resource group to ensure that the MySQL source connector can be created.

Create and deploy a MySQL source connector

1. Log on to the ApsaraMQ for Kafka console.
2. In the Resource Distribution section of the Overview page, select the region where your instance is deployed.
3. In the left-side navigation pane, click Connectors.
4. On the Connectors page, select the instance in which the data source topic resides from the Select Instance drop-down list and click Create Connector.
5. In the Create Connector wizard, perform the following operations:
   1. In the Configure Basic Information step, enter the connector name in the Name field and click Next.

      Parameter	Description	Example
      Name	The name of the connector. Specify a connector name based on the following naming conventions: The connector name must be 1 to 48 characters in length and can contain digits, lowercase letters, and hyphens (-). The name cannot start with a hyphen (-). Each connector name must be unique within a ApsaraMQ for Kafka instance. The connector must use a consumer group that is named in the connect-Connector name format.Group If you have not created this consumer group, Message Queue for Apache Kafka automatically creates the consumer group for you.Group	kafka-source-mysql
      Instance	The information about the Message Queue for Apache Kafka instance. By default, the name and ID of the instance are displayed.	demo alikafka_post-cn-st21p8vj****

   2. In the Configure Source Service step, select ApsaraDB RDS for MySQL as the source service, configure the parameters that are described in the following table, and then click Next.

      Parameter	Description	Example
      Region of ApsaraDB RDS for MySQL Instance	The region ID of the ApsaraDB RDS for MySQL instance. Select a region ID from the drop-down list.	China (Shenzhen)
      ApsaraDB RDS for MySQL Instance ID	The ID of the ApsaraDB RDS for MySQL instance from which you want to synchronize data.	rm-wz91w3vk6owmz****
      Database Name	The name of the ApsaraDB RDS for MySQL database from which you want to synchronize data.	mysql-to-kafka
      Database Account	The username of the account that is used to access the ApsaraDB RDS for MySQL database.	mysql_to_kafka
      Password of Database Account	The password of the account that is used to access the ApsaraDB RDS for MySQL database.	N/A
      Database Table	The name of one or more ApsaraDB RDS for MySQL tables from which you want to synchronize data. Separate multiple table names with commas (,). Source tables and destination topics are related based on one-to-one mappings.	mysql_tbl
      Tables to Automatically Add	The regular expression based on which the system can automatically add other tables in the database. When the system detects that a table is created in the database and matches the regular expression, the system automatically synchronizes the data in the table to Message Queue for Apache Kafka. You can use .* to specify all tables in the database.	.*
      Topic Prefix	The prefix that is used to name the automatically created topics for the connector in ApsaraMQ for Kafka. Each topic name consists of the prefix and the name of the corresponding source table in the ApsaraDB RDS for MySQL database. Make sure that the prefix is globally unique.	mysql

      Important

      Make sure that the ApsaraDB RDS for MySQL database account is granted at least the following permissions:

      • SELECT
      • REPLICATION SLAVE
      • REPLICATION CLIENT

      The following statement provides an example on how to grant permissions to the account:

      GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'Username of an account' @'%'; // Grant the SELECT, REPLICATION SLAVE, and REPLICATION CLIENT permissions to the database account. 

   3. In the Configure Destination Service step, the destination ApsaraMQ for Kafka instance to which you want to synchronize the data appears. Confirm the information and click Create.
6. Go to the Connectors page, find the connector that you created, and then click Deploy in the Actions column.
   On the Connectors page, if the value of the Status parameter is displayed as Running for the connector, the connector is created.

   Note If the connector fails to be created, check whether all the prerequisites that are described in this topic are met.

   If you need to modify the configurations of the MySQL source connector, click Task Configurations in the Actions column. In the DataWorks console to which you are redirected, modify the configurations of the connector.

Verify the result

1. Insert data into a source table in the ApsaraDB RDS for MySQL database.
   The following sample code provides an example on how to insert data into a source table:

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

   For more information about SQL statements, see Commonly used SQL statements for MySQL.
2. Use the message query feature of ApsaraMQ for Kafka to check whether the data of the source table in the ApsaraDB RDS for MySQL database is synchronized to the corresponding topic in your ApsaraMQ for Kafka instance.
   For more information, see Query messages.
   The following sample code provides an example of the data that is synchronized from a source table in ApsaraDB RDS for MySQL to a topic in ApsaraMQ for Kafka. For more information about message structures and fields, see Appendix: Message formats.

   {
       "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"
   }