全部產品
Search
文件中心

Lindorm:介面說明

更新時間:Jul 06, 2024

本文介紹Lindorm TSDB SDK常用介面的使用方法。

Lindorm TSDB SDK的介面主要分成管理介面、寫入介面以及查詢介面等3類。

管理介面

Lindorm時序引擎支援使用SQL進行DDL、DCL操作,這裡管理介面相關的樣本是直接通過提交SQL實現對應的DDL和DCL操作。在Lindorm TSDB SDK中,支援以下兩個重載方法向Lindorm時序引擎提交SQL實現DDL和DCL操作。

// 直接提交SQL
Result execute(String sql);

// 在指定的資料庫database下進行DDL等操作,比如在指定的database下建立表
Result execute(String database, String sql);

其中,Result對象表示SQL的執行結果,比如提交資料庫的列表語句("SHOW DATABASES")後,可以通過Result擷取到已經存在的資料庫列表。Result對象有columns、metadata、rows等3個欄位,其中,columns為結果的列名集合,metadata為對應列的資料類型集合,rows為結果的按行返回的集合。

public class Result {
    private List<String> columns;
    private List<String> metadata;
    private List<List<Object>> rows;
    
    ....
 }

資料庫和表管理樣本

資料庫和表的DDL操作主要包括建立、刪除、查看等。程式碼範例如下:

// 1.查看資料庫列表
String showDatabase = "show databases";
Result result = lindormTSDBClient.execute(showDatabase);
System.out.println("before create, db list: " +  result.getRows().stream().map(e -> (String) e.get(0)).collect(Collectors.toList()));

// 2.建立資料庫,此處樣本建立了名為demo的資料庫
String createDatabase = "create database demo";
result = lindormTSDBClient.execute(createDatabase);
System.out.println("create database:" + result.isSuccessful());

// 3.查看資料庫列表
result = lindormTSDBClient.execute(showDatabase);
System.out.println("after create, db list: " +  result.getRows().stream().map(e -> (String) e.get(0)).collect(Collectors.toList()));

String database = "demo";

// 4.查看所有表
String showTables = "show tables";
result = lindormTSDBClient.execute(database, showTables);
System.out.println("before create, table list: " +  result.getRows().stream().map(e -> (String) e.get(0)).collect(Collectors.toList()));

// 5.建立表
String createTable = "CREATE TABLE sensor (device_id VARCHAR TAG,region VARCHAR TAG,time BIGINT,temperature DOUBLE,humidity DOUBLE,PRIMARY KEY(device_id))";
result = lindormTSDBClient.execute(database, createTable);
System.out.println("create table: " + result.isSuccessful());

// 6.查看所有表,驗證是否成功建立了表sensor
result = lindormTSDBClient.execute(database, showTables);
System.out.println("after create, table list: " +  result.getRows().stream().map(e -> (String) e.get(0)).collect(Collectors.toList()));

// 7.描述表詳情
String describeTable = "describe table sensor";
result = lindormTSDBClient.execute(database, describeTable);
System.out.println("------------ describe table -------------------");
List<String> columns = result.getColumns();
System.out.println("columns: " + columns);
List<String> metadata = result.getMetadata();
System.out.println("metadata: " + metadata);
List<List<Object>> rows = result.getRows();
for (int i = 0, size = rows.size(); i < size; i++) {
    List<Object> row = rows.get(i);
    System.out.println("column #" + i + " : " + row);
}

System.out.println("------------ describe table -------------------");
// 8.刪除表
String dropTable = "drop table sensor";
result = lindormTSDBClient.execute(database, dropTable);
System.out.println("drop table: " + result.isSuccessful());

// 9.查看錶列表,驗證是否成功刪除
result = lindormTSDBClient.execute(database, showTables);
System.out.println("after drop, table list: " +  result.getRows().stream().map(e -> (String) e.get(0)).collect(Collectors.toList()));

// 10.刪除指定資料庫
String dropDatabase = "drop database demo";
result = lindormTSDBClient.execute(dropDatabase);
System.out.println("drop database:" + result.isSuccessful());

// 11.查看資料庫列表,驗證指定資料庫是否刪除成功
result = lindormTSDBClient.execute(showDatabase);
System.out.println("after drop, db list : " +  result.getRows().stream().map(e -> (String) e.get(0)).collect(Collectors.toList()));

運行結果樣本:

