关于tdengine:如何使用-TDengine-Sink-Connector

小 T 导读：TDengine Kafka Connector 在 TDengine 的官网文档上放进去曾经有一段时间了，咱们也收到了一些开发者的反馈。文档中的教程应用 Confluent 平台（集成了 Kafka）演示了如何应用 Source Connector 和 Sink Connector，然而很多开发者在生产环境中并没有应用 Confluent，所以为不便大家，本文将应用独立部署的 Kafka 来演示。

本文蕴含以下内容：

如何应用 TDengine Sink Connector，把数据从 Kafka 同步到 TDengine。

TDengine Sink Connector 的实现原理。

一个简略的测试脚本，帮忙你在本人的环境中疾速测试。通过更改生成测试数据的程序和配置参数，你能够模仿本人的应用场景。

测试同步同一个 topic，应用不同分区数和不同 Sink 工作数对性能的影响。

如果你对文章结尾呈现的术语并不生疏，那么能够跳过这一部分。

Kafka 的外围是一个通用的、分布式的、可反复生产的音讯队列。

与之相比，作为一款时序数据库（Time-Series Database），TDengine 也可看作针对结构化的时序数据的音讯队列。

Kafka Connect 是 Kafka 的一个组件，简化了 Kafka 与其它数据源的集成。用户通过 Kafka Connect 读写 Kafka；通过 Kafka Connect 插件（也称 Kafka Connector）来读写各种数据源。

为不便集成，Kafka 曾经提供了生产者和消费者 API 以及客户端库，那为什么还须要 Kafka Connect 呢？因为一个好的 Kafka 客户端程序，不是单单生产或生产数据，还须要思考容错、重启、日志、弹性伸缩、序列化以及反序列化等。当开发者本人实现了这所有，就相当于开发了一个和 Kafka Connect 相似的货色。

与 Kafka 集成是 Kafka Connect 曾经解决的问题，用户不须要反复造轮子，只有多数边缘场景才须要定制化的集成计划。

TDengine Sink Connector 用于将 Kafka 中指定 topic 的数据（批量或实时）同步到 TDengine 的 database 中。

启动 Sink Connector 须要一个 properties 配置文件。具体配置见官网文档的配置参考。

Sink Connector 外部的实现非常简单，整体工作流程分为以下几个步骤：

Connect 框架依据配置启动 N 个消费者线程。
N 个消费者同时订阅数据，并用配置文件中指定的 key.converter 和 value.converter 做反序列化。
Connect 框架把反序列化后的数据传递给 N 个 SinkTask 的实例。
SinkTask 应用 TDengine 提供的 schemaless 写入接口来写入数据。

上述 4 个步骤，只有最初一步写数据是 Sink Connector 须要关怀的，其它都是 Connect 框架主动实现的。

上面重点探讨几个问题。

因为应用了 schemaless 写入接口，因而 TDengine Sink Connector 只反对三种格局的数据：InfluxDB 行协定格局、OpenTSDB Telnet 协定格局和 OpenTSDB JSON 协定格局。应用配置项 db.schemaless 来指定写入时应用的数据格式。例如：
db.schemaless=line
如果 Kafka 中的数据曾经是这三种格局之一，那么配置文件中的 value.converer，只需指定为 Connnect 内置的 org.apache.kafka.connect.storage.StringConverter。

value.converter=org.apache.kafka.connect.storage.StringConverter

如果 Kafka 中已有的数据不是上述三种之一，则须要实现本人的 Converter 类，将其转换为三种格局之一，这个链接兴许能帮到你。

既然 Connect 框架曾经帮咱们做了 Consumer 要做的事，那么咱们怎么来管制 Consumer 的行为呢？比方如何管制 Consumer 订阅的主题？如何管制 Consumer 每次 poll 的音讯数和工夫距离？

对于订阅哪些主题，能够用配置项 topics 来指定。

如果想笼罩 Consumer 的其它默认配置，能够间接在 Sink Connector 的配置文件中编写，然而要加前缀“consumer.override.”，比方想把每次 poll 的最大音讯数改为 3000，能够这样配置：
consumer.override.max.poll.records=3000

对于 Kafka Connect Sink，task 实质上就是消费者线程，接管从 topic 的分区读出来的数据。用配置参数 tasks.max 来管制最大工作数，一个工作一个线程。理论启动的工作数还与 topic 的分区数无关。如果你有 10 个分区，并且 tasks.max 设置为 5，那么每个 task 会收到 2 个分区的数据，并跟踪 2 个分区的 offsets。如果你配置的 tasks.max 比 partition 数大，Connect 会启动的 task 数与 topic 的 partition 数雷同。如果你订阅了 5 个 topic，每个 topic 都是 1 个分区，并且设置 tasks.max = 5, 那么理论会启动多少个工作呢？答案是 1 个，工作数与 topic 数量没有关系。

