TDengine Kafka Connector 使用教程

TDengine Kafka Connector 包含两个插件: TDengine Source Connector 和 TDengine Sink Connector。用户只需提供简单的配置文件，就可以将 Kafka 中指定 topic 的数据（批量或实时）同步到 TDengine，或将 TDengine 中指定数据库的数据（批量或实时）同步到 Kafka。

什么是 Kafka Connect？

Kafka Connect 是 Apache Kafka 的一个组件，用于使其它系统，比如数据库、云服务、文件系统等能方便地连接到 Kafka。数据既可以通过 Kafka Connect 从其它系统流向 Kafka, 也可以通过 Kafka Connect 从 Kafka 流向其它系统。从其它系统读数据的插件称为 Source Connector, 写数据到其它系统的插件称为 Sink Connector。Source Connector 和 Sink Connector 都不会直接连接 Kafka Broker，Source Connector 把数据转交给 Kafka Connect。Sink Connector 从 Kafka Connect 接收数据。

TDengine Database Kafka Connector -- Kafka Connect structure

TDengine Source Connector 用于把数据实时地从 TDengine 读出来发送给 Kafka Connect。TDengine Sink Connector 用于从 Kafka Connect 接收数据并写入 TDengine。

TDengine Database Kafka Connector -- streaming integration with kafka connect

什么是 Confluent？

Confluent 在 Kafka 的基础上增加很多扩展功能。包括：

Schema Registry
REST 代理
非 Java 客户端
很多打包好的 Kafka Connect 插件
管理和监控 Kafka 的 GUI —— Confluent 控制中心

这些扩展功能有的包含在社区版本的 Confluent 中，有的只有企业版能用。 TDengine Database Kafka Connector -- Confluent introduction

Confluent 企业版提供了 confluent 命令行工具管理各个组件。

前置条件

运行本教程中示例的前提条件。

Linux 操作系统
已安装 Java 8 和 Maven
已安装 Git
已安装并启动 TDengine。如果还没有可参考安装和卸载

安装 Confluent

Confluent 提供了 Docker 和二进制包两种安装方式。本文仅介绍二进制包方式安装。

在任意目录下执行：

curl -O http://packages.confluent.io/archive/7.1/confluent-7.1.1.tar.gz
tar xzf confluent-7.1.1.tar.gz -C /opt/test

然后需要把 $CONFLUENT_HOME/bin 目录加入 PATH。

export CONFLUENT_HOME=/opt/confluent-7.1.1
PATH=$CONFLUENT_HOME/bin
export PATH

以上脚本可以追加到当前用户的 profile 文件（~/.profile 或 ~/.bash_profile）

安装完成之后，可以输入confluent version做简单验证：

# confluent version
confluent - Confluent CLI

Version:     v2.6.1
Git Ref:     6d920590
Build Date:  2022-02-18T06:14:21Z
Go Version:  go1.17.6 (linux/amd64)
Development: false

安装 TDengine Connector 插件

从源码安装

git clone --branch master https://github.com:taosdata/kafka-connect-tdengine.git
cd kafka-connect-tdengine
mvn clean package
unzip -d $CONFLUENT_HOME/share/java/ target/components/packages/taosdata-kafka-connect-tdengine-*.zip

以上脚本先 clone 项目源码，然后用 Maven 编译打包。打包完成后在 target/components/packages/ 目录生成了插件的 zip 包。把这个 zip 包解压到安装插件的路径即可。上面的示例中使用了内置的插件安装路径： $CONFLUENT_HOME/share/java/。

用 confluent-hub 安装

Confluent Hub 提供下载 Kafka Connect 插件的服务。在 TDengine Kafka Connector 发布到 Confluent Hub 后可以使用命令工具 confluent-hub 安装。 TDengine Kafka Connector 目前没有正式发布，不能用这种方式安装。

启动 Confluent

confluent local services start

备注

一定要先安装插件再启动 Confluent, 否则加载插件会失败。

提示

若某组件启动失败，可尝试清空数据，重新启动。数据目录在启动时将被打印到控制台，比如：

Using CONFLUENT_CURRENT: /tmp/confluent.106668
Starting ZooKeeper
ZooKeeper is [UP]
Starting Kafka
Kafka is [UP]
Starting Schema Registry
Schema Registry is [UP]
Starting Kafka REST
Kafka REST is [UP]
Starting Connect
Connect is [UP]
Starting ksqlDB Server
ksqlDB Server is [UP]
Starting Control Center
Control Center is [UP]

清空数据可执行 rm -rf /tmp/confluent.106668。

验证各个组件是否启动成功

输入命令：

