Java API

若将单条数据记录保存到 DolphinDB 内存表，可以使用类似 SQL 语句 insert into。

public void test_save_Insert(String str,int i, long ts,double dbl) throws IOException{
    conn.run(String.format("insert into sharedTable values('%s',%s,%s,%s)",str,i,ts,dbl));
}

tableInsert函数比较适合用来批量保存数据，它可将多个数组追加到 DolphinDB 内存表中。若 Java 程序获取的数据可以组织成 List 方式，可使用tableInsert函数保存。

public void test_save_TableInsert(List<String> strArray,List<Integer> intArray,List<Long> tsArray,List<Double> dblArray) throws IOException{
    //用数组构造参数
    List<Entity> args = Arrays.asList(new BasicStringVector(strArray),new BasicIntVector(intArray),new BasicTimestampVector(tsArray),new BasicDoubleVector(dblArray));
    conn.run("tableInsert{sharedTable}", args);
}

在本例中，使用了 DolphinDB 中的“部分应用”这一特性，将服务端表名以 tableInsert的方式固化到tableInsert中，作为一个独立函数来使用。具体文档请参考部分应用文档。

若 Java 程序获取的数据处理后组织成 BasicTable 对象，tableInsert函数也可以接受一个表对象作为参数，批量添加数据。

public void test_save_table(BasicTable table1) throws IOException {
    List<Entity> args = Arrays.asList(table1);
    conn.run("tableInsert{shareTable}", args);
}

dbPath = 'dfs://testDatabase'
tbName = 'tb1'

if(existsDatabase(dbPath)){dropDatabase(dbPath)}
db = database(dbPath,RANGE,2018.01.01..2018.12.31)
db.createPartitionedTable(t,tbName,'ctimestamp')

DolphinDB 提供loadTable方法可以加载分布式表，通过tableInsert方式追加数据，具体的脚本示例如下：

public void test_save_table(String dbPath, BasicTable table1) throws IOException{
    List<Entity> args = new ArrayList<Entity>(1);
    args.add(table1);
    conn.run(String.format("tableInsert{loadTable('%s','tb1')}",dbPath), args);
}

Java 程序中的数组或列表，也可以很方便的构造出 BasicTable 用于追加数据。例如若有 boolArray, intArray, dblArray, dateArray, strArray 这 5 个列表对象 (List<T>),可以通过以下语句构造 BasicTable 对象：

List<String> colNames =  Arrays.asList("cbool","cint","cdouble","cdate","cstring");
List<Vector> cols = Arrays.asList(new BasicBooleanVector(boolArray),new BasicIntVector(intArray),new BasicDoubleVector(dblArray),new BasicDateVector(dateArray),new BasicStringVector(strArray));
BasicTable table1 = new BasicTable(colNames,cols);

DolphinDB 的分布式表支持并发读写，下面展示如何在 Java 客户端中将数据并发写入 DolphinDB 的分布式表。

注意：DolphinDB 不允许多个 writer 同时将数据写入到同一个分区，因此在客户端多线程并行写入数据时，需要确保每个线程分别写入不同的分区。Java API 提供了自动按分区分流数据并行写入的简便方法，函数定义如下

public PartitionedTableAppender(String dbUrl, String tableName, String partitionColName, String appendFunction, DBConnectionPool pool)

• dbUrl: 必填，分布式数据库地址
• tableName: 必填，分布式表名
• partitionColName: 必填，分区字段
• appendFunction: 可选，自定义写入函数名，不填此参数则调用内置 tableInsert 函数。
• pool: 连接池，并行写入数据。

DBConnectionPool pool = new ExclusiveDBConnectionPool(HOST, PORT, "admin", "123456", 3, true, true);
PartitionedTableAppender appender = new PartitionedTableAppender(dbUrl, tableName , "sym", pool);

首先，在 DolphinDB 服务端执行以下脚本，创建分布式数据库"dfs://DolphinDBUUID"和分布式表"device_status"。其中，数据库按照 VALUE-HASH-HASH 的组合进行三级分区。

