时序数据存储管理 - 阿里云物联网平台

如果您购买了企业版实例，可以享受时序数据存储服务。公共实例不提供此服务。关于如何购买企业版实例，请参见实例管理。

什么是时序数据

时序数据存储将一个metric与一组tag的组合称为一条时间线，在一条时间线下面，连续时间点的采样数据则为时序数据。

例如：下图中有3条时间线，{“metric”: “cpu”,“tags”: “site”: “et2”, “ip”: “1.1.1.1”}这个metric与tag的组合为一条时间线，在同一条时间线下面存储连续的时序数据(timestamp,value)。

时序数据存储功能

时序数据存储提供以下功能：

时序数据高效读写：
- 数据写入：支持通过配置数据流转规则，将数据转发到实例内的时序数据存储进行数据写入；也支持通过SDK写入数据。最高可以支持每秒千万数据点的写入。
- 数据查询：支持通过SDK使用SQL，或通过JDBC连接使用TSQL进行数据的查询操作。百万数据点的读取，响应时间小于5秒。
数据管理：
您可以通过控制台设置数据的有效期。开启并设置数据时效后，系统将过期数据及时标记为失效，并自动在特定时间清除。
高压缩比存储：
单个数据点的平均使用存储空间仅需要1~2个字节，相较于常规存储降低90%存储使用空间。高压缩比存储同时能加快数据写入的速度。
数据安全：
为充分保证数据的可用性，时序数据存储默认采取三副本策略。

开通时序数据存储

时序数据存储随您购买的企业版实例开通。具体操作，请参见实例管理。

需要选择时序数据存储的规格，包括以下项目：


项目	说明
时序数据写入TPS	每秒钟写入时序数据存储的最大数据条数，达到上限后会写入失败。时序数据写入TPS选择5千条/秒时，不支持TSQL。
时序数据存储空间	时序数据达到存储空间上限后会写入失败。默认时序数据时效为永久，购买后时序数据时效可配置，请参见管理时序数据存储。

说明当实例被释放时，时序数据存储也将同时被释放。

管理时序数据存储

开通时序数据存储后，您可以设置时序数据时效，查看已使用的存储空间、时间线数，查看用户名、密码、连接地址，重新创建时序存储用户。

登录物联网平台控制台。
在实例概览页面，找到对应的实例，单击实例进入实例详情页面。
在实例详情页面下方，查看实时时序存储写入TPS、实时时序存储空间数据，及对应的监控图表，设置相关报警规则

单击相应区域框中的报警配置，在云监控控制台，设置阈值报警规则。具体操作，请参见创建阈值报警规则。
在实例详情页面，单击右上角的查看开发配置，查看下方的时序存储配置。
- 查看时序存储用户名和密码。
  单击查看，短信校验码验证通过后显示密码。
  
  用户名、密码、连接地址用于查询时序存储中的数据，购买企业版实例时，会自动生成。
- 重新创建时序存储用户。
  
  说明重新创建时序存储用户后，旧的用户名和密码将失效，会导致正在进行的时序数据查询失败，请及时更新SDK使用的用户名和密码。
  
  单击重新创建，输入新的用户名、密码，短信校验码验证通过后，创建成功。
- 设置时序数据存储时效。
  购买企业版实例时，时效配置默认关闭。您可打开存储时效开关，设置保留数据天数。取值范围为1~365天。

通过云产品流转写入数据

您可以通过配置数据流转规则，将数据转发到实例内的时序数据存储。

说明

在设置转发之前，请参见设置数据流转规则，创建数据转发规则和编写处理数据的SQL。
只支持JSON格式数据转发。
转发的消息中，除了配置的timestamp、tag值字段外，其他字段都将作为metric写入时序数据存储。metric的数据类型支持数值型、字符串，其他类型会导致写入失败。

登录物联网平台控制台。
在实例概览页面，找到对应的实例，单击实例进入实例详情页面。
在左侧导航栏，选择规则引擎 > 云产品流转。
单击规则对应的查看，进入数据流转规则页。
单击转发数据一栏对应的添加操作。

