通过 SDK 使用分析存储

注意事项

• 时序分析存储功能从 2023 年 12 月 20 日正式开始收费。

• 目前支持使用时序分析存储功能的地域有华东1（杭州）、华东2（上海）、华北2（北京）和华北3（张家口）。

• 只支持在时序模型实例中使用分析存储功能。

如果在使用过程中遇到问题，请通过钉钉加入用户群 44327024（物联网存储 IoTstore 开发者交流群）联系我们。

准备工作

• 已创建时序模型实例。具体操作，请参见创建时序模型实例。

• 已初始化客户端。具体操作，请参见初始化 OTSClient。

接口

步骤一：创建分析存储

为时序表创建分析存储后，您可以使用分析存储快速查询与分析时序数据。您可以在创建时序表时创建分析存储，也可以为已有时序表创建分析存储，请根据实际情况选择。

private static void createTimeseriesTable(TimeseriesClient client) {
    String tableName = "test_timeseries_table";
    TimeseriesTableMeta timeseriesTableMeta = new TimeseriesTableMeta(tableName);
    int timeToLive = -1;
    timeseriesTableMeta.setTimeseriesTableOptions(new TimeseriesTableOptions(timeToLive));
    CreateTimeseriesTableRequest request = new CreateTimeseriesTableRequest(timeseriesTableMeta);
    
    client.createTimeseriesTable(request);
}

private static void createTimeseriesTable(TimeseriesClient client) {
    // 设置时序表名称和TTL。
    String tableName = "test_timeseries_table";
    TimeseriesTableMeta timeseriesTableMeta = new TimeseriesTableMeta(tableName);
    int timeToLive = -1;
    timeseriesTableMeta.setTimeseriesTableOptions(new TimeseriesTableOptions(timeToLive));
    CreateTimeseriesTableRequest request = new CreateTimeseriesTableRequest(timeseriesTableMeta);

    // 自定义分析存储。
    List<TimeseriesAnalyticalStore> analyticalStores = new ArrayList<TimeseriesAnalyticalStore>();
    analyticalStores.add(new TimeseriesAnalyticalStore("test_analytical_store"));
    request.setAnalyticalStores(analyticalStores);

    client.createTimeseriesTable(request);
}

public void createAnalyticalStore(TimeseriesClient client) {
    //设置分析存储名称。
    TimeseriesAnalyticalStore analyticalStore = new TimeseriesAnalyticalStore("test_analytical_store");
    //设置分析存储的数据生命周期。单位为秒。
    analyticalStore.setTimeToLive(2592000);
    //设置分析存储同步时序表中数据的方式。
    analyticalStore.setSyncOption(AnalyticalStoreSyncType.SYNC_TYPE_INCR);
    //设置时序表名称。
    CreateTimeseriesAnalyticalStoreRequest request = new CreateTimeseriesAnalyticalStoreRequest("test_timeseries_table", analyticalStore);
    client.createTimeseriesAnalyticalStore(request);
}

步骤二：使用 SQL 查询数据

创建分析存储的映射关系后，使用 SELECT 语句查询与分析时序。

1. 创建分析存储的映射关系。

   示例如下：

   private static void createTable(SyncClient client) {
       // 创建SQL请求。
       SQLQueryRequest createTableRequest = new SQLQueryRequest("CREATE TABLE `device::cpu` " +
           "(`_m_name` VARCHAR(1024)," +
           " `_data_source` VARCHAR(1024)," +
           " `_tags` VARCHAR(1024)," +
           " `_time` BIGINT(20), " +
           " `usage_user` BIGINT(20), " +
           " `usage_system` BIGINT(20), " +
           " `usage_idle` BIGINT(20), " +
           " `usage_nice` BIGINT(20), " +
           " `usage_iowait` BIGINT(20), " +
           " `usage_irq` BIGINT(20), " +
           " `usage_softirq` BIGINT(20), " +
           " `usage_steal` BIGINT(20)," +
           " `usage_guest` BIGINT(20), " +
           " `usage_guest_nice` BIGINT(20)," +
           " `one_hour_stamp` BIGINT(20)," +
           " `one_minute_stamp` BIGINT(20)," +
           " PRIMARY KEY(`_m_name`,`_data_source`,`_tags`,`_time`)) ENGINE=AnalyticalStore;");
       // 获取SQL的响应结果。
       SQLQueryResponse response = client.sqlQuery(createTableRequest);
   }

