数据迁移脚本案例

使用Python脚本，将MySQL数据迁移至TDengine

项目背景：

公司调研过程时序数据库，做技术储备。最开始调研TDengine2.4，随时间推进官方升级2.6，在2.6上做了数据迁移，一批手机上传上来的定位数据，使用DATAX，使用mysqlreader加淘思官方提供的tdenginewriter插件，做好字段对应关系，写入数据，按阿里DATAX效率，开3个管道速度也并不快，走内网平均迁移速递大概在5000条/秒；到最近，涛思升级TDengine为3.0版本，很多附属工具基本重构，支持了流式计算，增加了社区版的数据删除等功能，这次使用DATAX适配失败，可能是改的太多了，索性尝试先使用python脚本迁移下。

说干就干，官方提供了快速迁移方案，但是数据源是虚拟出来的，并没有时间主键，而且拆分子表是按量拆分然后平均分配，索性直接放弃，但是里面的多线程写入还是要参考的。

由于是python初学者，类和一些库函数使用还不是很熟，简简单单一把梭，一条流水从头撸到尾。迁移也是尝试了很多方法，遇到了很多问题这里逐一介绍。

TDengine连接建立

官方提供文档，两种连接方式，原生客户端，restful方式。

taosc 本地客户端方式

安装驱动

安装包见官网 Linux|Windows

win安装，linux解压，执行 chmod+x install_client.sh && install_client.sh

配置客户端

前提是与服务端的6030端口，开启TCP和UDP的放行，如果是集群服务端，所有FQDN都需要添加端口两个协议的放行

修改cfg配置文件(win: C:\TDengine\cfg\taos.cfg linux：/etc/taos/taos.cfg) 修改 firstEP 为TDengine服务端的FirstEP


 # The end point of the first dnode in the cluster to be connected to when 
 `taosd` or `taos` is started
 firstEp                   tdengine1:6030

修改host文件，添加FQDN解析映射


 vim /etc/hosts
 192.168.1.xxx tdengine1
 192.168.1.xxx tdengine2
 192.168.1.xxx tdengine3

验证连接


 ping tdengine1
 ping tdengine2
 ping tdengine3
 telnet tdengine1 6030
 telnet tdengine2 6030
 telnet tdengine3 6030

taosc验证


$ taos -h tdengine1 -u root -p

taos > show databases;

安装python库


pip uninstall taos taospy
pip install taospy

此种安装方法最高版本2.4.x，使用pip_search查看最高2.5.1(2022.09.30)，还是使用git源直接安装靠谱：


pip install git+https://github.com/taosdata/taos-connector-python.git

但是吧，github方式国内连接不畅，可以下载包后来本地安装。

注意：要求python版本3.8以上，非官方文档上的3.6以上。

测试连接


import taos

def test_connection():
    # all parameters are optional.
    # if database is specified,
    # then it must exist.
    conn = taos.connect(host="localhost",
                        port=6030,
                        user="root",
                        password="taosdata",
                        database="log")
    print('client info:', conn.client_info)
    print('server info:', conn.server_info)
    conn.close()


if __name__ == "__main__":
    test_connection()

restful 方式

万能连接大法，能发送GET、POST请求就行。

与原生连接器的一个区别是，RESTful 接口是无状态的，因此 USE db_name 指令没有效果，所有对表名、超级表名的引用都需要指定数据库名前缀。支持在 RESTful URL 中指定 db_name，这时如果 SQL 语句中没有指定数据库名前缀的话，会使用 URL 中指定的这个 db_name。

确认服务端开启 taosadapter

firstEP服务端运行


systemctl status taosadapter

确保开放6041


lsof -nP -iTCP -sTCP:LISTEN

测试连接


curl -u root:password http://192.168.1.xxx:6041/rest/sql -d "select server_version()"

HTTP 响应码 response code 说明 200 正确返回和 C 接口错误返回 400 参数错误返回 401 鉴权失败 404 接口不存在 500 内部错误 503 系统资源不足

安装python包

同上一章
测试连接


from taosrest import connect, TaosRestConnection, TaosRestCursor

conn: TaosRestConnection = connect(url="http://localhost:6041",
                                   user="root",
                                   password="taosdata",
                                   timeout=30)

源数据batch

拆分、循环方法

因源数据有上亿级别数据，不可能一个SELECT撸到低，首先源表必须加上自增主键，或者带索引的可以作为分片键的字段(可以现加)，然后取个最大值，用for循环分批次去读取，如果数据处理模块、函数是要分多线程来取数据，需要有个全局变量记录不同线程之间取完数据后的增量值。

子表名称拼接

这里会考虑将源表代表设备、人等可分子表的关键键值，作为分组条件，取出各个唯一值，全量再去循环每个子表的固定batch，分批拼接INSERT语句后写入。这里结合上面的分批取源表数据，会有嵌套循环的场景，

先从全量数据取出固定分组，然后循环每个分组内的Id做for循环，这里好处是可以保证子循环里Id可以连续，不会按自增来跑空；
先取全量Id的连续值，在子循环里再找分组，这里好处是分组的压力对源库压力小，不至于直接给上亿条数据直接GROUP BY ；
还有一种是先取全量Id连续值，子循环里拆分组，这个分组之后，由于有设备、人等键的限制，Id是不连续的，那就再次来个for循环遍历分组内的结果，直接拼出来INSERT语句。

这里各种循环，没得用函数，也没得分文件，最开始需求就是要迁移数据，有思路更清晰的应该能更好的给模块化下。

TDengine语句拼接

TDengine对标签数据和子表内数据写入，和正常的SQL标准内的还稍微有些区别，和其他时序类的应该很相似，这里在官方文档里有段


INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) VALUES ('2021-07-13 14:06:34.630', 10.2, 219, 0.32) ('2021-07-13 14:06:35.779', 10.15, 217, 0.33)
            d21002 USING meters (groupId) TAGS (2) VALUES ('2021-07-13 14:06:34.255', 10.15, 217, 0.33)
            d21003 USING meters (groupId) TAGS (2) (ts, current, phase) VALUES ('2021-07-13 14:06:34.255', 10.27, 0.31);

需要拿出标签数据写在INSERT段里，然后VALUES值按行写在括号内。

简单日志

上面讲到有实际已经到了三个循环，可以用全局变量累计每个子循环的时间、返回行数的累计，计算出平均迁移速度，返回给前台。

后台运行


nohup python3.9 mysql2tdnative.py &

实际脚本

放托管平台了

GitHub

Gitee

Codeberg

本来想写一堆，写着写着忘词了~

目录