before create, db list: [default]
create database:true
after create, db list: [default, demo]
before create, table list: []
create table: true
after create, table list: [sensor]
------------ describe table -------------------
columns: [columnName, typeName, columnKind]
metadata: [VARCHAR, VARCHAR, VARCHAR]
column #0 : [device_id, VARCHAR, TAG]
column #1 : [region, VARCHAR, TAG]
column #2 : [time, TIMESTAMP, TIMESTAMP]
column #3 : [temperature, DOUBLE, FIELD]
column #4 : [humidity, DOUBLE, FIELD]
------------ describe table -------------------
drop table: true
after drop, table list: []
drop database:true
after drop, db list : [default]
說明

對於這裡沒有列舉的其他管理相關的操作,比如連續查詢的建立、刪除等DDL操作,都可以根據Lindorm時序引擎的對應的SQL文法說明構造對應的SQL語句來實現。關於Lindorm時序引擎的SQL語句說明,可以參考SQL文法參考

寫入介面

Lindom TSDB SDK的寫入介面為多個相同名稱但參數不同的write方法,但根據處理寫入結果方式,可以分成兩類,分別為使用CompletableFuture<WriteResult>處理寫入結果的介面和使用回呼函數Callback處理非同步寫入結果的介面。

預設情況下,LindormTSDBClient為提高寫入效能,使用非同步攢批的方式進行資料寫入。若需要同步寫入,只需要調用下write方法返回的CompletableFuture<WriteResult>的join方法即可。

寫入記錄

Lindorm TSDB SDK使用Record對象表示表中的一行寫入記錄。Record需要指定表名、時間戳記、標籤和量測值。其中,時間戳記和標籤將用於構建索引。

預設構建Record會進行字元合法性校正,可以通過在build中添加參數'false'關閉校正。

Record record = Record
    // 指定表名
    .table("sensor")
    // 指定時間戳記,單位為毫秒
    .time(currentTime)
    // 指定標籤
    .tag("device_id", "F07A1260")
    .tag("region", "north-cn")
    // 指定量測值
    .addField("temperature", 12.1)
    .addField("humidity", 45.0)
    .build();

使用CompletableFuture處理寫入結果的介面

  • 每次只提交一行記錄。

// 寫入預設資料庫 
CompletableFuture<WriteResult> future = lindormTSDBClient.write(record);

// 寫入指定的資料庫
String database = "demo";
CompletableFuture<WriteResult> future = lindormTSDBClient.write(database, record);
  • 每次提交多行記錄(推薦)。這種方式提交方式,可減少提交時的SDK非同步隊列鎖競爭,效率較高。

List<Record> records;
// 寫入預設資料庫
CompletableFuture<WriteResult> future = lindormTSDBClient.write(records);

// 寫入指定的資料庫
String database = "demo";
CompletableFuture<WriteResult> future = lindormTSDBClient.write(database, records);
  • 處理CompletableFuture<WriteResult>完成時的結果。這裡僅是一個樣本,業務上可以結合實際情況,調用CompletableFuture的其他方法進行結果處理。

CompletableFuture<WriteResult> future = lindormTSDBClient.write(records);
// 處理非同步寫入結果
future.whenComplete((r, ex) -> {
    if (ex != null) { // 發送異常,一般都是寫入失敗
        System.out.println("Failed to write.");
        Throwable throwable = ExceptionUtils.getRootCause(ex);
        if (throwable instanceof LindormTSDBException) {
            LindormTSDBException e = (LindormTSDBException) throwable;
            System.out.println("Caught an LindormTSDBException, which means your request made it to Lindorm TSDB, "
                               + "but was rejected with an error response for some reason.");
            System.out.println("Error Code: " + e.getCode());
            System.out.println("SQL State:  " + e.getSqlstate());
            System.out.println("Error Message: " + e.getMessage());
        }  else {
            throwable.printStackTrace();
        }
    } else  { // 一般都是寫入成功
        if (r.isSuccessful()) {
            System.out.println("Write successfully.");
        } else {
            System.out.println("Write failure.");
        }
    }
});
重要

不要在CompletableFuture的whenComplete等方法中做複雜耗時的計算,否則可能會阻塞寫入;若確實需要做複雜的耗時計算,請將計算邏輯提交到其他獨立現場池中。關於錯誤碼的取值和含義,請參見常見錯誤碼參考

使用回呼函數Callback處理非同步寫入結果的介面

  • 寫入回調介面Callback的定義如下所示。onCompletion方法上的result表示寫入結果,records表示這個回呼函數關聯的寫入記錄,異常e表示寫入失敗時的異常。

