All Products
Search

This Product

    ApsaraMQ for Kafka:Use Canal to synchronize MySQL to ApsaraMQ for Kafka

    Document Center

    ApsaraMQ for Kafka:Use Canal to synchronize MySQL to ApsaraMQ for Kafka

    Last Updated:Jul 10, 2026

    Canal captures incremental data changes from MySQL binary logs and writes them to ApsaraMQ for Kafka for downstream analysis.

    Background

    Canal parses incremental logs from a MySQL database, enabling you to subscribe to and consume incremental data changes. It acts as a MySQL slave and sends a dump request to the MySQL master. The master then streams its binary log to Canal, which parses the log to synchronize data. By integrating Canal with ApsaraMQ for Kafka, you can write data changes from MySQL into ApsaraMQ for Kafka for analysis. For details, see the official Canal website.背景介绍

    Prerequisites

    Before you start, make sure the following prerequisites are met:

    Procedure

    1. Download the Canal package. In the example of this topic, V1.1.5 is used.

    2. Run the following command to create a directory. In the example, the /home/doc/tools/canal.deployer-1.1.5 directory is created. 

      mkdir -p /home/doc/tools/canal.deployer-1.1.5

    3. Copy the Canal package to the /home/doc/tools/canal.deployer-1.1.5 directory and decompress the package. 

      tar -zxvf canal.deployer-1.1.5-SNAPSHOT.tar.gz -C /home/doc/tools/canal.deployer-1.1.5

    4. In the /home/doc/tools/canal.deployer-1.1.5 directory, run the following command to modify the instance.properties file: 

      vi conf/example/instance.properties

      Configure the parameters as described in the instance.properties parameter list.

      # Modify the following parameters based on your database information.
#################################################
...
# The database address.
canal.instance.master.address=192.168.XX.XX:3306
# The username and password for the database.
...
canal.instance.dbUsername=****
canal.instance.dbPassword=****
...
# mq config
# The topic you created in the ApsaraMQ for Kafka console.
canal.mq.topic=mysql_test
# Route data to dynamic topics based on the database or table name.
#canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\\..*,.*\\..*
# Synchronize data to the specified partition of an ApsaraMQ for Kafka topic.
canal.mq.partition=0
# The following two parameters are mutually exclusive with canal.mq.partition. Use them to send data to different partitions within a topic.
#canal.mq.partitionsNum=3
# Format: database.table:unique_primary_key. Separate multiple tables with commas.
#canal.mq.partitionHash=mytest.person:id,mytest.role:id
#################################################
      Table 1. instance.properties parameters

      Parameter

      Required

      Description

      canal.instance.master.address

      Yes

      The connection address of your MySQL database.

      canal.instance.dbUsername

      Yes

      The username for your MySQL database.

      canal.instance.dbPassword

      Yes

      The password for your MySQL database.

      canal.mq.topic

      Yes

      The topic in your ApsaraMQ for Kafka instance. You can create a topic on the Topics page of the ApsaraMQ for Kafka console. For more information, see Step 3: Create resources.

      canal.mq.dynamicTopic

      No

      A regular expression for dynamic topic routing. This allows you to synchronize data from different database tables to different topics. For configuration details, see Parameter description.

      canal.mq.partition

      No

      The specific partition of the ApsaraMQ for Kafka topic to which you want to synchronize data.

      canal.mq.partitionsNum

      No

      The number of partitions in the topic. Use this parameter with canal.mq.partitionHash to synchronize data to different partitions of an ApsaraMQ for Kafka topic.

      canal.mq.partitionHash

      No

      A regular expression for the partitioning rule. For configuration details, see Parameter description.

    5. Run the following command to open the canal.properties file: 

      vi conf/canal.properties

      Configure the parameters as described in the canal.properties parameter list.

      • If you connect from the public network, use an SSL endpoint and the SASL_SSL protocol for authentication and encryption to access ApsaraMQ for Kafka. For more information about endpoints, see Endpoint comparison.

        # ...
