dolphinDB

2021-04-20

DolphinDB

国产高性能分布式时序数据库

~~写在开头：本人初涉dolphinDB 此文仅用于私人翻阅如有错误指出还请指出~~

写这篇文章最主要的原因是：官方没有一套快速开始的文档（粘来即用）~~可能是本人太菜看不懂~~ 所以我整理出来一套东西存档

快速开始（如何使用API创建数据库，表）

/**
 * 创建分布式库 dfs://dbName 以及 分布式表 tableName
 * 其中 分布式库 value - hash 分区
 * dolphin 允许并行写入 但是 不允许同时写入同一个分区
 * 多线程并发写入分布式表时 根据哈希分区字段数据的哈希值分组 每组指定一个写线程 这样就能保证每个线程同时将数据写到不同的哈希分区
 * @param conn
 * @param dbName
 * @param tableName
 * @throws IOException
 */
public static void createDBAndTableIfNotExist(DBConnection conn, String dbName, String tableName) throws IOException {
    String dbScript = "if(!existsDatabase('dfs://%s'))\n" +
            "{tableName=\"%s\"\n" +
            "db1 = database('',VALUE,2018.01.01..2030.12.31)\n" +
            "db2 = database('',HASH,[SYMBOL,128 ])\n" +
            "t = table(10000:0,`time`id`v`q,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE])\n" +
            "db = database('dfs://%s',COMPO,[db1,db2])\n" +
            "db.createPartitionedTable(t,\"%s\",`time`id)}";
    // 数据名称 表名 数据库名 表名
    String createDbScript = String.format(dbScript,dbName, tableName, dbName,tableName);
    conn.run(createDbScript);
}

批量插入

官方提供分布式批量插入接口

public static void insert(DBConnection conn, String dbName, String tableName, BasicTable table) throws Exception {
    List<Entity> args = Arrays.asList(table);  
    conn.run(String.format("tableInsert{loadTable('dfs://%s','%s')}",dbName,tableName),args);
}

说明：

时序数据库中表的字段本人预先就定义好了
时序数据库中数据的插入可以看做是通过集合一次性添加

   // 时序数据库的列名集合
   public static final List<String> colNames= Arrays.asList(TIME,ID, VALUE,QUALITY);

// 每个字段所对应的值的集合
List<String> point_id = new ArrayList();
   List<Double> point_q = new ArrayList<>();
   List<Long> point_time = new ArrayList<>();
   List<Double> point_v = new ArrayList<>();

   // 构建插入数据
   List<com.xxdb.data.Vector> point_cols = Arrays.asList(new BasicTimestampVector(point_time), new BasicStringVector(point_id), new BasicDoubleVector(point_v), new BasicDoubleVector(point_q.subList(0, point_time.size())));
   BasicTable point_table = new BasicTable(colNames, point_cols);

   try {
       insert(conn, database, tableName, point_table);
   } catch (Exception e) {
       e.printStackTrace();
   }