ApsaraDB for SelectDB:Import data by using Doris Kafka Connector

Background information

Kafka Connect is a reliable tool for transferring data between Apache Kafka and other systems. You can define connectors to import data to or export data from a Kafka data source.

Doris Kafka Connector provided by Apache Doris runs in a Kafka Connect cluster. Doris Kafka Connector can read data from Kafka topics and write the data to ApsaraDB for SelectDB.

In business scenarios, you may push the updated data in databases to Kafka by using Debezium Connector or call API operations to write JSON-formatted data to Kafka in real time. Then, you can use Doris Kafka Connector to automatically subscribe to data in Kafka and synchronize the data to an ApsaraDB for SelectDB instance.

Operation modes of Kafka Connect

Kafka Connect provides standalone mode and distributed mode. You can use an operation mode based on your business requirements.

Standalone mode

Configure the standalone mode

Configure the connect-standalone.properties file.

# Modify the IP address of the broker.
bootstrap.servers=127.0.0.1:9092

Create the connect-selectdb-sink.properties file in the Kafka config directory and configure the following items in the file:

name=test-selectdb-sink
connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
topics=topic_test
doris.topic2table.map=topic_test:test_kafka_tbl
buffer.count.records=10000
buffer.flush.time=120
buffer.size.bytes=5000000
doris.urls=selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com
doris.http.port=8030
doris.query.port=9030
doris.user=admin
doris.password=****
doris.database=test_db
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

Enable the standalone mode

$KAFKA_HOME/bin/connect-standalone.sh -daemon $KAFKA_HOME/config/connect-standalone.properties $KAFKA_HOME/config/connect-selectdb-sink.properties

Distributed mode

Configure the distributed mode

Configure the connect-distributed.properties file.

# Modify the IP address of the broker.
bootstrap.servers=127.0.0.1:9092
 
# Modify the group ID. The group ID must be consistent in the same cluster.
group.id=connect-cluster

Enable the distributed mode

$KAFKA_HOME/bin/connect-distributed.sh -daemon $KAFKA_HOME/config/connect-distributed.properties

Configure a connector

curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
  "name":"test-selectdb-sink-cluster",
  "config":{
    "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
    "topics":"topic_test",
    "doris.topic2table.map": "topic_test:test_kafka_tbl",
    "buffer.count.records":"10000",
    "buffer.flush.time":"120",
    "buffer.size.bytes":"5000000",
    "doris.urls":"selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com",
    "doris.user":"admin",
    "doris.password":"***",
    "doris.database":"test_db",
    "doris.http.port":"8080",
    "doris.query.port":"9030",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "value.converter":"org.apache.kafka.connect.json.JsonConverter"
  }
}'

Parameters

Examples

Prepare the environment

1. Install an Apache Kafka cluster whose version is 2.4.0 or later or Confluent Cloud. In this example, a standalone Kafka cluster is used.

   # Download and decompress the package. 
   wget https://archive.apache.org/dist/kafka/2.4.0/kafka_2.12-2.4.0.tgz
   tar -zxvf kafka_2.12-2.4.0.tgz
   bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
   bin/kafka-server-start.sh -daemon config/server.properties

2. Download the doris-kafka-connector-1.0.0.jar package and store the JAR package in the KAKFA_HOME/libs directory.

3. Create an ApsaraDB for SelectDB instance. For more information, see Create an instance.

4. Connect to the ApsaraDB for SelectDB instance over the MySQL protocol. For more information, see Connect to an instance.

5. Create a test database and a test table.

   1. Execute the following statement to create a test database:

      CREATE DATABASE test_db;

   2. Execute the following statements to create a test table:

      USE test_db;
      CREATE TABLE employees (
          emp_no       int NOT NULL,
          birth_date   date,
          first_name   varchar(20),
          last_name    varchar(20),
          gender       char(2),
          hire_date    date
      )
      UNIQUE KEY(`emp_no`)
      DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1;

Example 1: Synchronize JSON-formatted data

1. Configure an ApsaraDB for SelectDB Sink.

   In this example, the standalone mode is used. Create the selectdb-sink.properties file in the Kafka config directory and configure the following items in the file:

   name=selectdb_sink
   connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
   topics=test_topic
   doris.topic2table.map=test_topic:example_tbl
   buffer.count.records=10000
   buffer.flush.time=120
   buffer.size.bytes=5000000
   doris.urls=selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com
   doris.http.port=8080
   doris.query.port=9030
   doris.user=admin
   doris.password=***
   doris.database=test_db
   key.converter=org.apache.kafka.connect.storage.StringConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   
   # Optional. Configure a dead-letter queue.
   errors.tolerance=all
   errors.deadletterqueue.topic.name=test_error
   errors.deadletterqueue.context.headers.enable = true
   errors.deadletterqueue.topic.replication.factor=1

2. Start Kafka Connect.

   bin/connect-standalone.sh -daemon config/connect-standalone.properties config/selectdb-sink.properties