这一部分咱们在一台 Linux 服务器上搭建测试环境，并运行简略的示例程序。示例中将 Kafka 部署到了集体的 home 目录。操作时请留神把门路中的用户名（bding）替换为本人的用户名。

Java 1.8
Maven
装置并启动了 TDengine 相干服务过程：taosd 和 taosAdapter。

wget https://dlcdn.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xzf kafka_2.13-3.2.0.tgz

编辑 .bash_profile，退出：

export KAFKA_HOME=/home/bding/kafka_2.13-3.2.0
export PATH=$PATH:$KAFKA_HOME/bin

source .bash_profile

配置 Kafka Connect 加载插件的门路。

cd kafka_2.13-3.2.0/config/
vi connect-standalone.properties

追加
plugin.path=/home/bding/connectors
批改 Connector 插件的日志级别。这一步十分重要，咱们将通过插件的日志统计同步数据破费的工夫。
vi connect-log4j.properties
追加

log4j.logger.com.taosdata.kafka.connect.sink=DEBUG

git clone git@github.com:taosdata/kafka-connect-tdengine.git
cd kafka-connect-tdengine
mvn clean package
unzip -d ~/connectors target/components/packages/taosdata-kafka-connect-tdengine-*.zip

zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

kafka-topics.sh --create --topic meters --partitions 1 --bootstrap-server localhost:9092

将下列脚本保留为 gen-data.py：

#!/usr/bin/python3
import random
import sys
topic = sys.argv[1]
count = int(sys.argv[2])
start_ts = 1648432611249000000
location = ["SanFrancisco", "LosAngeles", "SanDiego"]
for i in range(count):
    ts = start_ts + i
    row = f"{topic},location={location[i % 3]},groupid=2 current={random.random() * 10},voltage={random.randint(100, 300)},phase={random.random()} {ts}"
    print(row)

而后执行：

python3 gen-data.py meters 10000  | kafka-console-producer.sh --broker-list localhost:9092 --topic meters

生成 10000 条 InfluxDB 行协定格局的数据到 topic meters。每条数据又蕴含 2 个标签字段和 3 个数据字段。

将下列配置保留为 sink-test.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
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

而后执行：

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

应用 TDengine CLI 查问 power 数据库 meters 表，查看是否正好蕴含 10000 条数据。

[bding@vm95 test]$ taos
 
Welcome to the TDengine shell from Linux, Client Version:2.6.0.4
Copyright (c) 2022 by TAOS Data, Inc. All rights reserved.
taos> select count(*) from power.meters;
       count(*)        |
========================
                 10000 |

这一部分，咱们将下面示例步骤中的第四步到第七步封装成可反复运行的 shell 脚本，并做以下批改：

将 topic 的分区数作为脚本的第 1 个参数，同时配置 tasks.max，使其等于分区数。这样咱们能够管制每次测试应用的写入线程数。
将生成测试数据的条数作为脚本的第 2 个参数，用来管制每次测试同步的数据量。
启动测试前清空所有数据，测试完结后进行 Connect、Kafka 和 ZooKeeper。

每次测试都先写数据到 Kafka，而后再启动 Connect 同步数据到 TDengine，这样做能够把同步数据的压力全副集中到 Sink 插件这边。咱们统计 Sink Connector 从接管到第一批数据到接管到最初一批数据之间的工夫，作为同步数据的总耗时。

残缺脚本如下：

