Rollup 预处理示例

1. 验证准备

通过实例控制台添加预降采样(Rollup)规则:

2. 写入验证数据

这里通过压测程序按照每秒一个数据点,持续写入数据:

 public static void main(String[] args) throws InterruptedException, IOException {
        //TSDS预发环境实例
     HiTSDBConfig config = HiTSDBConfig.address("{tsdb实例ID}.hitsdb.rds.aliyuncs.com", 3242)
          .listenBatchPut(new BatchPutCallback()
                          {
                            @Override
                            public void response(String s, List<Point> list, Result result) {
                              Point p = Iterables.getFirst(list, null);
                              System.out.println("Put datapoint in second:" + (p == null ? null : p.getTimestamp()));
                            }
                            @Override
                            public void failed(String address, List<Point> request, Exception ex) {
                              System.err.println(ex);
                            }
                          })
          .config();
        HiTSDB tsdb = HiTSDBClientFactory.connect(config);
        // 构造数据并写入 HiTSDB
        for (int i = 0; i < 3600 * 2; i++) {
            int second = Math.toIntExact(System.currentTimeMillis() / 1000);
            List<Point> points = new ArrayList<Point>();
            points.add(Point
                           .metric("test.rollup.t5")
                           .tag("tagk1", "tagv1")
                           .value(second, Math.random())
                           .build()
            );
            points.add(Point
                           .metric("test.rollup.t5")
                           .tag("tagk1", "tagv1")
                           .tag("tagk2", "tagv2")
                           .value(second, Math.random())
                           .build()
            );
            points.add(Point
                           .metric("test.rollup.t5")
                           .tag("tagk1", "tagv1")
                           .tag("tagk2", "tagv2")
                           .tag("tagk3", "tagv3")
                           .value(second, Math.random())
                           .build()
            );
            tsdb.put(points);
            System.out.println(i);
            Thread.sleep(1000);  // 1秒提交1次
        }
        // 安全关闭客户端，以防数据丢失。
        System.out.println("关闭");
        tsdb.close();
    }

3.验证任务状态

数据写入过程中,规则的状态会不断更新:

三个状态指标:

• 规则命中数据点(5分钟命中率)
• 规则堆积数据点
• 规则输出数据点(5分钟命中率)

4.查询验证

4.1 验证原始数据查询

查询请求(存储为 query.json 文件):

{
    "start":1560742264,
    "end":1560742564,
    "queries":[
        {
            "aggregator":"sum",
            "downsample":"1m-sum",
            "metric":"test.rollup.t5",
            "tags":{
                "tagk1":"tagv1",
                "tagk2":"tagv2",
                "tagk3":"tagv3"
            }
        }
    ]
}

查询命令:

curl -H "Accept: application/json" -H "Content-type: application/json" -X POST --data @query.json http://实例IP:3242/api/query

查询结果[原始]:

4.1 验证预处理数据查询

查询请求(存储为 query_rollup.json 文件):

{
    "start":1560742264,
    "end":1560742564,
    "queries":[
        {
            "aggregator":"avg",
            "downsample":"1m-sum",
            "metric":"test.rollup.t5",
            "tags":{
                "tagk1":"tagv1",
                "tagk2":"tagv2",
                "tagk3":"tagv3",
                "__granularity__":"1m.sum"
            }
        }
    ]
}

查询命令:

curl -H "Accept: application/json" -H "Content-type: application/json" -X POST --data @query_rollup.json http://实例IP:3242/api/query

查询结果[预降采样]:

注意预降采样的查询结果中, 包含了 rollupGranularity 这个Tag, 用于表示本次启用了预计算查询,且具体的扫描算子.