# Set the server mode to kafka.
canal.serverMode = kafka
# ...
# Kafka configurations.
# The SSL endpoint obtained from the Instance Details page in the ApsaraMQ for Kafka console.
kafka.bootstrap.servers = alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093
# The following are default settings. You can adjust them as needed.
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0
# For public network access, configure the following network protocol and authentication mechanism for SASL_SSL.
kafka.ssl.truststore.location= ../conf/kafka_client_truststore_jks
kafka.ssl.truststore.password= KafkaOnsClient
kafka.security.protocol= SASL_SSL
kafka.sasl.mechanism = PLAIN
kafka.ssl.endpoint.identification.algorithm =
        Table 2. canal.properties parameters

        Parameter

        Required

        Description

        canal.serverMode

        Yes

        Set this value to kafka.

        kafka.bootstrap.servers

        Yes

        The endpoint of your ApsaraMQ for Kafka instance. You can obtain the endpoint from the Endpoint Information section on the Instance Details page in the ApsaraMQ for Kafka console.

        kafka.ssl.truststore.location

        Yes

        The path to the SSL root certificate, kafka.client.truststore.jks.

        Note

        When connecting from the public network, you must use authentication and encryption to ensure data security. This requires using an SSL endpoint with the SASL_SSL protocol. For more information, see Endpoint comparison.

        kafka.acks

        Yes

        The acknowledgment that ApsaraMQ for Kafka sends to the client after it receives data. The values are described as follows:

        • 0: The producer does not wait for any acknowledgment.

        • 1: The producer waits for an acknowledgment from the leader only.

        • all: The producer waits for an acknowledgment from the leader and all in-sync replicas.

        kafka.compression.type

        Yes

        The data compression algorithm. Defaults to none. Valid values:

        • none

        • gzip

        • snappy

        kafka.batch.size

        Yes

        The maximum size of a message batch, in bytes.

        This parameter specifies the maximum number of bytes that can be sent in a batch. The client groups messages into batches before sending them to brokers, which reduces the number of requests. A small batch size may reduce throughput, while a large batch size uses more memory.

        Note

        The client sends a batch when either the kafka.batch.size or the kafka.linger.ms condition is met.

        kafka.linger.ms

        Yes

        The maximum time the client waits to fill a batch. Unit: ms.

        The client sends a batch to the broker after this wait time elapses, which helps reduce the total number of requests.

        kafka.max.request.size

        Yes

        The maximum request size, in bytes.

        kafka.buffer.memory

        Yes

        The total memory size available to the producer for buffering.

        kafka.max.in.flight.requests.per.connection

        Yes

        The maximum number of unacknowledged requests the client can send on a single connection. If this is set to 1, the client cannot send new requests to the same broker until the previous request is acknowledged.

        kafka.retries

        Yes

        The number of times to retry sending a message after a failure. A value of 0 disables retries.

        kafka.ssl.truststore.password

        Yes

        The password for the SSL root certificate. Set this to KafkaOnsClient.

        kafka.security.protocol

        Yes

        The protocol used for authentication and encryption. Set this to SASL_SSL.

        kafka.sasl.mechanism

        Yes

        The SASL mechanism for authentication. An SSL endpoint uses the PLAIN mechanism.

        For public network access with SASL authentication, you must configure an environment variable in bin/startup.sh and edit the kafka_client_producer_jaas.conf file to provide the username and password for your ApsaraMQ for Kafka instance.

        1. Run the vi bin/startup.sh command to edit the startup.sh file and configure the environment variable.

          JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8 -Djava.security.auth.login.config=/home/doc/tools/canal.deployer-1.1.5/conf/kafka_client_jaas.conf"

        2. Run the vi conf/kafka_client_producer_jaas.conf command to edit the kafka_client_producer_jaas.conf file and configure the instance username and password.

          Note

          • If the Access Control List (ACL) feature is disabled for the instance, you can find the default username and password on the Instance Details page in the ApsaraMQ for Kafka console.

          • If the Access Control List (ACL) is enabled, make sure the SASL user is of the PLAIN type and is granted permissions to send and receive messages. For more information, see Grant permissions to a SASL user.

          KafkaClient {  org.apache.kafka.common.security.plain.PlainLoginModule required
               username="your_instance_username"  
               password="your_instance_password";
};

      • In a VPC environment, messages are transmitted by using the PLAINTEXT protocol without authentication or encryption. To access ApsaraMQ for Kafka through the default endpoint, you only need to configure the canal.serverMode and kafka.bootstrap.servers parameters. For more information about endpoints, see Endpoint comparison.

        # ...