#!/bin/bash
if [$# -lt 2];then
        echo  "Usage: ./run-test.sh <num_of_partitions>  <total_records>"
        exit 0
fi
echo "---------------------------TEST STARTED---------------------------------------"
echo clean data and logs
taos -s "DROP DATABASE IF EXISTS power"
rm -rf /tmp/kafka-logs /tmp/zookeeper
rm -f $KAFKA_HOME/logs/connect.log
np=$1     # number of partitions
total=$2  # number of records
echo number of partitions is $np, number of recordes is $total.
echo start zookeeper
zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties
echo start kafka
sleep 3
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
sleep 5
echo create topic
kafka-topics.sh --create --topic meters --partitions $np --bootstrap-server localhost:9092
kafka-topics.sh --describe --topic meters --bootstrap-server localhost:9092
echo generate test data
python3 gen-data.py meters $total  | kafka-console-producer.sh --broker-list localhost:9092 --topic meters
 
echo alter connector configuration setting tasks.max=$np
sed -i  "s/tasks.max=.*/tasks.max=${np}/"  sink-test.properties
echo start kafka connect
connect-standalone.sh -daemon $KAFKA_HOME/config/connect-standalone.properties sink-test.properties
 
echo -e "\e[1;31m open another console to monitor connect.log. press enter when no more data received.\e[0m"
read
 
echo stop connect
jps | grep ConnectStandalone | awk '{print $1}' | xargs kill
echo stop kafka server
kafka-server-stop.sh
echo stop zookeeper
zookeeper-server-stop.sh
# extract timestamps of receiving the first batch of data and the last batch of data
grep "records" $KAFKA_HOME/logs/connect.log  | grep meters- > tmp.log
start_time=`cat tmp.log | grep -Eo "[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2},[0-9]{3}" | head -1`
stop_time=`cat tmp.log | grep -Eo "[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2},[0-9]{3}" | tail -1`
 
echo "--------------------------TEST FINISHED------------------------------------"
echo "| records | partitions | start time | stop time |"
echo "|---------|------------|------------|-----------|"
echo "| $total | $np | $start_time | $stop_time |"

如果要测试应用 1 个分区，共 100 万条数据的性能，能够这样执行：
./run-test.sh 1 1000000
执行过程的截图如下：

留神两头有一个交互过程。因为脚本无奈确定数据是否同步完，须要用户监控 connect.log 来确定是否曾经生产完了所有数据，例如：

[bding@vm95 ~]$ cd kafka_2.13-3.2.0/logs/
[bding@vm95 logs]$ tail -f connect.log
[2022-06-21 17:39:00,176] DEBUG [TDengineSinkConnector|task-0] Received 500 records. First record kafka coordinates:(meters-0-314496). Writing them to the database... (com.taosdata.kafka.connect.sink.TDengineSinkTask:101)
[2022-06-21 17:39:00,180] DEBUG [TDengineSinkConnector|task-0] Received 500 records. First record kafka coordinates:(meters-0-314996). Writing them to the database... (com.taosdata.kafka.connect.sink.TDengineSinkTask:101)

当日志不再滚动，就阐明曾经生产完了

写入速度与数据量和线程数的关系表

上表第 1 列为总数据量，第 1 行为消费者线程数，也是写入线程数。两头为均匀每秒写入记录数。

写入速度与数据量和线程数的关系图

从上图能够看出，雷同数据量，线程越多写入速度越快。当应用单线程写入时，每秒能写入大略 10 万以上。当应用 5 个线程写入时，每秒写入大略 35 万左右。当应用 10 个线程时，每秒能写入 55 万左右。

写入速度比拟安稳，与总数据量关系不大。

同时也发现线程减少越多，线程减少带来的速度晋升越少。线程数从 1 变到 10，速度只从 10 万变到 50 万。可能的起因是数据在各个分区散布不平均。有的 task 执行工夫长，有的 task 执行工夫短，数据量越大，数据歪斜越大。比方 1000 万数据，10 个分区的时候，各分区的数据量：

[bding@vm95 kafka-logs]$ du -h ./ -d 1
125M    ./meters-8
149M    ./meters-7
119M    ./meters-9
138M    ./meters-4
110M    ./meters-3
158M    ./meters-6
131M    ./meters-5
105M    ./meters-0
113M    ./meters-2
99M     ./meters-1

另一个影响多线程写入速度的是数据的乱序水平。本测试场景中，多条工夫线的数据随机调配到了不同分区，当单线程写入时（即 1 个分区时），数据是严格有序的，写入速度最快。线程越多乱序水平越大。

所以在理论利用场景中，倡议将同一个子表的数据，放在 Kafka 同一个分区中。

本文中用到的所有代码和原始测试后果数据都已上传到 GitHub 仓库。

想理解更多 TDengine Database 的具体细节，欢送大家在 GitHub 上查看相干源代码。

关于tdengine:如何使用-TDengine-Sink-Connector

背景常识

· 什么是 Kafka？

· 什么是 Kafka Connect? 为什么应用 Kafka Connect？

TDengine Sink Connector 的实现原理

· 反对的数据格式

· 如何指定 Consumer 的参数？

· 如何管制写入线程数？

TDengine Sink Connector 应用示例

· 环境筹备

第一步：装置 Kafka

第二步：配置 Kafka

第三步：编译并装置插件

第四步：启动 ZooKeeper Server 和 Kafka Server

第五步：创立 topic

第六步：生成测试数据

第七步：启动 Kafka Connect

第八步：查看 TDengine 中的数据

TDengine Sink Connector 性能测试

· 测试流程

· 测试后果

后果剖析

附录

· 测试程序

· 测试环境