confluent local services status

如果各组件都启动成功，会得到如下输出：

Connect is [UP]
Control Center is [UP]
Kafka is [UP]
Kafka REST is [UP]
ksqlDB Server is [UP]
Schema Registry is [UP]
ZooKeeper is [UP]

验证插件是否安装成功

在 Kafka Connect 组件完全启动后，可用以下命令列出成功加载的插件：

confluent local services connect plugin list

如果成功安装，会输出如下：

Available Connect Plugins:
[
  {
    "class": "com.taosdata.kafka.connect.sink.TDengineSinkConnector",
    "type": "sink",
    "version": "1.0.0"
  },
  {
    "class": "com.taosdata.kafka.connect.source.TDengineSourceConnector",
    "type": "source",
    "version": "1.0.0"
  },
......

如果插件安装失败，请检查 Kafka Connect 的启动日志是否有异常信息，用以下命令输出日志路径：

echo `cat /tmp/confluent.current`/connect/connect.stdout

该命令的输出类似： /tmp/confluent.104086/connect/connect.stdout。

与日志文件 connect.stdout 同一目录，还有一个文件名为： connect.properties。在这个文件的末尾，可以看到最终生效的 plugin.path，它是一系列用逗号分割的路径。如果插件安装失败，很可能是因为实际的安装路径不包含在 plugin.path 中。

TDengine Sink Connector 的使用

TDengine Sink Connector 的作用是同步指定 topic 的数据到 TDengine。用户无需提前创建数据库和超级表。可手动指定目标数据库的名字（见配置参数 connection.database），也可按一定规则生成(见配置参数 connection.database.prefix)。

TDengine Sink Connector 内部使用 TDengine 无模式写入接口写数据到 TDengine，目前支持三种格式的数据：InfluxDB 行协议格式、 OpenTSDB Telnet 协议格式和 OpenTSDB JSON 协议格式。

下面的示例将主题 meters 的数据，同步到目标数据库 power。数据格式为 InfluxDB Line 协议格式。

添加配置文件

mkdir ~/test
cd ~/test
vi sink-demo.properties

sink-demo.properties 内容如下：

sink-demo.properties
name=TDengineSinkConnector
connector.class=com.taosdata.kafka.connect.sink.TDengineSinkConnector
tasks.max=1
topics=meters
connection.url=jdbc:TAOS://127.0.0.1:6030
connection.user=root
connection.password=taosdata
connection.database=power
db.schemaless=line
data.precision=ns
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

关键配置说明：

topics=meters 和 connection.database=power, 表示订阅主题 meters 的数据，并写入数据库 power。
db.schemaless=line, 表示使用 InfluxDB Line 协议格式的数据。

创建 Connector 实例

confluent local services connect connector load TDengineSinkConnector --config ./sink-demo.properties

若以上命令执行成功，则有如下输出：

{
  "name": "TDengineSinkConnector",
  "config": {
    "connection.database": "power",
    "connection.password": "taosdata",
    "connection.url": "jdbc:TAOS://127.0.0.1:6030",
    "connection.user": "root",
    "connector.class": "com.taosdata.kafka.connect.sink.TDengineSinkConnector",
    "data.precision": "ns",
    "db.schemaless": "line",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "tasks.max": "1",
    "topics": "meters",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "name": "TDengineSinkConnector"
  },
  "tasks": [],
  "type": "sink"
}

写入测试数据

准备测试数据的文本文件，内容如下：

test-data.txt
meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249000000
meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250000000
meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249000000
meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250000000

使用 kafka-console-producer 向主题 meters 添加测试数据。

cat test-data.txt | kafka-console-producer --broker-list localhost:9092 --topic meters

备注

如果目标数据库 power 不存在，那么 TDengine Sink Connector 会自动创建数据库。自动创建数据库使用的时间精度为纳秒，这就要求写入数据的时间戳精度也是纳秒。如果写入数据的时间戳精度不是纳秒，将会抛异常。

验证同步是否成功

使用 TDengine CLI 验证同步是否成功。

taos> use power;
Database changed.

taos> select * from meters;
              ts               |          current          |          voltage          |           phase           | groupid |            location            |
===============================================================================================================================================================
 2022-03-28 09:56:51.249000000 |              11.800000000 |             221.000000000 |               0.280000000 | 2       | California.LosAngeles          |
 2022-03-28 09:56:51.250000000 |              13.400000000 |             223.000000000 |               0.290000000 | 2       | California.LosAngeles          |
 2022-03-28 09:56:51.249000000 |              10.800000000 |             223.000000000 |               0.290000000 | 3       | California.LosAngeles          |
 2022-03-28 09:56:51.250000000 |              11.300000000 |             221.000000000 |               0.350000000 | 3       | California.LosAngeles          |
Query OK, 4 row(s) in set (0.004208s)

若看到了以上数据，则说明同步成功。若没有，请检查 Kafka Connect 的日志。配置参数的详细说明见配置参考。

TDengine Source Connector 的使用

TDengine Source Connector 的作用是将 TDengine 某个数据库某一时刻之后的数据全部推送到 Kafka。TDengine Source Connector 的实现原理是，先分批拉取历史数据，再用定时查询的策略同步增量数据。同时会监控表的变化，可以自动同步新增的表。如果重启 Kafka Connect, 会从上次中断的位置继续同步。

TDengine Source Connector 会将 TDengine 数据表中的数据转换成 InfluxDB Line 协议格式或 OpenTSDB JSON 协议格式，然后写入 Kafka。

下面的示例程序同步数据库 test 中的数据到主题 tdengine-source-test。

添加配置文件

vi source-demo.properties

输入以下内容：

source-demo.properties
name=TDengineSourceConnector
connector.class=com.taosdata.kafka.connect.source.TDengineSourceConnector
tasks.max=1
connection.url=jdbc:TAOS://127.0.0.1:6030
connection.username=root
connection.password=taosdata
connection.database=test
connection.attempts=3
connection.backoff.ms=5000
topic.prefix=tdengine-source-
poll.interval.ms=1000
fetch.max.rows=100
out.format=line
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

准备测试数据

准备生成测试数据的 SQL 文件。

prepare-source-data.sql
DROP DATABASE IF EXISTS test;
CREATE DATABASE test;
USE test;
CREATE STABLE meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT);
INSERT INTO d1001 USING meters TAGS(California.SanFrancisco, 2) VALUES('2018-10-03 14:38:05.000',10.30000,219,0.31000) d1001 USING meters TAGS(California.SanFrancisco, 2) VALUES('2018-10-03 14:38:15.000',12.60000,218,0.33000) d1001 USING meters TAGS(California.SanFrancisco, 2) VALUES('2018-10-03 14:38:16.800',12.30000,221,0.31000) d1002 USING meters TAGS(California.SanFrancisco, 3) VALUES('2018-10-03 14:38:16.650',10.30000,218,0.25000) d1003 USING meters TAGS(California.LosAngeles, 2) VALUES('2018-10-03 14:38:05.500',11.80000,221,0.28000) d1003 USING meters TAGS(California.LosAngeles, 2) VALUES('2018-10-03 14:38:16.600',13.40000,223,0.29000) d1004 USING meters TAGS(California.LosAngeles, 3) VALUES('2018-10-03 14:38:05.000',10.80000,223,0.29000) d1004 USING meters TAGS(California.LosAngeles, 3) VALUES('2018-10-03 14:38:06.500',11.50000,221,0.35000);