在添加操作对话框中，选择操作为存储到实例内的时序数据存储中。按照界面提示，设置其他信息，单击确认。


参数	描述
选择操作	选择存储到实例内的时序数据存储中。
地域	固定为您的物联网平台实例所在地域。
metric数据类型	选择metric的数据类型。支持数值型和字符串。更多信息，可单击帮助按钮查看。
timestamp	时间戳。支持：使用转义符`${}`表达式，例如`${time}`，表示取值为数据源Topic消息中time字段对应的值。使用数据流转函数`timestamp()`，表示取值为数据流转服务器的时间戳。输入值，必须为Unix时间戳，例如1404955893000。
tag名称	设置标记数据的标签名。支持中文汉字、英文字母、数字、和特殊字符，包括：半角冒号（:）、逗号（,）、英文句号（.）、单引号（'）、正斜线（/）、短划线（-）、下划线（_）、圆括号（()）、方括号（[]）。
tag值	设置标签值。支持：使用转义符`${}`表达式。例如，数据源Topic的消息结构中，包含一个位置属性，标识符为city，则可以指定标签值为`${city}`，表示消息中city字段对应的值。建议使用此方式。使用数据流转函数规定的一些函数，例如`deviceName()`，表示标签值为设备名称。支持的函数，请参见函数列表。输入常量，例如beijing。支持输入中文汉字、英文字母、数字、和特殊字符包括：半角冒号（:）、逗号（,）、英文句号（.）、单引号（'）、正斜线（/）、短划线（-）、下划线（_）、圆括号（()）、方括号（[]）。说明最多可添加8个tag键值对。需保证时序数据存储能够获取到配置的tag键值对，如果获取不到任意一个tag键值对，会导致写入数据库失败。

回到云产品流转页，单击规则对应的启动按钮启动规则。

下面给出一个数据流转示例。

示例规则的SQL：

SELECT time,city,power,distance, FROM "/alprodu****/myDevice/user/update"；

规则引擎根据SQL处理数据和写入数据到时序数据存储如下。

根据该SQL，规则引擎从Topic /alprodu****/myDevice/user/update的消息中，筛选出time、city、power和distance字段内容，作为转发的消息内容。
通过以上SQL处理后的转发消息内容示例如下：
```
{
"time": 1513677897,
"city": "beijing",
"distance": 8545,
"power": 93.0
}
```

根据已配置的数据流转操作，规则引擎向时序数据存储中写入两条数据。

示例中写入时序数据存储的数据如下：

数据: timestamp:1513677897, [metric:power value:93.0]
tag: cityName=beijing

数据: timestamp:1513677897, [metric:distance value:8545]
tag: cityName=beijing

通过SDK写入数据

SDK使用Point类表示一个时间点。一个 Point对象表示一个时间序列（时间线）某个时刻上的数据。

获取SDK。
使用Maven做项目构建工具，在pom.xml文件的<dependencies>标签中添加hitsdb-client依赖。代码如下：
```
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>hitsdb-client</artifactId>
    <version>0.2.7</version>
</dependency>
```
说明 SDK版本号必须为0.2.7及以上。

配置客户端。

客户端的所有配置均由TSDBConfig类进行配置。示例代码如下：

//实例详情页面中的时序存储SQL连接地址、端口。
String connectString = "XXX";
int port = XXX;
//实例详情页面中的时序数据存储用户名、密码。
String username = "XXX";
String password = "XXXX";
TSDBConfig config = TSDBConfig.address(connectString, port)
    .basicAuth(username, password)
    // 网络连接池大小，默认为64。
    .httpConnectionPool(64)
    // HTTP 等待时间，单位为秒，默认为90秒。
    .httpConnectTimeout(90)
    // IO 线程数，默认为1。
    .ioThreadCount(1)
    .config();

构造时间点。

Point（时间点）有多种构造方法，形式比较多样化。下面提供三种构造时间点的方法示例。

说明 Point的构建至少需要一个tag键值对。

示例一

构建一个时间点。用单位为秒的时间戳表示时间，指定Point数据的Metric与多个Tag。

