Import data using Kafka

Standalone mode

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

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

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

Distributed mode

# Modify the broker address
bootstrap.servers=127.0.0.1:9092
 
# Modify group.id. The ID must be the same for all workers in the same cluster.
group.id=connect-cluster

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

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":"8030",
    "doris.query.port":"9030",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "value.converter":"org.apache.kafka.connect.json.JsonConverter"
  }
}'

Parameters

Prerequisites

1. Install an Apache Kafka cluster or Confluent Cloud, version 2.4.0 or later. This example uses a single-node Kafka environment.

   # 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
   cd kafka_2.12-2.4.0/
   bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
   bin/kafka-server-start.sh -daemon config/server.properties

2. Download doris-kafka-connector-1.0.0.jar and place the JAR file in the KAFKA_HOME/libs directory.

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

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

5. Create a test database and a test table.

   1. Create a test database.

      CREATE DATABASE test_db;

   2. 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: Sync JSON data

1. Configure the SelectDB sink

   Using standalone mode as an example, create a selectdb-sink.properties file in the Kafka config directory and add the following content:

   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=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
   
   #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 to synchronize MySQL data to ApsaraDB for SelectDB

In many business scenarios, you need to sync data from an operational database in real time. This requires using the database's change data capture (CDC) mechanism.

Debezium is a CDC tool based on Kafka Connect that can connect to various databases such as MySQL, PostgreSQL, SQL Server, Oracle, and MongoDB. It continuously sends data changes to a Kafka topic in a unified format for real-time consumption by downstream sinks. This example uses MySQL.

1. Download Debezium.

   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 file.

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

3. Place all the extracted JAR files into the KAFKA_HOME/libs directory.

4. Configure the MySQL source.

   Create a mysql-source.properties file in the Kafka config directory and add the following content:

   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
   # A unique identifier for this client in Kafka
   database.server.name=test123
   # The databases and tables to sync. By default, all databases and tables are synced.
   database.include.list=test
   table.include.list=test.test_table
   database.history.kafka.bootstrap.servers=localhost:9092
   # The Kafka topic used to store database schema changes
   database.history.kafka.topic=dbhistory
   transforms=unwrap
   # See https://debezium.io/documentation/reference/stable/transformations/event-flattening.html
   transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
   # Record delete events
   transforms.unwrap.delete.handling.mode=rewrite

   After configuration, the default Kafka topic name format is SERVER_NAME.DATABASE_NAME.TABLE_NAME.

   Note

   For Debezium configurations, see Debezium connector for MySQL.

5. Configure ApsaraDB for SelectDB sink.

   Create a selectdb-sink.properties file in the Kafka config directory and add the following content:

   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=8030
   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

   When you synchronize data to ApsaraDB for SelectDB, you must create a database and a table in advance.

6. Start Kafka Connect.

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

   Note

   After startup, you can check the logs/connect.log file to verify that the service started successfully.

Connector operations

# Check the connector status
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
# Resume the current connector
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/resume -X PUT
# Restart tasks within the connector
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/tasks/0/restart -X POST

For more information, see Connect REST Interface.

Dead-letter queue

By default, a conversion error causes the connector to fail. However, you can configure the connector to tolerate such errors by skipping them. You can also write the error details, the failed operation, and the problematic record to a dead-letter queue for later analysis.

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.

Connect to an SSL-enabled Kafka cluster

To access an SSL-enabled Kafka cluster via Kafka Connect, you need to provide a certificate file (client.truststore.jks) to authenticate the public key of the Kafka broker. You can add the following configuration to your 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 configuring Kafka Connect to connect to an SSL-enabled Kafka cluster, see Configure Kafka Connect.