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

Doris 社区提供了 doris-kafka-connector 插件,可以将 Kafka topic 中的数据写入到 Doris 中。

Doris Kafka Connector 使用

下载

doris-kafka-connector

maven 依赖

  1. <dependency>
  2. <groupId>org.apache.doris</groupId>
  3. <artifactId>doris-kafka-connector</artifactId>
  4. <version>1.0.0</version>
  5. </dependency>

Standalone 模式启动

在 $KAFKA_HOME 下创建 plugins 目录,将下载好的 doris-kafka-connector jar 包放入其中

配置 config/connect-standalone.properties

  1. # 修改 broker 地址
  2. bootstrap.servers=127.0.0.1:9092
  3. # 修改为创建的 plugins 目录
  4. # 注意:此处请填写 Kafka 的直接路径。例如:plugin.path=/opt/kafka/plugins
  5. plugin.path=$KAFKA_HOME/plugins

配置 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

启动 Standalone

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

Doris Kafka Connector - 图1备注

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

Distributed 模式启动

在 $KAFKA_HOME 下创建 plugins 目录,将下载好的 doris-kafka-connector jar 包放入其中

配置 config/connect-distributed.properties

  1. # 修改 broker 地址
  2. bootstrap.servers=127.0.0.1:9092
  3. # 修改 group.id,同一集群的需要一致
  4. group.id=connect-cluster
  5. # 修改为创建的 plugins 目录
  6. # 注意:此处请填写 Kafka 的直接路径。例如:plugin.path=/opt/kafka/plugins
  7. plugin.path=$KAFKA_HOME/plugins

启动 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. }
  19. }'

操作 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

Doris Kafka Connector - 图2备注

注意 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
  5. # Embedded consumer for sink connectors
  6. consumer.security.protocol=SSL
  7. consumer.ssl.truststore.location=/var/ssl/private/client.truststore.jks
  8. 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

配置项

KeyEnumDefault ValueRequiredDescription
name--YConnect 应用名称,必须是在 Kafka Connect 环境中唯一
connector.class--Yorg.apache.doris.kafka.connector.DorisSinkConnector
topics--Y订阅的 topic 列表,逗号分隔:topic1,topic2
doris.urls--YDoris FE 连接地址。如果有多个,中间用逗号分割:10.20.30.1,10.20.30.2,10.20.30.3
doris.http.port--YDoris HTTP 协议端口
doris.query.port--YDoris MySQL 协议端口
doris.user--YDoris 用户名
doris.password--YDoris 密码
doris.database--Y要写入的数据库。多个库时可以为空,同时在 topic2table.map 需要配置具体的库名称
doris.topic2table.map--Ntopic 和 table 表的对应关系,例:topic1:tb1,topic2:tb2
默认为空,表示 topic 和 table 名称一一对应。
多个库的格式为 topic1:db1.tbl1,topic2:db2.tbl2
buffer.count.records-10000N在 flush 到 doris 之前,每个 Kafka 分区在内存中缓冲的记录数。默认 10000 条记录
buffer.flush.time-120Nbuffer 刷新间隔,单位秒,默认 120 秒
buffer.size.bytes-5000000(5MB)N每个 Kafka 分区在内存中缓冲的记录的累积大小,单位字节,默认 5MB
jmx-trueN通过 JMX 获取 Connector 内部监控指标,请参考:Doris-Connector-JMX
enable.2pc-trueN是否开启 Stream Load 的两阶段提交 (TwoPhaseCommit),默认为 true。
enable.delete-falseN是否同步删除记录,默认 false
label.prefix-${name}NStream load 导入数据时的 label 前缀。默认为 Connector 应用名称。
auto.redirect-trueN是否重定向 StreamLoad 请求。开启后 StreamLoad 将通过 FE 重定向到需要写入数据的 BE,并且不再显示获取 BE 信息
load.modelstream_load,
copy_into
stream_loadN导入数据的方式。支持 stream_load 直接数据导入到 Doris 中;同时支持 copy_into 的方式导入数据至对象存储中,然后将数据加载至 Doris 中
sink.properties.*-‘sink.properties.format’:’json’,
‘sink.properties.read_json_by_line’:’true’
NStream Load 的导入参数。
例如:定义列分隔符‘sink.properties.column_separator’:’,’
详细参数参考这里