// 以'秒'为时间戳。
int timestamp = (int)(System.currentTimeMillis()/1000);
// 构造Point。
Point point = Point.metric("test1")
                    .tag("tagk1", "tagv1")
                    .tag("tagk2", "tagv2")
                    .tag("tagk3", "tagv3")
                   .timestamp(timestamp).value(123.456)
                   .build();

示例二

构建一个时间点。用单位为毫秒的时间戳表示时间，指定Point数据的Metric，Tag使用Map形式的键值对。

// 也可以毫秒为时间戳。
long timestamp = System.currentTimeMillis();
// 使用HashMap表示Tags。
Map<String,String> tagsMap = new HashMap<String,String>();
tagsMap.put("tagk1", "tagv1");
tagsMap.put("tagk2", "tagv2");
// 构造Point。
Point point = Point.metric("test1")
                    .tag(tagsMap)
                   .value(timestamp,123.456)
                   .build();

示例三

构建一个时间点。使用java.util.Date表示时间。

// 使用java.util.Date表示时间。
Point point = Point.metric("test1")
                    .tag("tagk1", "tagv1")
                   .value(new Date,123.456)
                   .build();

写入数据。
SDK有两种写数据的方式：同步阻塞的写数据、异步非阻塞的写数据。
- 异步非阻塞的写数据
  异步写数据的方式比较简单，只要有Point就可以提交数据。Client会自动帮助您批量地提交数据，不需要您手动打包。如果没有特殊需求，推荐您使用异步非阻塞的写数据的方式。示例代码：
```
// 构建Point。
Point point = Point.metric("test1")
                   .tag("tagk1", "tagv1")
                   .value(timestamp, Math.random())
                   .build();
// 直接提交数据。
tsdb.put(point);
```
  当您使用异步写数据的方式时，若需要获取Put接口的响应结果，则需要设置回调接口。示例代码：
```
final AtomicLong num = new AtomicLong();
// 回调对象。
BatchPutCallback cb = new BatchPutCallback() {
    @Override
    public void response(String address, List<Point> input, Result output) {
        long afterNum = num.addAndGet(input.size());
        System.out.println("成功处理" + input.size() + ",已处理" + afterNum);
    }
    @Override
    public void failed(String address, List<Point> input, Exception ex) {
        ex.printStackTrace();
        long afterNum = num.addAndGet(input.size());
        System.out.println("失败处理" + input.size() + ",已处理" + afterNum);
    }
};
HiTSDBConfig config = HiTSDBConfig
                        .address("example.hitsdb.com", 8242)
                        .listenBatchPut(cb)    // 设置回调接口。
                        .config();
tsdb = HiTSDBClientFactory.connect(config);
```
  SDK提供了以下三种类型的PutCallback接口来根据需要选择返回的数据内容：
  - BatchPutCallback：即调用POST /api/put。
  - BatchPutSummaryCallback：即调用POST /api/put?summary=true。
  - BatchPutDetailsCallback：即调用POST /api/put?details=true。
  说明
  
  不要在回调方法中做耗时操作。若有此需要，可以再把操作交给其他工作线程执行。
  
  回调方法response()与failed()被调用的时机是不一样的：
  
  response()：指的是写入请求合法且被服务端处理后执行的回调。此时可以根据回调对象种类的不同，获取一些不同的反馈信息。
  
  failed()：指的是写入请求本身存在问题导致请求不合法（例如报文非法，触发限流，鉴权失败等）被服务端拒绝服务时所执行的回调。
  
  因此，建议在实现PutCallback接口时，不要忘记实现failed()方法。如果不实现的话，则会执行默认的failed()方法，其本身是一个空方法。
- 同步阻塞的写数据
  假设我们现在需要构建500个时间点提交给时序数据存储，示例代码：