使用 TDengine CLI, 执行 SQL 文件。

taos -f prepare-source-data.sql

创建 Connector 实例

confluent local services connect connector load TDengineSourceConnector --config source-demo.properties

查看 topic 数据

使用 kafka-console-consumer 命令行工具监控主题 tdengine-source-test 中的数据。一开始会输出所有历史数据，往 TDengine 插入两条新的数据之后，kafka-console-consumer 也立即输出了新增的两条数据。

kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic tdengine-source-test

输出：

......
meters,location="California.SanFrancisco",groupid=2i32 current=10.3f32,voltage=219i32,phase=0.31f32 1538548685000000000
meters,location="California.SanFrancisco",groupid=2i32 current=12.6f32,voltage=218i32,phase=0.33f32 1538548695000000000
......

此时会显示所有历史数据。切换到 TDengine CLI，插入两条新的数据：

USE test;
INSERT INTO d1001 VALUES (now, 13.3, 229, 0.38);
INSERT INTO d1002 VALUES (now, 16.3, 233, 0.22);

再切换回 kafka-console-consumer，此时命令行窗口已经打印出刚插入的 2 条数据。

unload 插件

测试完毕之后，用 unload 命令停止已加载的 connector。

查看当前活跃的 connector：

confluent local services connect connector status

如果按照前述操作，此时应有两个活跃的 connector。使用下面的命令 unload：

confluent local services connect connector unload TDengineSourceConnector
confluent local services connect connector unload TDengineSourceConnector

配置参考

通用配置

以下配置项对 TDengine Sink Connector 和 TDengine Source Connector 均适用。

