使用datax将数据从InfluxDB抽取到TDengine过程记录

1. 编写InfluxDB数据查询语句

select  time as ts,device as tbname, ip,device as district_code from "L2_CS" limit 1000

2. 创建TDengine表

create database if not exists sensor; create stable if not exists sensor.water(ts timestamp, ip varchar(50), device varchar(50)) tags(district_code nchar(12));

3, 编写datax配置json文件

{ "job": { "content": [ { "reader": { "name": "influxdbreader", "parameter": { "column": [ "*" ], "connection": [ { "endpoint": "http://127.0.0.7:8086", "database": "DisasterCloudDB" } ], "querySql": "select time as ts,device as tbname, ip,device as district_code from \"L2_CS\" limit 1000", "connTimeout": 15, "readTimeout": 20, "writeTimeout": 20, "username": "monitor_ro", "password": "HGhcxCFkacxdmBeU" } }, "writer": { "name": "tdengine30writer", "parameter": { "username": "jcyj", "password": "XquFvQoCM#I1sn6jc", "column": [ "ts", "tbname", "ip", "district_code" ], "connection": [ { "table": [ "water" ], "username": "jcyj", "password": "XquFvQoCM#I1sn6jc", "jdbcUrl": "jdbc:TAOS-RS://127.0.0.1:6041/jcyj?timestampFormat=TIMESTAMP" } ], "batchSize": 100, "ignoreTagsUnmatched": true } }, "transformer": [ { "name": "dx_replace", "parameter": { "columnIndex":2, "paras":["1","2","d50"] } } ] } ], "setting": { "speed": { "channel": 1 } } } }

增加  device as tbname 后，会根据设备id自动创建子表，不同的设备会创建不同的表

{ "name": "dx_replace", "parameter": { "columnIndex":2, "paras":["1","2","d50"] } }

使用 dx_replace 插件处理设备id全是数字，建表之后不能查看问题，这里替换前两个字符 增加字母d

最后来一张生成的结果图