```
List<Point> points = new ArrayList<Point>();
// 构建Point。
for(int i = 0; i<500; i++) {
    long timestamp = System.currentTimeMillis();
    Point point = Point.metric("test1")
                   .tag("tagk1", "tagv1")
                   .value(timestamp, Math.random())
                   .build();
    // 手动打包数据。
    points.add(point);
}
// 手动打包后提交数据。
tsdb.putSync(points)
```
  说明出于写入性能的考虑，同步写的方式一般需要您手动将数据点打包成一批数据，并且建议一批数据包含500～1000个数据点。此外，也建议多个线程并发进行提交。
  
  您可以根据需要选择返回的数据内容：
  - 返回空对象，无内容。本质是调用POST /api/put。
```
Result result = tsdb.putSync(ps);
```
  - 返回提交概述，包含成功数和返回数。本质是调用POST /api/put?summary=true。
```
SummaryResult summaryResult = tsdb.putSync(ps,SummaryResult.class);
```
  - 返回提交概述。包含成功数、返回数和失败原因。本质是调用POST /api/put?details=true。
```
DetailsResult detailsResult = tsdb.putSync(ps,DetailsResult.class);
```

通过SDK使用SQL查询数据

您可以通过SDK使用SQL查询已写入实例内时序数据存储的数据。

获取SDK。
使用Maven做项目构建工具，在pom.xml文件的<dependencies>标签中添加hitsdb-client依赖。代码如下：
```
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>hitsdb-client</artifactId>
    <version>0.2.7</version>
</dependency>
```
说明 SDK版本号必须为0.2.7及以上。

配置客户端。

客户端的所有配置均由TSDBConfig类进行配置。示例代码如下：

//实例详情页面中的时序存储SQL连接地址、端口。
String connectString = "XXX";
int port = XXX;
//实例详情页面中的时序数据存储用户名、密码。
String username = "XXX";
String password = "XXXX";
TSDBConfig config = TSDBConfig.address(connectString, port)
    .basicAuth(username, password)
    // 网络连接池大小，默认为64。
    .httpConnectionPool(64)
    // HTTP 等待时间，单位为秒，默认为90秒。
    .httpConnectTimeout(90)
    // IO 线程数，默认为1。
    .ioThreadCount(1)
    .config();

查询数据。

示例代码如下：

TSDB tsdbClient = TSDBClientFactory.connect(config);
//按标签筛选数据。
Map<String, String> tags = new HashMap<>();
String metric = "XXX";
long now = System.currentTimeMillis();
LastPointQuery query = LastPointQuery.builder()
    .timestamp(now)
    .backScan(-1)
    .msResolution(true)
    .sub(LastPointSubQuery.builder(metric, tags).build()).build();
List<LastDataValue> lastDataValues = tsdbClient.queryLast(query);

System.out.println(lastDataValues);

说明 SDK不能用于将数据写入实例内的时序数据存储。

通过JDBC连接使用TSQL查询数据

您可以通过JDBC连接使用TSQL进行数据的查询操作。

说明时序数据写入TPS为5千条/秒时，不支持TSQL。

引入Driver依赖。

运行时要求：

Java 1.8 Runtime。
配置实例JBDC访问，获取JDBC URL。

TSQL的JDBC driver的依赖已经发布到Maven仓库，这里以Maven来管理项目（tsql_jdbc_app）为例，把下面的内容加入pom.xml。获得一个包括所有依赖.jar文件，以及应用程序.class文件的JAR包。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.alibaba.tsdb.tsql</groupId>
    <artifactId>tsql_jdbc_app</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.drill.exec/drill-jdbc -->
        <dependency>
            <groupId>org.apache.drill.exec</groupId>
            <artifactId>drill-jdbc-all</artifactId>
            <version>1.15.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>log4j-over-slf4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.alibaba.tsdb.tsql.TsqlJdbcSampleApp</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id> <!-- this is used for inheritance merges -->
                        <phase>package</phase> <!-- bind to the packaging phase -->
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

在Java应用项目下，创建一个package com.alibaba.tsdb.tsql文件，并创建一个Java源文件TsqlJdbcSampleApp。

在以下示例中，根据您的实际情况，修改：

host：实例详情页面中的时序存储TSQL连接地址。
port：实例详情页面中的时序存储TSQL端口。
sql: 您所需要执行的TSQL查询语句。