2. 查询数据。

   使用 SELECT 语句查询所需数据。更多示例请参见数据查询示例。

   • 示例一：查询所有设备在 2023-01-05 05:14:00 到 2023-01-07 09:14:00 时间段内每天的最大 usage_irq 和最大 usage_softirq。

     private static void queryData(SyncClient client) {      
         SQLQueryRequest selectRequest = new SQLQueryRequest("SELECT time_bin(_time,\"1day\"), max(usage_irq),max(usage_softirq) " +
                 "FROM `device::cpu` " +
                 "WHERE _time > unix_timestamp_micros(\"2023-01-05 05:14:00.000000\") " +
                 "AND _time < unix_timestamp_micros(\"2023-01-07 09:14:00.000000\") " +
                 "GROUP BY 1 ORDER BY 1;");  //生成sql查询请求
         SQLQueryResponse selectResponse = client.sqlQuery(selectRequest); //执行SQL请求，获取查询请求应答。
         System.out.println("response type: " + selectResponse.getSQLStatementType());
         SQLTableMeta selectMeta = selectResponse.getSQLResultSet().getSQLTableMeta();//获取结果schema。
         System.out.println(selectMeta.getSchema());
         SQLResultSet selectResultSet = selectResponse.getSQLResultSet();//获取查询结果集。
         while (selectResultSet.hasNext()) {
             SQLRow row = selectResultSet.next();
             System.out.println(row.toDebugString());
             System.out.println(row.getDateTime(0).withZoneSameInstant(ZoneId.systemDefault()) + ", " + row.getLong(1) + ", " + row.getLong(2));
         } //打印结果。
     }

   • 示例二：查询在 host_50625 且 cpu 架构为 x64 的设备所有时间平均 usage_nice。

     private static void queryData(SyncClient client) {
         SQLQueryRequest selectRequest = new SQLQueryRequest("select avg(usage_nice) from `device::cpu` " +
                 "where _data_source = \"host_50625\" " +
                 "and tag_value_at(_tags,\"arch\") = \"x64\";");  //生成SQL查询请求。
         SQLQueryResponse selectResponse = client.sqlQuery(selectRequest); //执行SQL请求，获取查询请求应答。
         System.out.println("response type: " + selectResponse.getSQLStatementType());
         SQLTableMeta selectMeta = selectResponse.getSQLResultSet().getSQLTableMeta();//获取结果schema。
         System.out.println(selectMeta.getSchema());
         SQLResultSet selectResultSet = selectResponse.getSQLResultSet();//获取查询结果集。
         while (selectResultSet.hasNext()) {
     	      SQLRow row = selectResultSet.next();
             System.out.println(row.toDebugString());
             System.out.println(row.getDouble(0));
         }  //打印结果。
     }  

附录：管理分析存储

创建分析存储后，请根据需要执行相应操作。

public void describeAnalyticalStore(TimeseriesClient client) {
    //设置时序表名称和分析存储名称。
    DescribeTimeseriesAnalyticalStoreRequest request = new DescribeTimeseriesAnalyticalStoreRequest("test_timeseries_table", "test_analytical_store");
    DescribeTimeseriesAnalyticalStoreResponse response = client.describeTimeseriesAnalyticalStore(request);
    //打印分析存储名称。
    System.out.println("AnalyticalStoreName: " + response.getAnalyticalStore().getAnalyticalStoreName());
    //打印分析存储的数据生命周期，单位为秒。
    System.out.println("TimeToLive: " + response.getAnalyticalStore().getTimeToLive());
    //打印分析存储的同步选项。
    System.out.println("SyncOption: " + response.getAnalyticalStore().getSyncOption());
    //打印分析存储当前时间的同步阶段。
    if (response.getSyncStat() != null) {
        System.out.println("SyncPhase: " + response.getSyncStat().getSyncPhase());
        System.out.println("CurrentSyncTimestamp: " + response.getSyncStat().getCurrentSyncTimestamp());
    }
    //打印分析存储当前时间的存储量大小。
    if (response.getStorageSize() != null) {
        System.out.println("StorageSize: " + response.getStorageSize().getSizeInBytes());
        System.out.println("StorageSizeTimestamp: " + response.getStorageSize().getTimestamp());
    }
}

public void updateAnalyticalStore(TimeseriesClient client) {
    //设置分析存储名称。
    TimeseriesAnalyticalStore store = new TimeseriesAnalyticalStore("test_analytical_store");
    //更新分析存储的数据生命周期为2592000秒。
    store.setTimeToLive(2592000);
    //设置时序表名称。
    UpdateTimeseriesAnalyticalStoreRequest request = new UpdateTimeseriesAnalyticalStoreRequest("test_timeseries_table");
    request.setAnalyticStore(store);
    client.updateTimeseriesAnalyticalStore(request);
}

public void deleteAnalyticalStore(TimeseriesClient client) {
    //设置时序表名称和分析存储名称。
    DeleteTimeseriesAnalyticalStoreRequest request = new DeleteTimeseriesAnalyticalStoreRequest("test_timeseries_table", "test_analytical_store");
    request.setDropMappingTable(true);
    client.deleteTimeseriesAnalyticalStore(request);
}