Kafka Connect 是一款可扩展、可靠的在 Apache Kafka 和其他系统之间进行数据传输的工具，可以定义 Connectors 将大量数据迁入迁出 Kafka。

Doris 提供了 Sink Connector 插件，可以将 Kafka topic 中的数据写入到 Doris 中。

Doris Kafka Connector 使用

Standalone 模式启动

配置 connect-standalone.properties

1. # 修改 broker 地址
2. bootstrap.servers=127.0.0.1:9092

配置 doris-connector-sink.properties 在 config 目录下创建 doris-connector-sink.properties，并配置如下内容：

1. name=test-doris-sink
2. connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
3. topics=topic_test
4. doris.topic2table.map=topic_test:test_kafka_tbl
5. buffer.count.records=10000
6. buffer.flush.time=120
7. buffer.size.bytes=5000000
8. doris.urls=10.10.10.1
9. doris.http.port=8030
10. doris.query.port=9030
11. doris.user=root
12. doris.password=
13. doris.database=test_db
14. key.converter=org.apache.kafka.connect.storage.StringConverter
15. value.converter=org.apache.kafka.connect.json.JsonConverter
16. key.converter.schemas.enable=false
17. value.converter.schemas.enable=false

启动 Standalone

1. $KAFKA_HOME/bin/connect-standalone.sh -daemon $KAFKA_HOME/config/connect-standalone.properties $KAFKA_HOME/config/doris-connector-sink.properties

危险

注意：一般不建议在生产环境中使用 standalone 模式

Distributed 模式启动

配置 connect-distributed.properties

1. # 修改 broker 地址
2. bootstrap.servers=127.0.0.1:9092

4. # 修改 group.id，同一集群的需要一致
5. group.id=connect-cluster

启动 Distributed

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

增加 Connector

1. curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
2. "name":"test-doris-sink-cluster",
3. "config":{
4. "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
5. "topics":"topic_test",
6. "doris.topic2table.map": "topic_test:test_kafka_tbl",
7. "buffer.count.records":"10000",
8. "buffer.flush.time":"120",
9. "buffer.size.bytes":"5000000",
10. "doris.urls":"10.10.10.1",
11. "doris.user":"root",
12. "doris.password":"",
13. "doris.http.port":"8030",
14. "doris.query.port":"9030",
15. "doris.database":"test_db",
16. "key.converter":"org.apache.kafka.connect.storage.StringConverter",
17. "value.converter":"org.apache.kafka.connect.json.JsonConverter",
18. "key.converter.schemas.enable":"false",
19. "value.converter.schemas.enable":"false",
20. }
21. }'

操作 Connector

1. # 查看 connector 状态
2. curl -i http://127.0.0.1:8083/connectors/test-doris-sink-cluster/status -X GET
3. # 删除当前 connector
4. curl -i http://127.0.0.1:8083/connectors/test-doris-sink-cluster -X DELETE
5. # 暂停当前 connector
6. curl -i http://127.0.0.1:8083/connectors/test-doris-sink-cluster/pause -X PUT
7. # 重启当前 connector
8. curl -i http://127.0.0.1:8083/connectors/test-doris-sink-cluster/resume -X PUT
9. # 重启 connector 内的 tasks
10. curl -i http://127.0.0.1:8083/connectors/test-doris-sink-cluster/tasks/0/restart -X POST

参考：Connect REST Interface

危险

注意 kafka-connect 首次启动时，会往 kafka 集群中创建 config.storage.topic offset.storage.topic status.storage.topic 三个 topic 用于记录 kafka-connect 的共享连接器配置、偏移数据和状态更新。How to Use Kafka Connect - Get Started

访问 SSL 认证的 Kafka 集群

通过 kafka-connect 访问 SSL 认证的 Kafka 集群需要用户提供用于认证 Kafka Broker 公钥的证书文件（client.truststore.jks）。您可以在 connect-distributed.properties 文件中增加以下配置：

1. # Connect worker
2. security.protocol=SSL
3. ssl.truststore.location=/var/ssl/private/client.truststore.jks
4. ssl.truststore.password=test1234

6. # Embedded consumer for sink connectors
7. consumer.security.protocol=SSL
8. consumer.ssl.truststore.location=/var/ssl/private/client.truststore.jks
9. consumer.ssl.truststore.password=test1234

关于通过 kafka-connect 连接 SSL 认证的 Kafka 集群配置说明可以参考：Configure Kafka Connect

死信队列

默认情况下，转换过程中或转换过程中遇到的任何错误都会导致连接器失败。每个连接器配置还可以通过跳过它们来容忍此类错误，可选择将每个错误和失败操作的详细信息以及有问题的记录（具有不同级别的详细信息）写入死信队列以便记录。

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

配置项

其他Kafka Connect Sink通用配置项可参考：Kafka Connect Sink Configuration Properties