t = table(timestamp(1..10)  as date,string(1..10) as sym)
db1=database(\"\",HASH,[DATETIME,10])
db2=database(\"\",HASH,[STRING,5])
if(existsDatabase(\"dfs://demohash\")){
    dropDatabase(\"dfs://demohash\")
}
db =database(\"dfs://demohash\",COMPO,[db2,db1])
pt = db.createPartitionedTable(t,`pt,`sym`date)

注意：DolphinDB 不允许多个 writer 同时将数据写入到同一个分区，因此在客户端多线程并行写入数据时，需要确保每个线程分别写入不同的分区。

使用 1.30 版本以上的 server，可以使用 java api 中的 PartitionedTableAppender 类来写入分布式表，其基本原理是设计一个连接池用于多线程写入，然后利用 server 的 schema 函数获取分布式表的分区信息，按指定的分区列将用户写入的数据进行分类分别交给不同的连接来并行写入。 使用示例脚本如下：

DBConnectionPool conn = new ExclusiveDBConnectionPool(host, Integer.parseInt(port), "admin", "123456", Integer.parseInt(threadCount), false, false);

PartitionedTableAppender appender = new PartitionedTableAppender(dbPath, tableName, "gid", "saveGridData{'" + dbPath + "','" + tableName + "'}", conn);
BasicTable table1 = createTable();
appender.append(table1);            

• 在 Java API 中读取分布式表使用如下代码一次性读取数据

  String dbPath = "dfs://testDatabase";
  String tbName = "tb1";
  DBConnection conn = new DBConnection();
  conn.connect(SERVER, PORT, USER, PASSWORD);
  BasicTable table = (BasicTable)conn.run(String.format("select * from loadTable('%s','%s') where cdate = 2017.05.03",dbPath,tbName));

• 对于大数据量的表，API 提供了分段读取方法。(此方法仅适用于 DolphinDB 1.20.5 及其以上版本)

  Java API 提供了 EntityBlockReader 对象，在 run 方法中使用参数 fetchSize 指定分段大小，通过 read() 方法一段段的读取数据，示例如下：

  DBConnection conn = new DBConnection();
  conn.connect(SERVER, PORT, USER, PASSWORD);
  EntityBlockReader v = (EntityBlockReader)conn.run("table(1..22486 as id)",(ProgressListener) null,4,4,10000);
  BasicTable data = (BasicTable)v.read();
  while(v.hasNext()){
      BasicTable t = (BasicTable)v.read();
      data = data.combine(t);
  }

  在使用上述分段读取的方法时，若数据未读取完毕，需要放弃后续数据的读取时，必须调用 skipAll 方法来显示忽略后续数据，否则会导致套接字缓冲区滞留数据，引发后续数据的反序列化失败。 正确使用的示例代码如下：

      EntityBlockReader v = (EntityBlockReader)conn.run("table(1..12486 as id)",(ProgressListener) null,4,4,10000);
      BasicTable data = (BasicTable)v.read();
      v.skipAll();
      BasicTable t1 = (BasicTable)conn.run("table(1..100 as id1)"); //若没有 skipAll 此段会抛出异常。

在 Java API 中，数据表保存为 BasicTable 对象。由于 BasicTable 是列式存储，所以若要在 Java API 中读取行数据需要先取出需要的列，再取出行。

以下例子中参数 BasicTable 的有 4 个列，列名分别为 cstring, cint, ctimestamp, cdouble，数据类型分别是 STRING, INT, TIMESTAMP, DOUBLE。

public void test_loop_basicTable(BasicTable table1) throws Exception{
    BasicStringVector stringv = (BasicStringVector) table1.getColumn("cstring");
    BasicIntVector intv = (BasicIntVector) table1.getColumn("cint");
    BasicTimestampVector timestampv = (BasicTimestampVector) table1.getColumn("ctimestamp");
    BasicDoubleVector doublev = (BasicDoubleVector) table1.getColumn("cdouble");
    for(int ri=0; ri<table1.rows(); ri++){
        System.out.println(stringv.getString(ri));
        System.out.println(intv.getInt(ri));
        LocalDateTime timestamp = timestampv.getTimestamp(ri);
        System.out.println(timestamp);
        System.out.println(doublev.getDouble(ri));
    }
}

MultithreadedTableWriter 支持多线程的并发写入。

MultithreadedTableWriter 对象的主要方法介绍如下：

MultithreadedTableWriter(String hostName, int port, String userId, String password,
        String dbName, String tableName, boolean useSSL,
        boolean enableHighAvailability, String[] highAvailabilitySites,
        int batchSize, float throttle,
        int threadCount, String partitionCol,
        int[] compressTypes, Mode mode, String[] pModeOption, Callback callbackHandler)

参数说明：

• hostName 字符串，表示所连接的服务器的地址。
• port 整数，表示服务器端口。
• userId / password: 字符串，登录时的用户名和密码。
• dbPath 字符串，表示分布式数据库地址。内存表时该参数为空。请注意，1.30.17 及以下版本 API，向内存表写入数据时，该参数需填写内存表表名。
• tableName 字符串，表示分布式表或内存表的表名。请注意，1.30.17 及以下版本 API，向内存表写入数据时，该参数需为空。
• useSSL 布尔值，表示是否启用加密通讯。
• enableHighAvailability 布尔值，表示是否开启 API 高可用。
• highAvailabilitySites 数组类型，表示所有可用节点的 ip:port 构成的 String 数组。
• batchSize 整数，表示批处理的消息的数量。如果该参数值为 1，表示客户端写入数据后就立即发送给服务器； 如果该参数大于 1，表示数据量达到 batchSize 时，客户端才会将数据发送给服务器。
• throttle 大于 0 的浮点数，单位为秒。若客户端有数据写入，但数据量不足 batchSize，则等待 throttle 的时间再发送数据。
• threadCount 整数，表示创建的工作线程数量，如果值为 1，表示单线程。对于维度表，其值必须为 1。
• partitionCol 字符串类型，默认为空，仅在 threadCount 大于 1 时起效。对于分区表，必须指定为分区字段名； 如果是流数据表，必须指定为表的字段名；对于维度表，该参数不起效。
• compressTypes 数组类型，用于指定每一列采用的压缩传输方式，为空表示不压缩。每一列可选的压缩方式（大小写不敏感） 包括：
  • Vector.COMPRESS_LZ4：LZ4 压缩
  • Vector.COMPRESS_DELTA：DELTAOFDELTA 压缩
• mode 写入模式，用于指定 MultithreadedTableWriter 对象写入数据的方式，包括两种：
  • Mode.M_Append：表示以 tableInsert 的方式向追加数据。
  • Mode.M_Upsert：表示以 upsert! 方式更新（或追加）数据。
• pModeOption：字符串数组，表示不同模式下的扩展选项，目前，仅当 mode 指定为 Mode.M_Upsert 时有效，表示由 upsert! 可选参数组成的字符串数组。
• callbackHandler：回调类，默认为空，表示不使用回调。开启回调后，将继承回调接口 Callback 并重载回调方法，将回调的接口对象传入 MultithreadedTableWriter。

以下是 MultithreadedTableWriter 对象包含的函数方法介绍：

ErrorCodeInfo insert(Object... args)

函数说明：

插入单行数据。返回一个 ErrorCodeInfo 类，包含 errorCode 和 errorInfo，分别表示错误代码和错误信息。 当 errorCode 不为 "" 时，表示 MTW 写入失败，此时，errorInfo 会显示失败的详细信息。之后的版本 中会对错误信息进行详细说明，给出错误信息的代码、错误原因及解决办法。另外，ErrorCodeInfo 类提供了 hasError() 和 succeed() 方法用于获取数据插入的结果。hasError() 返回 True，则表示存在错误，否则表示无错误。succeed() 返回 True，则表示插入成功， 否则表示插入失败。如果构造 MultithreadedTableWriter 时开启了回调，则每次调用 insert 时，需要在每行数据前面增加一列 string 类型的数据作为每行的标识符（id），此 id 列仅用于回调时返回给用户，不会写入表中。

参数说明：

• args: 是变长参数，代表插入一行数据

示例：

ErrorCodeInfo pErrorInfo = multithreadedTableWriter_.insert(new Date(2022, 3, 23), "AAAAAAAB", random.nextInt() % 10000);

List<List<Entity>> getUnwrittenData()

函数说明：

返回一个嵌套列表，表示未写入服务器的数据。

注意：该方法获取到数据资源后，MultithreadedTableWriter 将释放这些数据资源。

示例：

List<List<Entity>> unwrittenData = multithreadedTableWriter_.getUnwrittenData();

ErrorCodeInfo insertUnwrittenData(List<List<Entity>> records)

函数说明：

将数据插入数据表。返回值同 insert 方法。与 insert 方法的区别在于，insert 只能插入单行数据，而 insertUnwrittenData 可以同时插入多行数据。

参数说明：

• records：需要再次写入的数据。可以通过方法 getUnwrittenData 获取该对象。

示例：

ErrorCodeInfo ret = multithreadedTableWriter_.insertUnwrittenData(unwrittenData);

Status getStatus()

函数说明：

获取 MultithreadedTableWriter 对象当前的运行状态。

参数说明：

• status：是 MultithreadedTableWriter.Status 类，具有以下属性和方法

示例：

MultithreadedTableWriter.Status writeStatus = new MultithreadedTableWriter.Status();
writeStatus = multithreadedTableWriter_.getStatus();

属性：

• isExiting：写入线程是否正在退出。
• errorCode：错误码。
• errorInfo：错误信息。
• sentRows：成功发送的总记录数。
• unsentRows：待发送的总记录数。
• sendFailedRows：发送失败的总记录数。
• threadStatus：写入线程状态列表。
  • threadId：线程 Id。
  • sentRows：该线程成功发送的记录数。
  • unsentRows：该线程待发送的记录数。
  • sendFailedRows：该线程发送失败的记录数。

方法：

• hasError()：true 表示数据写入存在错误；false 表示数据写入无错误。
• succeed()：true 表示数据写入成功；false 表示数据写入失败。

waitForThreadCompletion()

函数说明：

调用此方法后，MTW 会进入等待状态，待后台工作线程全部完成后退出等待状态。

示例：

multithreadedTableWriter_.waitForThreadCompletion();

MultithreadedTableWriter 的正常使用示例如下：

DBConnection conn= new DBConnection();
conn.connect(HOST, PORT, "admin", "123456");
Random random = new Random();
String script =
        "dbName = 'dfs://valuedb3'" +
                "if (exists(dbName))" +
                "{" +
                "dropDatabase(dbName);" +
                "}" +
                "datetest = table(1000:0,`date`symbol`id,[DATE, SYMBOL, LONG]);" +
                "db = database(directory= dbName, partitionType= HASH, partitionScheme=[INT, 10]);" +
                "pt = db.createPartitionedTable(datetest,'pdatetest','id');";
conn.run(script);
MultithreadedTableWriter multithreadedTableWriter_ = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo ret;
try
{
    //插入 100 行正确数据
    for (int i = 0; i < 100; ++i)
    {
        ret = multithreadedTableWriter_.insert(new Date(2022, 3, 23), "AAAAAAAB", random.nextInt() % 10000);
    }
} 
catch (Exception e)
{   //MTW 抛出异常
    System.out.println("MTW exit with exception {0}" + e.getMessage());
}

//等待 MTW 插入完成
multithreadedTableWriter_.waitForThreadCompletion();
MultithreadedTableWriter.Status writeStatus = new MultithreadedTableWriter.Status();
writeStatus = multithreadedTableWriter_.getStatus();
if (!writeStatus.errorInfo.equals(""))
{
    //如果写入时发生错误
    System.out.println("error in writing !");
}
System.out.println("writeStatus: {0}\n" + writeStatus.toString());
System.out.println(((BasicLong)conn.run("exec count(*) from pt")).getLong());

以上代码输出结果为：

      writeStatus: {0}
      errorCode     : 
      errorInfo     : 
      isExiting     : true
      sentRows      : 100
      unsentRows    : 0
      sendFailedRows: 0
      threadStatus  :
              threadId        sentRows      unsentRows  sendFailedRows
                    13              30               0               0
                    14              18               0               0
                    15              15               0               0
                    16              20               0               0
                    17              17               0               0
    
      100

调用 writer.insert() 方法向 writer 中写入数据，并通过 writer.getStatus() 获取 writer 的状态。 注意，使用 writer.waitForThreadCompletion() 方法等待 MTW 写入完毕，会终止 MTW 所有工作线程，保留最后一次写入信息。此时如果需要再次将数据写入 MTW，需要重新获取新的 MTW 对象，才能继续写入数据。

由上例可以看出，MTW 内部使用多线程完成数据转换和写入任务。但在 MTW 外部，API 客户端同样支持以多线程方式将数据写入 MTW，且保证了多线程安全。

MultithreadedTableWriter 在开启回调后，用户会在回调的方法中获取到一个 BasicTable 类型的回调表，该表由两列构成： 第一列（String 类型），存放的是调用 MultithreadedTableWriter.insert 时增加的每一行的 id；第二列（布尔值），表示每一行写入成功与否，true 表示写入成功，false 表示写入失败。

-继承 Callback 接口并重载 writeCompletion 方法用于获取回调数据

示例：

Callback callbackHandler = new Callback(){
    public void writeCompletion(Table callbackTable){
        List<String> failedIdList = new ArrayList<>();
        BasicStringVector idVec = (BasicStringVector) callbackTable.getColumn(0);
        BasicBooleanVector successVec = (BasicBooleanVector) callbackTable.getColumn(1);
        for (int i = 0; i < successVec.rows(); i++){
            if (!successVec.getBoolean(i)){
                failedIdList.add(idVec.getString(i));
            }
        }
    }
};

-构造 MultithreadedTableWriter 对象并传入回调对象

示例：

MultithreadedTableWriter mtw = new MultithreadedTableWriter(host, port, userName, password, dbName, tbName, useSSL,
        enableHighAvailability, null, 10000, 1, 1, "price", callbackHandler);

-调用 MultithreadedTableWriter 的 insert 方法并在第一列中为每一行写入 id

String theme = "theme1";
for (int id = 0; id < 1000000; id++){
    mtw.insert(theme + id, code, price); //theme+id 为每一行对应的 id，将在回调时返回
}

MultithreadedTableWriter 类调用 insert 方法插入数据时发生异常：

在调用 MultithreadedTableWriter 的 insert 方法时，若插入数据的类型与表对应列的类型不匹配，则 MultithreadedTableWriter 会立刻返回错误信息并打印出堆栈。

示例：

DBConnection conn= new DBConnection();
conn.connect(HOST, PORT, "admin", "123456");
Random random = new Random();
String script =
        "dbName = 'dfs://valuedb3'" +
                "if (exists(dbName))" +
                "{" +
                "dropDatabase(dbName);" +
                "}" +
                "datetest = table(1000:0,`date`symbol`id,[DATE, SYMBOL, LONG]);" +
                "db = database(directory= dbName, partitionType= HASH, partitionScheme=[INT, 10]);" +
                "pt = db.createPartitionedTable(datetest,'pdatetest','id');";
conn.run(script);
MultithreadedTableWriter multithreadedTableWriter_ = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo ret;
//插入 1 行类型错误数据，MTW 立刻返回错误信息
ret = multithreadedTableWriter_.insert(new Date(2022, 3, 23), 222, random.nextInt() % 10000);
if (!ret.errorInfo.equals(""))
    System.out.println("insert wrong format data: {2}\n" + ret.toString());

以上代码输出结果为：

      java.lang.RuntimeException: Failed to insert data. Cannot convert int to DT_SYMBOL.
       at com.xxdb.data.BasicEntityFactory.createScalar(BasicEntityFactory.java:795)
       at com.xxdb.data.BasicEntityFactory.createScalar(BasicEntityFactory.java:505)
       at com.xxdb.multithreadedtablewriter.MultithreadedTableWriter.insert(MultithreadedTableWriter.java:594)
       at com.xxdb.BehaviorTest.testMul(BehaviorTest.java:89)
       at com.xxdb.BehaviorTest.main(BehaviorTest.java:168)
        code=A1 info=Invalid object error java.lang.RuntimeException: Failed to insert data. Cannot convert int to DT_SYMBOL.

在调用 MultithreadedTableWriter 的 insert 方法时，若 insert 插入数据的列数和表的列数不匹配，MultithreadedTableWriter 会立刻返回错误信息。

示例：

DBConnection conn= new DBConnection();
conn.connect(HOST, PORT, "admin", "123456");
Random random = new Random();
String script =
        "dbName = 'dfs://valuedb3'" +
                "if (exists(dbName))" +
                "{" +
                "dropDatabase(dbName);" +
                "}" +
                "datetest = table(1000:0,`date`symbol`id,[DATE, SYMBOL, LONG]);" +
                "db = database(directory= dbName, partitionType= HASH, partitionScheme=[INT, 10]);" +
                "pt = db.createPartitionedTable(datetest,'pdatetest','id');";
conn.run(script);
MultithreadedTableWriter multithreadedTableWriter_ = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo ret;
//插入 1 行数据，插入数据的列数和表的列数不匹配，MTW 立刻返回错误信息
ret = multithreadedTableWriter_.insert(new Date(2022, 3, 23), random.nextInt() % 10000);
if (!ret.errorInfo.equals(""))
    System.out.println("insert wrong format data: {3}\n" + ret.toString());

以上代码输出结果为：

    insert wrong format data: {3}
      code=A2 info=Column counts don't match.  

如果 MultithreadedTableWriter 在运行时连接断开，则所有工作线程被终止。继续通过 MultithreadedTableWriter 向服务器写数据时，会因为工作线程终止而抛出异常，且数据不会被写入。此时， 可通过调用 MultithreadedTableWriter 的 getUnwrittenData 获取未插入的数据，并重新插入。

示例：

List<List<Entity>> unwriterdata = new ArrayList<>();
unwriterdata = multithreadedTableWriter_.getUnwrittenData();
System.out.println("{5} unwriterdata: " + unwriterdata.size());
//重新获取新的 MTW 对象
MultithreadedTableWriter newmultithreadedTableWriter = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
try
{
    boolean writesuccess = true;
    //将没有写入的数据写到新的 MTW 中
    ret = newmultithreadedTableWriter.insertUnwrittenData(unwriterdata);
}
finally
{
    newmultithreadedTableWriter.waitForThreadCompletion();
    writeStatus = newmultithreadedTableWriter.getStatus();
    System.out.println("writeStatus: {6}\n" + writeStatus.toString());
}

以上代码输出结果为：

  {5} unwriterdata: 10
  writeStatus: {6}
  errorCode     : 
  errorInfo     : 
  isExiting     : true
  sentRows      : 10
  unsentRows    : 0
  sendFailedRows: 0
  threadStatus  :
          threadId        sentRows      unsentRows  sendFailedRows
                23               3               0               0
                24               2               0               0
                25               1               0               0
                26               3               0               0
                27               1               0               0