展开查看详情
1 .使用SeaTunnel搞定
TDengine数据同步
TDengine-Connector 解析与使用示例
李宏宇
2023.4.18
2 . 01 SeaTunnel基本概念介绍
02 TDengine Connector 功能特性
03 TDengine Connector 读取解析
CONTENT 04 TDengine Connector 写入解析
05 TDengine Connector 使用示例
06 如何参与贡献
4 .SeaTunnel 示例
数据集成领域的DSL
5 .SeaTunnel 整体架构
Source Sink
API
DataType State
Translation
Engine
Zeta
Zeta:https://www.bilibili.com/read/cv20989971
6 .SeaTunnel Connector 简介
Connector 由 Source + Sink 组成
Source 负责读取数据
Sink 负责写出数据 Source
抽象层SeaTunnelRow负责装载数据本身和元数据
Abstract Layer SeaTunnelRow
通过 Source → 抽象层 → Sink 组合，
完成多种数据源间搬运数据
Sink
7 .SeaTunnel Connector 生态
已完成 100+ 进行中 300+
具体详见：
https://seatunnel.incubator.apache.org/docs/2.3.1/Connector-v2-release-state
https://github.com/apache/incubator-seatunnel/issues/3018
8 .02 TDengine Connector 功能特性
9 .TDengine重点概念说明
TDengine是一款国产时序数据库，数据按时间先后顺序存储和检索
Super Table (STable) Subtable
用于代表类型相同的一类数据采集点， 用于代表某一个数据采集点，如xxx户的电
如电表 表
规定schema（如电压、电流等） 使用Stable schema，具体存储数据
面向用户 内部结构
可类比Class 可类比Object
1 N
理念：一个设备一张subtable
10 .Source功能特性
功能 解释 是否支持 如何支持
通过查询stable实现批量读取设备id
batch read 批量读取 OK 把subtable_name放到SeatunnelRow第
一位（携带元数据）
stream read 流式读取 需要引入TDengine Data Subscription
schema projection 字段投影 OK 使用 sql 选择字段
datatype mapping 数据类型映射 OK 全部支持
parallelism 并行读取 OK 按subtable 切分split，不按时间
state-restore 状态恢复 OK 存储未读取的split到state
at-least-once 至少一次 OK
at-most-once 至多一次
exactly-once 刚好一次
11 .Sink功能特性
功能 解释 是否支持 如何支持
Metrics/tags 等元数
metadata inject OK 支持从 SeaTunnelRow 中提取
据自动识别
datatype mapping 数据类型映射 OK 全部支持
batch submit 批量提交 受限于Seatunnel的SinkWriter API
受限于taos-ws driver的访问方式，需要在
retry 异常重试
应用程序中实现
at-least-once 至少一次 OK
timezone 目标端时区 OK 支持动态指定数据时区，如UTC等
12 .03 TDengine Connector 读取解析
13 .字段类型映射
TDengine dataType SeaTunnel dataType
BOOL BOOLEAN
SMALLINT
INT
INT
INT_UNSIGNED
LONG
BIGINT
BIGINT_UNSIGNED
DecimalType
DECIMAL
FLOAT FLOAT
DOUBLE DOUBLE
JSON
STRING
VARCHAR
TIMESTAMP LOCAL_DATE_TYPE
BINARY INSTANCE
14 .批量读取
Select f1,f2
from subtable1（子表）
where
ts > 1681352000
and ts <1681352777
Select f1,f2 Select f1,f2
from stable（超级表） from subtable2（子表）
where where
ts > 1681352000 ts > 1681352000
and ts <1681352777 and ts <1681352777
Select f1,f2
from subtable3（子表）
where
ts > 1681352000
and ts <1681352777
15 . 并行读取：切分Split，分配Reader
Stable
TDengine
Subtable1 Subtable2 Subtable3
Split1 Split2 Split3
Seatunnel
Reader1 Reader2
Flink/Spark subtask1 subtask2
16 .状态恢复
Reader 存储未读取的 Split 到 State
Reader 读取 Split 时获取 State Lock
17 .04 TDengine Connector 写入解析
19 .元数据的识别和注入
INSERT INTO subtable1 using stable tags ( tag1,tag2 ) VALUES ( v1,v2,v3 )
TDengine metadata自动识别
SeatunnelRow里提取 Config配置里读取
+SeatunnelRow里提取
20 .其它功能
状态管理和数据恢复
同Source逻辑
时区配置
时序数据库特有需求场景，有些时候需要集成时更改时区
因此提供了timezone的配置参数，转换指定时区
21 .05 TDengine Connector 使用示例
23 .TDengine Connector 的下一步计划
Source端：
支持streaming数据集成（目前仅支持batch操作）
Sink端：
支持数据写入异常重试机制
支持数据批量写入（目前仅支持批量读取，单条写入）
26 .欢迎参与 SeaTunnel 社区贡献
寻找你感兴趣的 issue
• https://github.com/apache/incubator-seatunnel/issues/2828
• https://github.com/apache/incubator-seatunnel/issues/3018
• https://github.com/apache/incubator-seatunnel/issues?q=is:open+is:issue+label:"help+wanted"
参考贡献指南
• https://github.com/apache/incubator-seatunnel/issues/2828
• https://github.com/apache/incubator-seatunnel/pull/2995【贡献指南】
• https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/contribution/setup.md 【编译指南】
新手教程
• https://mp.weixin.qq.com/s/evYRsrpPb-MXEm4DqLdnvQ【SeaTunnel 连接器极简开发流程】
• https://mp.weixin.qq.com/s/8ivksAePmVZe2_2ldGi0AA 【新 API Connector 开发解析】
• https://mp.weixin.qq.com/s/qpO2SVRl9KAY-Ib2kaTDig 【Source 与 Sink API 设计解析】
参与讨论 & 寻求帮助
• 在邮件列表、Slack 中讨论
• 通过微信群沟通（如果没有加入请关注 SeaTunnel 公众号入群）
• 参与 PR Review 发表你的见解
27 .website: https://seatunnel.apache.org
GitHub：https://github.com/apache/incubator-seatunnel
Slack: https://apacheseatunnel.slack.com
Twitter : https://twitter.com/asfseatunnel
Video：https://space.bilibili.com/1542095008
热门城市