# Set the server mode to kafka.
canal.serverMode = kafka
# ...
# Kafka configurations.
# The default endpoint obtained from the Instance Details page in the ApsaraMQ for Kafka console.
kafka.bootstrap.servers = alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092
# The following are default settings. You can adjust them as needed or keep the defaults.
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

    6. In the /home/doc/tools/canal.deployer-1.1.5 directory, run the following command to start Canal: 

      sh bin/startup.sh

      • Check the /home/doc/tools/canal.deployer-1.1.5/logs/canal/canal.log file to confirm that Canal is connected to ApsaraMQ for Kafka and is running.

        2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.XX.XX:11111]
2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

      • Check the /home/doc/tools/canal.deployer-1.1.5/logs/example/example.log file to confirm that the Canal instance has started.

        2013-02-05 22:50:45.636 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....

    Test the configurations

    After Canal starts, test data synchronization to verify the setup.

    1. In your MySQL database, create a table named T_Student. The following example shows sample data in the table:

      mysql> select * from T_Student;
+--------+---------+------+------+
| stuNum | stuName | age  | sex  |
+--------+---------+------+------+
|      1 | Wang    |   18 | girl |
|      2 | Zhang   |   17 | boy  |
+--------+---------+------+------+
2 rows in set (0.00 sec)

      View the /home/doc/tools/canal.deployer-1.1.5/logs/example/meta.log log file. Each create, update, and delete operation on the database generates a record in meta.log, which you can use to confirm that Canal is capturing data.

      tail -f example/meta.log
2020-07-29 09:21:05.110 - clientId:1001 cursor:[log.000001,29723,1591190230000,1,] address[/192.168.XX.XX:3306]
2020-07-29 09:23:46.109 - clientId:1001 cursor:[log.000001,30047,1595985825000,1,] address[localhost/192.168.XX.XX:3306]
2020-07-29 09:24:50.547 - clientId:1001 cursor:[log.000001,30047,1595985825000,1,] address[/192.168.XX.XX:3306]
2020-07-29 09:26:45.547 - clientId:1001 cursor:[log.000001,30143,1595986005000,1,] address[localhost/192.168.XX.XX:3306]
2020-07-29 09:30:04.546 - clientId:1001 cursor:[log.000001,30467,1595986204000,1,] address[localhost/192.168.XX.XX:3306]
2020-07-29 09:30:16.546 - clientId:1001 cursor:[log.000001,30734,1595986215000,1,] address[localhost/192.168.XX.XX:3306]
2020-07-29 09:30:36.547 - clientId:1001 cursor:[log.000001,31001,1595986236000,1,] address[localhost/192.168.XX.XX:3306]

    2. Log on to the ApsaraMQ for Kafka console and query for messages to confirm that the data from MySQL is synchronized to ApsaraMQ for Kafka. For more information about how to query messages in the console, see Message Query.

      The query results show two message records in the mysql_test topic. Each message is a JSON object containing fields such as stuNum, stuName, age, sex, and database, confirming that the MySQL data was synchronized to ApsaraMQ for Kafka.

    3. After data is synchronized, run the following command to stop Canal: 

      sh bin/stop.sh