开启 Group Commit,例如开启 sync_mode 模式的 group commit:“sink.properties.group_commit”:”sync_mode”。 Group Commit 可以配置 off_modesync_modeasync_mode 三种模式 ,具体使用参考:Group-Commit

开启部分列更新,例如开启更新指定 col2 的部分列:“sink.properties.partial_columns”:”true”, “sink.properties.columns”: “col2”,
delivery.guaranteeat_least_once,
exactly_once
at_least_onceN消费 Kafka 数据导入至 doris 时,数据一致性的保障方式。支持 at_least_once exactly_once,默认为 at_least_once 。Doris 需要升级至 2.1.0 以上,才能保障数据的 exactly_once
converter.modenormal,
debezium_ingestion
normalN使用 Connector 消费 Kafka 数据时,上游数据的类型转换模式。
normal表示正常消费 Kafka 中的数据,不经过任何类型转换。
debezium_ingestion表示当 Kafka 上游的数据通过 Debezium 等 CDC (Changelog Data Capture,变更数据捕获)工具采集时,上游数据需要经过特殊的类型转换才能支持。
debezium.schema.evolutionnone,
basic
noneN通过 Debezium 采集上游数据库系统(如 MySQL),发生结构变更时,可以将增加的字段同步到 Doris 中。
none表示上游数据库系统发生结构变更时,不同步变更后的结构到 Doris 中。
basic表示同步上游数据库的数据变更操作。由于列结构变更是一个危险操作(可能会导致误删 Doris 表结构的列),目前仅支持同步上游增加列的操作。当列被重命名后,则旧列保持原样,Connector 会在目标表中新增一列,将重命名后的新增数据 Sink 到新列中。
database.time_zone-UTCNconverter.mode 为非 normal 模式时,对于日期数据类型(如 datetime, date, timestamp 等等)提供指定时区转换的方式,默认为 UTC 时区。
avro.topic2schema.filepath--N通过读取本地提供的 Avro Schema 文件,来解析 Topic 中的 Avro 文件内容,实现与 Confluent 提供 Schema 注册中心解耦。
此配置需要与 key.convertervalue.converter 前缀一起使用,例如配置 avro-user、avro-product Topic 的本地 Avro Schema 文件如下: “value.converter.avro.topic2schema.filepath”:”avro-user:file:///opt/avro_user.avsc, avro-product:file:///opt/avro_product.avsc”
具体使用可以参考:#32

其他Kafka Connect Sink通用配置项可参考:connect_configuring

类型映射

Doris-kafka-connector 使用逻辑或原始类型映射来解析列的数据类型。

原始类型是指使用 Kafka connect 的 `Schema` 表示的简单数据类型。逻辑数据类型通常是采用 `Struct` 结构表示复杂类型,或者日期时间类型。

Kafka 原始类型Doris 类型
INT8TINYINT
INT16SMALLINT
INT32INT
INT64BIGINT
FLOAT32FLOAT
FLOAT64DOUBLE
BOOLEANBOOLEAN
STRINGSTRING
BYTESSTRING
Kafka 逻辑类型Doris 类型
org.apache.kafka.connect.data.DecimalDECIMAL
org.apache.kafka.connect.data.DateDATE
org.apache.kafka.connect.data.TimeSTRING
org.apache.kafka.connect.data.TimestampDATETIME
Debezium 逻辑类型Doris 类型
io.debezium.time.DateDATE
io.debezium.time.TimeString
io.debezium.time.MicroTimeDATETIME
io.debezium.time.NanoTimeDATETIME
io.debezium.time.ZonedTimeDATETIME
io.debezium.time.TimestampDATETIME
io.debezium.time.MicroTimestampDATETIME
io.debezium.time.NanoTimestampDATETIME
io.debezium.time.ZonedTimestampDATETIME
io.debezium.data.VariableScaleDecimalDOUBLE