name: connector 名称。
connector.class: connector 的完整类名，如: com.taosdata.kafka.connect.sink.TDengineSinkConnector。
tasks.max: 最大任务数, 默认 1。
topics: 需要同步的 topic 列表，多个用逗号分隔, 如 topic1,topic2。
connection.url: TDengine JDBC 连接字符串，如 jdbc:TAOS://127.0.0.1:6030。
connection.user： TDengine 用户名，默认 root。
connection.password ：TDengine 用户密码，默认 taosdata。
connection.attempts ：最大尝试连接次数。默认 3。
connection.backoff.ms ：创建连接失败重试时间隔时间，单位为 ms。默认 5000。

TDengine Sink Connector 特有的配置

connection.database：目标数据库名。如果指定的数据库不存在会则自动创建。自动建库使用的时间精度为纳秒。默认值为 null。为 null 时目标数据库命名规则参考 connection.database.prefix 参数的说明
connection.database.prefix：当 connection.database 为 null 时, 目标数据库的前缀。可以包含占位符 ' $\{topic}'。比如 kafka_$ {topic}, 对于主题 'orders' 将写入数据库 'kafka_orders'。默认 null。当为 null 时，目标数据库的名字和主题的名字是一致的。
batch.size: 分批写入每批记录数。当 Sink Connector 一次接收到的数据大于这个值时将分批写入。
max.retries: 发生错误时的最大重试次数。默认为 1。
retry.backoff.ms: 发送错误时重试的时间间隔。单位毫秒，默认为 3000。
db.schemaless: 数据格式，可选值为：
1. line ：代表 InfluxDB 行协议格式
2. json : 代表 OpenTSDB JSON 格式
3. telnet ：代表 OpenTSDB Telnet 行协议格式
data.precision: 使用 InfluxDB 行协议格式时，时间戳的精度。可选值为：
1. ms ：表示毫秒
2. us ：表示微秒
3. ns ：表示纳秒。默认为纳秒。

TDengine Source Connector 特有的配置

connection.database: 源数据库名称，无缺省值。
topic.prefix：数据导入 kafka 后 topic 名称前缀。使用 topic.prefix + connection.database 名称作为完整 topic 名。默认为空字符串 ""。
timestamp.initial: 数据同步起始时间。格式为'yyyy-MM-dd HH:mm:ss'。默认为 "1970-01-01 00:00:00"。
poll.interval.ms: 拉取数据间隔，单位为 ms。默认为 1000。
fetch.max.rows : 检索数据库时最大检索条数。默认为 100。
out.format: 数据格式。取值 line 或 json。line 表示 InfluxDB Line 协议格式， json 表示 OpenTSDB JSON 格式。默认为 line。

其他说明

插件的安装位置可以自定义，请参考官方文档：https://docs.confluent.io/home/connect/self-managed/install.html#install-connector-manually。
本教程的示例程序使用了 Confluent 平台，但是 TDengine Kafka Connector 本身同样适用于独立安装的 Kafka, 且配置方法相同。关于如何在独立安装的 Kafka 环境使用 Kafka Connect 插件，请参考官方文档： https://kafka.apache.org/documentation/#connect。

问题反馈

无论遇到任何问题，都欢迎在本项目的 Github 仓库反馈： https://github.com/taosdata/kafka-connect-tdengine/issues。

什么是 Kafka Connect？​

什么是 Confluent？​

前置条件​

安装 Confluent​

安装 TDengine Connector 插件​

从源码安装​

用 confluent-hub 安装​

启动 Confluent​

验证各个组件是否启动成功​

验证插件是否安装成功​

TDengine Sink Connector 的使用​

添加配置文件​

创建 Connector 实例​

写入测试数据​

验证同步是否成功​

TDengine Source Connector 的使用​

添加配置文件​

准备测试数据​

创建 Connector 实例​

查看 topic 数据​

unload 插件​

配置参考​

通用配置​

TDengine Sink Connector 特有的配置​

TDengine Source Connector 特有的配置​

其他说明​

问题反馈​

参考​

支持和反馈

什么是 Kafka Connect？

什么是 Confluent？

前置条件

安装 Confluent

安装 TDengine Connector 插件

从源码安装

用 confluent-hub 安装

启动 Confluent

验证各个组件是否启动成功

验证插件是否安装成功

TDengine Sink Connector 的使用

添加配置文件

创建 Connector 实例

写入测试数据

验证同步是否成功

TDengine Source Connector 的使用

添加配置文件

准备测试数据

创建 Connector 实例

查看 topic 数据

unload 插件

配置参考

通用配置

TDengine Sink Connector 特有的配置

TDengine Source Connector 特有的配置

其他说明

问题反馈

参考