public interface Callback {
    void onCompletion(WriteResult result, List<Record> records, Throwable e);
}

下面是使用Callback介面處理寫入結果的一個樣本。

Callback callback =  new Callback() {
    @Override
    public void onCompletion(WriteResult result, List<Record> list,
                             Throwable throwable) {
        if (throwable != null) { // 寫入失敗
            if (throwable instanceof LindormTSDBException) {
                LindormTSDBException ex = (LindormTSDBException) throwable;
                System.out.println("errorCode: " + ex.getCode());
                System.out.println("sqlstate: " + ex.getSqlstate());
                System.out.println("message: " + ex.getMessage());
            } else {
                // 其他錯誤
                throwable.printStackTrace();
            }
        } else {
            if (result.isSuccessful()) {
                System.out.println("Write successfully.");
            } else {
                System.out.println("Write failure.");
            }
        }
    }
};
重要

不要在Callback的onCompletion中做複雜耗時的計算,否則可能會阻塞寫入;若確實需要做複雜的耗時計算,請將計算邏輯提交到其他獨立現場池中。關於錯誤碼的取值和含義,請參見常見錯誤碼參考

  • 每次提交一行記錄。

// 寫入預設資料庫 
lindormTSDBClient.write(record, callback);

// 寫入指定的資料庫
String database = "demo";
lindormTSDBClient.write(database, record, callback);
  • 每次提交多行記錄(推薦)。這種方式提交方式,可減少提交時的SDK非同步隊列鎖競爭,效率較高。

List<Record> records;
// 寫入預設資料庫
lindormTSDBClient.write(records, callback);

// 寫入指定的資料庫
String database = "demo";
lindormTSDBClient.write(database, records, callback);

查詢介面

Lindorm TSDB SDK的查詢介面是通過SQL查詢資料的,關於SQL語句說明可以參考SQL文法參考

Lindorm TSDB SDK的查詢介面如下所示,其中,入參有要查詢的資料庫database, 查詢語句sql,查詢結果每批返回的行數大小chunkSize。

ResultSet query(String database, String sql, int chunkSize);

Lindorm TSDB SDK 使用ResultSet介面表示SQL查詢結果,其定義如下所示。

public interface ResultSet extends Closeable {

    QueryResult next();

    void close();
}

在處理查詢結果時,可以迴圈調用ResultSet的next方法擷取查詢結果,當next方法返回的QueryResult對象為null時,表示已經讀取完全部查詢結果。 當查詢結束時,調用ResultSet的close方法釋放對應的IO資源。

QueryResult表示每批返回的查詢結果,該對象有columns、metadata、rows等3個欄位,其中,columns為查詢結果的查詢的列名集合,metadata為對應列的資料類型集合,rows為查詢結果的按行返回的集合。其中,關於Lindorm時序引擎查詢支援的資料類型集合,請參考資料類型。下面是處理查詢結果ResultSet的樣本。

String sql = "select * from sensor";
int chunkSize = 100;
ResultSet resultSet = lindormTSDBClient.query("demo", sql, chunkSize);

// 處理查詢結果
try {
    QueryResult result = null;
    // 當resultSet的next()方法返回為null,表示已經讀取完所有的查詢結果
    while ((result = resultSet.next()) != null) {
        List<String> columns = result.getColumns();
        System.out.println("columns: " + columns);
        List<String> metadata = result.getMetadata();
        System.out.println("metadata: " + metadata);
        List<List<Object>> rows = result.getRows();
        for (int i = 0, size = rows.size(); i < size; i++) {
            List<Object> row = rows.get(i);
            System.out.println("row #" + i + " : " + row);
        }
    }
} finally {
    // 查詢結束後,需確保調用ResultSet的close方法,以釋放IO資源
    resultSet.close();
}
重要

不管查詢是否成功,在查詢結束時,都需顯式調用ResultSet的close方法,以釋放IO資源,否則會造成網路連接泄露。

另外,Lindorm TSDB SDK也提供了幾個查詢介面重載實現,如下所示,業務上可以根據實際情況選擇使用不同實現。

// SQL查詢語句
String sql = "xxxx";

// 1.使用SQL語句,查詢預設資料庫, 預設每批返回1000行資料
ResultSet resultSet = lindormTSDBClient.query(sql);

String database = "demo";
// 2.使用SQL語句,查詢指定資料庫,預設每批返回1000行資料
ResultSet resultSet = lindormTSDBClient.query(database, sql);