赞
踩
参考文档:时间序列数据库 TSDB_时间序列数据库 TSDB-阿里云帮助中心
时序数据是随时间不断产生的一系列数据,简单来说,就是带时间戳的数据。数据可能来自服务器和应用程序的指标、物联网传感器的读数、网站或应用程序上的用户交互或金融市场上的交易活动等。
时序数据的主要数据属性如下:
TSDB 技术支持与语法
参考文档:调用SDK写入数据_时间序列数据库 TSDB-阿里云帮助中心
TSDBConfig config = TSDBConfig // 配置地址,第一个参数可以是 TSDB 的域名或 IP。第二个参数表示 TSDB 端口。 .address(host, port) // user和password 表示用于用户认证的用户名和密码。TSDB用户可在实例管理控制台创建。如果实例未启用用户鉴权功能,则创建Config对象时无需调用basicAuth()方法。 .basicAuth(tsdbUser, basicPwd) // 只读开关,默认为 false。当 readonly 设置为 true 时,异步写开关会被关闭。 .readonly(false) // 网络连接池大小,默认为64。 .httpConnectionPool(connectPool) // HTTP 等待时间,单位为秒,默认为90秒。 .httpConnectTimeout(timeOut) // HTTP连接存在时间长度。单位为秒。默认为0,即不生效,为长连接。建议设置为一个合理值,服务端故障恢复后,客户端通过重新建立连接可达到服务端负载均衡。 .httpConnectionLiveTime(liveTime) // IO 线程数,默认为1。 .ioThreadCount(1) // 异步写开关。默认为 true。推荐异步写。 .asyncPut(true) // 异步写相关,客户端缓冲队列长度,默认为10000。 .batchPutBufferSize(batchPutBufferSize) // 异步写相关,缓冲队列消费线程数,默认为 1。 .batchPutConsumerThreadCount(batchPutConsumerThreadCount) // 异步写相关,每次批次提交给客户端点的个数,默认为 500。 .batchPutSize(batchPutSize) // 异步写相关,每次等待最大时间限制,单位为 ms,默认为 300。 .batchPutTimeLimit(batchPutTimeLimit) // 异步写相关,写请求队列数,默认等于连接池数。可根据读写次数的比例进行配置。 .putRequestLimit(putRequestLimit) //多值写入相关配置 .multiFieldBatchPutBufferSize(multiFieldBatchPutBufferSize) .multiFieldBatchPutConsumerThreadCount(multiFieldBatchPutConsumerThreadCount) // 流量限制,设置每秒最大提交 Point 的个数。 .maxTPS(maxTps) // 多值写入回调函数 .listenMultiFieldBatchPut(multiFieldCallback) // 单值写入回调函数 .listenBatchPut(callback) .config(); // 创建客户端 TSDB tsdb = TSDBClientFactory.connect(config):
// 构造 Point Point point = Point.metric("test1") .tag("tagk1", "tagv1") .tag("tagk2", "tagv2") .tag("tagk3", "tagv3") .timestamp(timestamp) .value(123.456) .build();
带回调可以获取:成功数、返回数和失败原因,不要在回调方法中做耗时操作;
出于写入性能的考虑,同步写建议手动将数据点打包成一批数据,并且建议这批数据包含500~1000个数据点。此外,也建议多个线程并发进行提交;
空返回:tsdb.putSync(points)
摘要返回:tsdb.putSync(points, SummaryResult.class);
详情返回:tsdb.putSync(points, DetailsResult.class);
空返回:tsdb.put(points)
摘要返回:tsdb.put(points, BatchPutSummaryCallback.class);
详情返回:tsdb.put(points, BatchPutDetailsCallback.class)
List<QueryResult> result = tsdb.query(query); System.out.println("返回结果:" + result);
QueryCallback callback = new QueryCallback() { @Override public void response(Query input, List<QueryResult> result) { System.out.println("查询参数:" + input); System.out.println("返回结果:" + result); } }; tsdb.query(query, callback);
语法如下
Query query = Query .timeRange(startTime, endTime) // 设置查询时间条件 // 设置子查询1 简单查询 .sub(SubQuery.metric("hello").aggregator(Aggregator.AVG).tag("tagk1", "tagv1").build()) // 设置子查询2 简单查询 .sub( SubQuery subQuery = SubQuery .metric("test-metric") .aggregator(Aggregator.AVG) .downsample("60m-avg") .tag("tagk1", "tagv1") .tag("tagk2", "tagv2") .build(); ) // 设置子查询3 复杂查询 .sub( SubQuery subQuery = SubQuery .metric("test-metric") .aggregator(Aggregator.AVG) .downsample("60m-avg") .filter(Filter .filter(FilterType.LiteralOr, "", "") .build() ) .build(); ) .build();
timeRange
左闭右闭,即包含开始事件,也包含结束事件;
FilterType 类别
LiteralOr:等于查询,或查询,类似 SQL 里的 IN 查询;
NotLiteralOr:等于查询,或查询,类似 SQL 里的 NOT IN 查询;
Wildcard:模糊匹配,类似 SQL 里的 like 查询;
Regexp:正则匹配;
假设有以下数据,共四条时间线;
时间线计算方式 = Metric + Tags, 即 Metric + orgCode + portId + inOut;
time | 数据组 | Timestamp | orgCode | portId | inOut | value |
19:10 | 1 | 1675077000000 | 123456 | 16604 | 8 | |
20:10 | 2 | 1675080600000 | 123456 | 16604 | 12 | |
21:10 | 3 | 1675084200000 | 123456 | 16601 | IN | 2 |
21:10 | 4 | 1675084200000 | 123456 | 16601 | OUT | 8 |
22:10 | 5 | 1675087800000 | 123456 | 16605 | 4 | |
23:10 | 6 | 1675091400000 | 123456 | 16606 | OUT | 6 |
aggregator 聚合
用于控制返回时间线的多少,即 QueryResult 的条数;
例如查询:
{ "start": 1675077000000, "end": 1675091400000, "queries": [ { "metric": "xx", "aggregator": "none", "tags": { "orgCode": "123456" } } ] }
会返回 5 组时间线,即 1、2 一组,3、4、5、6 各一组,时间线计算方式如上图所示;
{"tags": {"orgCode": "123456", "portId": "16604"},"dps": {1675077000000: 8.0, 1675080600000: 12.0}}, {"tags": {"orgCode": "123456","inOut": "IN", "portId": "16601"},"dps": {1675084200000: 2.0}}, {"tags": {"orgCode": "123456", "portId": "16605"},"dps": {1675087800000: 4.0}}, {"tags": {"orgCode": "123456","inOut": "OUT", "portId": "16606"},"dps": {1675091400000: 6.0}} {"tags": {"orgCode": "123456","portId": "16601", "inOut": "OUT"},"dps": {1675084200000: 8.0}},
aggregator 的作用则是:将除查询 tag 外的其他时间线合并,当 Timestamp 相同时,则会用 aggregator 指定的算法计算,例如:
{ "start": 1675077000000, "end": 1675091400000, "queries": [ { "metric": "xx", "aggregator": "sum", "tags": { "orgCode": "123456" } } ] }
返回 1 组时间线,因为 1、2、3、4、5 的 orgCode 都相同,会将 4 条时间合并,但 1675084200000 变成了 10,因为该时间点有两条数据,只是 tag 不同罢了,两条数据用了 sum:
{ "aggregateTags": [ "inOut", "portId" ], "dps": { 1675077000000: 8.0, 1675080600000: 12.0, 1675084200000: 10.0, 1675087800000: 4.0, 1675091400000: 6.0 }, "tags": { "orgCode": "123456" } }
downsample 降采样
用于控制返回时间线中数据点的多少,即 dps 的数;
格式:
[0all | (1)[ s | m | h | d ] ] - [ avg | sum | last | first | count | min | max .... ] [-none | null | zero ]
采样区间 区间操作 补全策略
采样区间计算方式:Timestamp - (Timestamp % Interval)
补全策略包括:
例如:
{ "start": 1675077000000, // 19:10 "end": 1675091400000, // 23:10 "queries": [ { "metric": "xx", "aggregator": "sum", "downsample": "2h-sum-none", "tags": { "orgCode": "123456" } } ] }
返回 2 小时一个汇总,即 18:00、20:00、22:00,关于区间的计算上面已有解释:
{ "aggregateTags": [ "inOut", "portId" ], "dps": { 1675072800000: 8.0, 1675080000000: 14.0, 1675087200000: 10.0 }, "tags": { "orgCode": "123456" } }
注:此处的 dps 开始时间 1675072800000 是 2023-01-30 18:00:00,为什么不是查询的开始时间,因为 tsdb 的采样区间算法是 Timestamp - (Timestamp % Interval),即 2023-01-30 19:10:00 - (2023-01-30 19:10:00 % 2h)= 2023-01-30 18:00:00,所以当使用采样时,往往计算的结果不准确,原因在这里;
以下参考文档:TSDB如何调用其他接口_时间序列数据库 TSDB-阿里云帮助中心
设置数据时效
tsdb.ttl(time);
查询 Metric,TagKey,TagValue
参数:(类型,模糊值,查询条数)
tsdb.suggest(Suggest.Metrics,"hel",10);
tsdb.suggest(Suggest.Tagk,"hel",10);
tsdb.suggest(Suggest.TagV,"hel",10);
TagKey 查 TagValue
参数:(类型,模糊值,查询条数)
tsdb.dumpMeta("tagk1","tagv1",10);
删除一段时间的数据
tsdb.deleteData("hello", startTime, nowTime);
删除指定时间线
Timeline timeline =Timeline.metric("hello").tag("tagk1","tagv1").build();
tsdb.deleteMeta(timeline);
清空所有表
tsdb.truncate()
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。