最佳实践

同步 JSON 序列化数据

  1. curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
  2. "name":"doris-json-test",
  3. "config":{
  4. "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
  5. "topics":"json_topic",
  6. "tasks.max":"10",
  7. "doris.topic2table.map": "json_topic:json_tab",
  8. "buffer.count.records":"100000",
  9. "buffer.flush.time":"120",
  10. "buffer.size.bytes":"10000000",
  11. "doris.urls":"127.0.0.1",
  12. "doris.user":"root",
  13. "doris.password":"",
  14. "doris.http.port":"8030",
  15. "doris.query.port":"9030",
  16. "doris.database":"test",
  17. "load.model":"stream_load",
  18. "key.converter":"org.apache.kafka.connect.json.JsonConverter",
  19. "value.converter":"org.apache.kafka.connect.json.JsonConverter"
  20. }
  21. }'

同步 Avro 序列化数据

  1. curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
  2. "name":"doris-avro-test",
  3. "config":{
  4. "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
  5. "topics":"avro_topic",
  6. "tasks.max":"10",
  7. "doris.topic2table.map": "avro_topic:avro_tab",
  8. "buffer.count.records":"100000",
  9. "buffer.flush.time":"120",
  10. "buffer.size.bytes":"10000000",
  11. "doris.urls":"127.0.0.1",
  12. "doris.user":"root",
  13. "doris.password":"",
  14. "doris.http.port":"8030",
  15. "doris.query.port":"9030",
  16. "doris.database":"test",
  17. "load.model":"stream_load",
  18. "key.converter":"io.confluent.connect.avro.AvroConverter",
  19. "key.converter.schema.registry.url":"http://127.0.0.1:8081",
  20. "value.converter":"io.confluent.connect.avro.AvroConverter",
  21. "value.converter.schema.registry.url":"http://127.0.0.1:8081"
  22. }
  23. }'

同步 Protobuf 序列化数据

  1. curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
  2. "name":"doris-protobuf-test",
  3. "config":{
  4. "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
  5. "topics":"proto_topic",
  6. "tasks.max":"10",
  7. "doris.topic2table.map": "proto_topic:proto_tab",
  8. "buffer.count.records":"100000",
  9. "buffer.flush.time":"120",
  10. "buffer.size.bytes":"10000000",
  11. "doris.urls":"127.0.0.1",
  12. "doris.user":"root",
  13. "doris.password":"",
  14. "doris.http.port":"8030",
  15. "doris.query.port":"9030",
  16. "doris.database":"test",
  17. "load.model":"stream_load",
  18. "key.converter":"io.confluent.connect.protobuf.ProtobufConverter",
  19. "key.converter.schema.registry.url":"http://127.0.0.1:8081",
  20. "value.converter":"io.confluent.connect.protobuf.ProtobufConverter",
  21. "value.converter.schema.registry.url":"http://127.0.0.1:8081"
  22. }
  23. }'

常见问题

1. 读取 JSON 类型的数据报如下错误:

  1. Caused by: org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.enable=false in your converter configuration.
  2. at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:337)
  3. at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:91)
  4. at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$4(WorkerSinkTask.java:536)
  5. at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:180)
  6. at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:214)

原因: 是因为使用 org.apache.kafka.connect.json.JsonConverter 转换器需要匹配 “schema” 和 “payload” 字段。

两种解决方案,任选其一:

  1. org.apache.kafka.connect.json.JsonConverter 更换为 org.apache.kafka.connect.storage.StringConverter
  2. 启动模式为 Standalone 模式,则将 config/connect-standalone.properties 中 value.converter.schemas.enablekey.converter.schemas.enable 改成false; 启动模式为 Distributed 模式,则将 config/connect-distributed.properties 中 value.converter.schemas.enablekey.converter.schemas.enable 改成false