Example 2: Use Debezium Connector to synchronize data from a MySQL database to an ApsaraDB for SelectDB instance

In some business scenarios, you need to synchronize data from business databases in real time. In this case, the change data capture (CDC) mechanism is required.

Debezium Connector is a CDC tool that is developed based on Kafka Connect. Debezium Connector can connect to various databases, such as MySQL, PostgreSQL, SQL Server, Oracle, and MongoDB, and continuously send data from the databases to Kafka topics in a unified format for downstream sinks to consume in real time. In this example, a MySQL data source is used.

1. Download the Debezium Connector package.

   wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.9.8.Final/debezium-connector-mysql-1.9.8.Final-plugin.tar.gz

2. Decompress the downloaded package.

   tar -zxvf debezium-connector-mysql-1.9.8.Final-plugin.tar.gz

3. Store all the extracted JAR packages in the KAKFA_HOME/libs directory.

4. Configure the MySQL data source.

   Create the mysql-source.properties file in the Kafka config directory and configure the following items in the file:

   name=mysql-source
   connector.class=io.debezium.connector.mysql.MySqlConnector
   database.hostname=rm-bp17372257wkz****.rwlb.rds.aliyuncs.com
   database.port=3306
   database.user=testuser
   database.password=****
   database.server.id=1
   # The unique identifier of the client in Kafka.
   database.server.name=test123
   # The database and table from which data is synchronized. By default, data is synchronized from all databases and tables.
   database.include.list=test
   table.include.list=test.test_table
   database.history.kafka.bootstrap.servers=localhost:9092
   # The Kafka topic that is used to store the schema change events of the database and table.
   database.history.kafka.topic=dbhistory
   transforms=unwrap
   # For more information, visit https://debezium.io/documentation/reference/stable/transformations/event-flattening.html.
   transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
   # Record change events for DELETE operations.
   transforms.unwrap.delete.handling.mode=rewrite

   After the configuration is complete, the Kafka topic is named in the SERVER_NAME.DATABASE_NAME.TABLE_NAME format by default.

   Note

   For more information about how to configure Debezium Connector, see Debezium connector for MySQL.

5. Configure an ApsaraDB for SelectDB Sink.

   Create the selectdb-sink.properties file in the Kafka config directory and configure the following items in the file:

   name=selectdb-sink
   connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
   topics=test123.test.test_table
   doris.topic2table.map=test123.test.test_table:test_table
   buffer.count.records=10000
   buffer.flush.time=120
   buffer.size.bytes=5000000
   doris.urls=selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com
   doris.http.port=8080
   doris.query.port=9030
   doris.user=admin
   doris.password=****
   doris.database=test
   key.converter=org.apache.kafka.connect.json.JsonConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   
   # Optional. Configure a dead-letter queue.
   #errors.tolerance=all
   #errors.deadletterqueue.topic.name=test_error
   #errors.deadletterqueue.context.headers.enable = true
   #errors.deadletterqueue.topic.replication.factor=1

   Note

   Before you synchronize data to an ApsaraDB for SelectDB instance, you must create a database and a table in the instance.

6. Start Kafka Connect.

   bin/connect-standalone.sh -daemon config/connect-standalone.properties config/mysql-source.properties config/selectdb-sink.properties

   Note

   After you start Kafka Connect, you can view the logs/connect.log file to check whether Kafka Connect is started.

Advanced usage

Manage connectors

# Query the status of the connector.
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/status -X GET
# Delete the current connector.
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster -X DELETE
# Pause the current connector.
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/pause -X PUT
# Restart the current connector.
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/resume -X PUT
# Restart the tasks in the connector.
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/tasks/0/restart -X POST

For more information, see Kafka Connect REST Interface for Confluent Platform.

Configure a dead-letter queue

By default, the conversion process or the errors that occur during the conversion process may cause the connector to fail. You can also tolerate such errors by configuring the connector to skip the errors. You can write the details of errors, failed operations, and abnormal data records with various levels of details to a dead-letter queue.

errors.tolerance=all
errors.deadletterqueue.topic.name=test_error_topic
errors.deadletterqueue.context.headers.enable=true
errors.deadletterqueue.topic.replication.factor=1

For more information, see Error Reporting in Connect.

Access a Kafka cluster with SSL authentication

To access a Kafka cluster with SSL authentication by using Kafka Connect, you must provide the certificate file client.truststore.jks that is used to authenticate the public key of a Kafka broker. You can add the following configurations to the connect-distributed.properties file:

# Connect worker
security.protocol=SSL
ssl.truststore.location=/var/ssl/private/client.truststore.jks
ssl.truststore.password=test1234
 
# Embedded consumer for sink connectors
consumer.security.protocol=SSL
consumer.ssl.truststore.location=/var/ssl/private/client.truststore.jks
consumer.ssl.truststore.password=test1234

For more information about the configurations for accessing a Kafka cluster with SSL authentication by using Kafka Connect, see Configure Kafka Connect.