package com.alibaba.tsdb.tsql;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
public class TsqlJdbcSampleApp {
    public static void main(String[] args) throws Exception {
        Connection connection = null;
        Statement stmt = null;
        try {
            // step 1: Register JDBC driver
            Class.forName("org.apache.drill.jdbc.Driver");
            // hostname or address of TSDB instance.
            String host = "ts-uf64t3199j58j8251.tsql.hitsdb.rds.aliyuncs.com";
            // port for TSQL JDBC service
            int port = 3306;
            String jdbcUrl = String.format("jdbc:drill:drillbit=%s:%s", host, port);
            // step 2: Open connection
            System.out.println("Connecting to database @ " + jdbcUrl + "  ...");
            connection = DriverManager.getConnection(jdbcUrl);
            // step 3: Create a statement
            System.out.println("Creating statement ...");
            stmt = connection.createStatement();
            // step 4: Execute a query using the statement.
            String sql = "select hostname, `timestamp`, `value` " +
                "from tsdb.`cpu.usage_system` " +
                "where `timestamp` between '2019-03-01' and '2019-03-01 00:05:00'";
            ResultSet rs = stmt.executeQuery(sql);
            // step 5: Extract data from ResultSet.
            int row = 0;
            System.out.println("hostname\ttimestamp\tvalue");
            System.out.println("-----------------------------------------------------");
            while (rs.next()) {
                row++;
                System.out.println(rs.getString("hostname") + "\t" + rs.getTimestamp("timestamp") + "\t" +rs.getDouble("value"));
            }
            System.out.println("-----------------------------------------------------");
            System.out.println( row + "rows returned");
        } catch(SQLException se){
            //Handle errors for JDBC
            se.printStackTrace();
        }catch(Exception e){
            //Handle errors for Class.forName
            e.printStackTrace();
        }finally{
            //finally block used to close resources
            try{
                if(stmt!=null)
                    stmt.close();
            }catch(SQLException se2){
            }// nothing we can do
            try{
                if(connection!=null)
                    connection.close();
            }catch(SQLException se){
                se.printStackTrace();
            }//end finally try
        }//end try
        System.out.println("Goodbye!");
    }
}

编译与执行。
在项目根目录下，执行下面的Maven命令：
```
maven clean install
```
执行完成后，您将在项目目录/targets下获得可执行程序tsql_jdbc_app-1.0-SNAPSHOT-jar-with-dependencies.jar。

执行该应用程序：
```
java -jar target/tsql_jdbc_app-1.0-SNAPSHOT-jar-with-dependencies.jar
```

JDBC协议使用限制

TSQL的JDBC协议存在功能限制。在使用TSQL JDBC协议前请对照检查，具体限制和说明如下：

TSQL目前仅支持时序数据查询和时序元数据查询，不支持数据写入，修改和删除。
时序数据存储没有事务（Transaction）的支持。

具体JDBC接口的限制说明如下：


接口	方法	TSDB JDBC支持情况
Connection	setAutoCommit(boolean)	仅允许true作为传入参数。
Connection	getAutoCommit()	返回true。
Connection	commit()	调用会引发异常：SQLFeatureNotSupportedException。
Connection	rollback()	调用会引发异常： SQLFeatureNotSupportedException。
Connection	setTransactionIsolation(int level)	仅允许TRANSACTION_NONE。
Connection	getTransactionIsolation()	仅允许TRANSACTION_NONE。
Connection	setSavePoint()	调用会引发异常 SQLFeatureNotSupportedException。
Connection	setSavePoint(String name)	调用会引发异常 SQLFeatureNotSupportedException。
Connection	rollback(Savepoint savepoint)	调用会引发异常 SQLFeatureNotSupportedException。
Connection	releaseSavePoint(Savepoint savepoint)	调用会引发异常 SQLFeatureNotSupportedException。
Connection	setNetworkTimeout()	调用会引发异常 SQLFeatureNotSupportedException。
Connection	getNetworkTimeout()	调用会引发异常 SQLFeatureNotSupportedException。